feat: support client-supplied uuid for multimodal processor cache.

xllm/api_service/mm_service_utils.h

@@ -78,6 +78,10 @@ bool build_messages(const google::protobuf::RepeatedPtrField<
                                 "message content type is invalid.");
         return false;
       }
+
+      if (item.type() != "text" && item.has_uuid()) {
+        contents.back().uuid = item.uuid();
+      }
     }
 
     out_messages.emplace_back(req_message.role(), std::move(contents));

xllm/core/common/message.h

@@ -64,6 +64,8 @@ struct MMContent {
   AudioURL audio_url;
 
   Embedding embedding;
+
+  std::optional<std::string> uuid;
 };
 using MMContentVec = std::vector<MMContent>;
 

xllm/core/distributed_runtime/vlm_master.cpp

@@ -469,20 +469,8 @@ std::shared_ptr<Request> VLMMaster::generate_request(
   xllm::ScopeGuard rate_limit_guard(
       [this] { get_rate_limiter()->decrease_one_request(); });
 
-  static MMInputTransfer mm_input_transfer;
-
-  MMInput mm_inputs(std::move(payload));
-  MMErrCode code = mm_input_transfer.trans(messages, mm_inputs);
-  if (code != MMErrCode::SUCCESS) {
-    std::string error_message = MMErrToString(code);
-    LOG(ERROR) << error_message;
-    CALLBACK_WITH_ERROR(StatusCode::INVALID_ARGUMENT, error_message);
-    return nullptr;
-  }
-
   MMData mm_data;
-  if (!mm_inputs.empty() &&
-      !processor_->process_multimodal(mm_inputs, mm_data)) {
+  if (!processor_->process_mm_input(messages, std::move(payload), mm_data)) {
     CALLBACK_WITH_ERROR(StatusCode::INVALID_ARGUMENT,
                         "Failed to process multimodal input.");
     return nullptr;

xllm/core/framework/multimodal/mm_data.cpp

@@ -20,10 +20,13 @@ limitations under the License.
 namespace xllm {
 
 MMData::MMData(uint32_t type, const MMItemVec& items)
+    : type_(type), items_(items) {}
+
+MMData::MMData(uint32_t type, MMItemVec&& items)
     : type_(type), items_(std::move(items)) {}
 
 MMData::MMData(uint32_t type, const MMDict& items)
-    : type_(type), items_(std::move(items)) {}
+    : type_(type), items_(items) {}
 
 bool MMData::has(const MMKey& key) const {
   if (!valid()) return false;

xllm/core/framework/multimodal/mm_data.h

@@ -61,6 +61,7 @@ class MMData {
  public:
   MMData() = default;
   MMData(uint32_t type, const MMItemVec& items);
+  MMData(uint32_t type, MMItemVec&& items);
   MMData(uint32_t type, const MMDict& items);
 
   bool has(uint32_t type) const { return type & type_ != 0; }

xllm/core/framework/multimodal/mm_handler.cpp

@@ -40,7 +40,7 @@ MMErrCode MMHandlerBase::process(const MMContent& content,
   code = this->decode(input);
   if (code != MMErrCode::SUCCESS) return code;
 
-  if (!input.raw_data.empty()) {
+  if (!input.raw_data.empty() && !input.hash_key.has_value()) {
     input.hash_key = hash_string(input.raw_data);
   }
 
@@ -117,8 +117,6 @@ MMErrCode MMHandlerBase::load_from_http(
 MMErrCode ImageHandler::load(const MMContent& content,
                              MMInputItem& input,
                              MMPayload& payload) {
-  input.clear();
-
   const auto& image_url = content.image_url;
   const auto& url = image_url.url;
 
@@ -154,8 +152,6 @@ MMErrCode ImageHandler::decode(MMInputItem& input) {
 MMErrCode VideoHandler::load(const MMContent& content,
                              MMInputItem& input,
                              MMPayload& payload) {
-  input.clear();
-
   const auto& video_url = content.video_url;
   const auto& url = video_url.url;
 
@@ -202,8 +198,6 @@ MMErrCode VideoHandler::decode(MMInputItem& input) {
 MMErrCode AudioHandler::load(const MMContent& content,
                              MMInputItem& input,
                              MMPayload& payload) {
-  input.clear();
-
   const auto& audio_url = content.audio_url;
   const auto& url = audio_url.url;
 

xllm/core/framework/multimodal/mm_input.cpp

@@ -38,6 +38,21 @@ bool is_url_type(const std::string& type) {
   return type == "image_url" || type == "video_url" || type == "audio_url";
 }
 
+// Base modality for a url content type, set on the shell so the uuid pre-filter
+// can classify items before download.
+MMType url_type_to_modality(const std::string& type) {
+  if (type == "image_url") {
+    return MMType::IMAGE;
+  }
+  if (type == "video_url") {
+    return MMType::VIDEO;
+  }
+  if (type == "audio_url") {
+    return MMType::AUDIO;
+  }
+  return MMType::NONE;
+}
+
 bool is_binary_data_url(std::string_view url) {
   constexpr std::string_view kPrefix = "data:";
   constexpr std::string_view kMarker = ";binary,";
@@ -94,6 +109,13 @@ MMPayload slice_payload(const MMContent& item, MMPayload& payload) {
   return MMPayload(std::move(buf));
 }
 
+void apply_uuid_key(const MMContent& content, MMInputItem& item) {
+  if (content.uuid.has_value()) {
+    item.uuid = content.uuid.value();
+    item.hash_key = hash_string(content.uuid.value());
+  }
+}
+
 }  // namespace
 
 bool MMInput::foreach (MMInputItem::IVisitor& v) const {
@@ -133,6 +155,106 @@ MMErrCode MMInputTransfer::trans(const std::vector<Message>& messages,
   return MMErrCode::SUCCESS;
 }
 
+MMErrCode MMInputTransfer::collect(const std::vector<Message>& messages,
+                                   MMInput& inputs,
+                                   std::vector<MMSourceRef>& refs) {
+  inputs.clear();
+  refs.clear();
+
+  for (const Message& message : messages) {
+    const MMContentVec& mmc = std::get<MMContentVec>(message.content);
+    for (const MMContent& content : mmc) {
+      if (content.type == "text") {
+        continue;
+      }
+      MMErrCode code = collect_content(content, inputs, refs);
+      if (code != MMErrCode::SUCCESS) {
+        return code;
+      }
+    }
+  }
+  return MMErrCode::SUCCESS;
+}
+
+MMErrCode MMInputTransfer::collect_content(const MMContent& content,
+                                           MMInput& inputs,
+                                           std::vector<MMSourceRef>& refs) {
+  const std::string& type = content.type;
+  MMInputItem item;
+
+  // Embedding items have no download/decode and consume the shared payload
+  // during load, so they are handled fully here in walk order.
+  if (!is_url_type(type)) {
+    MMErrCode code =
+        mm_handlers_->process(type, content, item, inputs.payload());
+    if (code != MMErrCode::SUCCESS) {
+      return code;
+    }
+    apply_uuid_key(content, item);
+    inputs.insert(std::move(item));
+    refs.push_back(
+        MMSourceRef{&content, MMPayload{}, /*needs_materialize=*/false});
+    return MMErrCode::SUCCESS;
+  }
+
+  MMPayload sliced = slice_payload(content, inputs.payload());
+  item.type = url_type_to_modality(type);
+  apply_uuid_key(content, item);
+  inputs.insert(std::move(item));
+  refs.push_back(
+      MMSourceRef{&content, std::move(sliced), /*needs_materialize=*/true});
+  return MMErrCode::SUCCESS;
+}
+
+MMErrCode MMInputTransfer::materialize(
+    const std::vector<MMSourceRef>& refs,
+    const std::vector<int32_t>& target_indices,
+    MMInput& inputs) {
+  std::vector<MMInputItem>& items = inputs.mutable_items();
+  CHECK_EQ(refs.size(), items.size()) << "materialize refs/items size mismatch";
+
+  if (target_indices.empty()) {
+    return MMErrCode::SUCCESS;
+  }
+
+  std::vector<int32_t> work;
+  work.reserve(target_indices.size());
+  for (int32_t input_index : target_indices) {
+    CHECK_GE(input_index, 0);
+    CHECK_LT(static_cast<size_t>(input_index), items.size());
+    if (refs[input_index].needs_materialize) {
+      work.push_back(input_index);
+    }
+  }
+  if (work.empty()) {
+    return MMErrCode::SUCCESS;
+  }
+
+  std::atomic<MMErrCode> error{MMErrCode::SUCCESS};
+  BlockingCounter counter(static_cast<int32_t>(work.size()));
+  for (size_t i = 0; i < work.size(); ++i) {
+    const int32_t input_index = work[i];
+    threadpool_->schedule([&, input_index]() {
+      if (error.load() == MMErrCode::SUCCESS) {
+        const MMSourceRef& ref = refs[input_index];
+        MMInputItem& item = items[input_index];
+        MMPayload payload = ref.payload;  // get() mutates offset; keep original
+        MMErrCode code = mm_handlers_->process(
+            ref.content->type, *ref.content, item, payload);
+        if (code != MMErrCode::SUCCESS) {
+          LOG(ERROR) << "materialize failed at input index " << input_index
+                     << ", type=" << ref.content->type;
+          MMErrCode expected = MMErrCode::SUCCESS;
+          error.compare_exchange_strong(expected, code);
+        }
+      }
+      counter.decrement_count();
+    });
+  }
+  counter.wait();
+  return error.load();
+}
+
 MMErrCode MMInputTransfer::trans_parallel(const MMContentVec& mmc,
                                           std::vector<MMInputItem>& inputs,
                                           MMPayload& payload) {

xllm/core/framework/multimodal/mm_input.h

@@ -42,6 +42,7 @@ struct MMInputItem {
     type = MMType::NONE;
     raw_data.clear();
     hash_key.reset();
+    uuid.reset();
   }
 
   std::optional<torch::Tensor> get_decode_data(MMType type_) const {
@@ -64,6 +65,7 @@ struct MMInputItem {
 
   std::string raw_data;  // binary
   std::optional<XXH3Key> hash_key;
+  std::optional<std::string> uuid;
 
   torch::Tensor decode_image;  // image: rgb, [c,h,w], uint8
   torch::Tensor decode_video;  // video: rgb, [t,c,h,w], uint8
@@ -109,11 +111,14 @@ class MMInput {
   size_t size() const { return items_.size(); }
 
   const std::vector<MMInputItem>& items() const { return items_; }
+  std::vector<MMInputItem>& mutable_items() { return items_; }
 
   void insert(const std::vector<MMInputItem>& inputs) {
     items_.insert(items_.end(), inputs.begin(), inputs.end());
   }
 
+  void insert(MMInputItem input) { items_.push_back(std::move(input)); }
+
   std::vector<torch::Tensor> get_decode_data(MMType type) const {
     std::vector<torch::Tensor> vec;
 
@@ -176,14 +181,36 @@ inline const char* MMErrToString(MMErrCode code) {
 
 class MMHandlerSet;
 class ThreadPool;
+
+// Download source for one item, produced by collect() and consumed by
+// materialize(). `content` points into the caller's `messages`. Scoped to a
+// single request; never stored on MMInputItem.
+struct MMSourceRef {
+  const MMContent* content = nullptr;
+  MMPayload payload;
+  bool needs_materialize =
+      false;  // false for embedding items (done in collect)
+};
+
 class MMInputTransfer {
  public:
   MMInputTransfer();
   ~MMInputTransfer();
 
   MMErrCode trans(const std::vector<Message>& messages, MMInput& inputs);
 
+  MMErrCode collect(const std::vector<Message>& messages,
+                    MMInput& inputs,
+                    std::vector<MMSourceRef>& refs);
+  MMErrCode materialize(const std::vector<MMSourceRef>& refs,
+                        const std::vector<int32_t>& target_indices,
+                        MMInput& inputs);
+
  private:
+  MMErrCode collect_content(const MMContent& content,
+                            MMInput& inputs,
+                            std::vector<MMSourceRef>& refs);
+
   MMErrCode trans_parallel(const MMContentVec& mmc,
                            std::vector<MMInputItem>& inputs,
                            MMPayload& payload);

xllm/core/framework/multimodal/mm_visitor.cpp

@@ -420,9 +420,41 @@ bool ProcessorCacheLookupVisitor::visit(const MMInputItem& input) {
   return true;
 }
 
+namespace {
+// True when exactly one of IMAGE/VIDEO/AUDIO is set. Multimodal-bit items
+// (audio-in-video) expand to multiple outputs and can't be keyed by one uuid,
+// so the pre-filter never treats them as hits.
+bool is_single_modality(uint32_t type) {
+  const uint32_t modality =
+      type & (MMType::IMAGE | MMType::VIDEO | MMType::AUDIO);
+  return modality != 0 && (modality & (modality - 1)) == 0;
+}
+}  // namespace
+
+UuidPrefilterVisitor::UuidPrefilterVisitor(ProcessorCache& cache,
+                                           size_t item_count)
+    : cache_(cache) {
+  hit_items_.reserve(item_count);
+  hit_indices_.reserve(item_count);
+  miss_indices_.reserve(item_count);
+}
+
+bool UuidPrefilterVisitor::visit(const MMInputItem& input) {
+  const int32_t current = index_++;
+  if (input.uuid.has_value() && is_single_modality(input.type)) {
+    std::optional<MMDataItem> hit = cache_.lookup(input.hash_key.value());
+    if (hit.has_value()) {
+      hit_items_.emplace_back(std::move(hit.value()));
+      hit_indices_.push_back(current);
+      return true;
+    }
+  }
+  miss_indices_.push_back(current);
+  return true;
+}
+
 ProcessorCacheInsertVisitor::ProcessorCacheInsertVisitor(ProcessorCache& cache)
     : cache_(cache) {}
-
 bool ProcessorCacheInsertVisitor::visit(MMDataItem& item) {
   if (!item.is_embedded()) {
     cache_.insert(item.state().schedule_data().key, item);

xllm/core/framework/multimodal/mm_visitor.h

@@ -180,6 +180,23 @@ class ProcessorCacheLookupVisitor final : public MMInputItem::IVisitor {
   ProcessorCache& cache_;
 };
 
+// Looks up each uuid-carrying item in the processor cache before download.
+// Hits are collected with their original index; misses only record the index
+// so they can be materialized and processed afterwards.
+class UuidPrefilterVisitor final : public MMInputItem::IVisitor {
+ public:
+  UuidPrefilterVisitor(ProcessorCache& cache, size_t item_count);
+  bool visit(const MMInputItem& input) override;
+
+  MMItemVec hit_items_;
+  std::vector<int32_t> hit_indices_;
+  std::vector<int32_t> miss_indices_;
+
+ private:
+  ProcessorCache& cache_;
+  int32_t index_ = 0;
+};
+
 class ProcessorCacheInsertVisitor final : public MMDataItem::IVisitor {
  public:
   explicit ProcessorCacheInsertVisitor(ProcessorCache& cache);