diff --git a/xllm/api_service/mm_service_utils.h b/xllm/api_service/mm_service_utils.h index 8e1c324d73..5aeece71c9 100644 --- a/xllm/api_service/mm_service_utils.h +++ b/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)); diff --git a/xllm/core/common/message.h b/xllm/core/common/message.h index a0315f4cb9..f655aafe80 100644 --- a/xllm/core/common/message.h +++ b/xllm/core/common/message.h @@ -64,6 +64,8 @@ struct MMContent { AudioURL audio_url; Embedding embedding; + + std::optional uuid; }; using MMContentVec = std::vector; diff --git a/xllm/core/distributed_runtime/vlm_master.cpp b/xllm/core/distributed_runtime/vlm_master.cpp index 95a85c0f6b..2447877f29 100644 --- a/xllm/core/distributed_runtime/vlm_master.cpp +++ b/xllm/core/distributed_runtime/vlm_master.cpp @@ -469,20 +469,8 @@ std::shared_ptr 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; diff --git a/xllm/core/framework/multimodal/mm_data.cpp b/xllm/core/framework/multimodal/mm_data.cpp index 182edf7b5a..6976386fcf 100644 --- a/xllm/core/framework/multimodal/mm_data.cpp +++ b/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; diff --git a/xllm/core/framework/multimodal/mm_data.h b/xllm/core/framework/multimodal/mm_data.h index a685895efc..1b44fffea1 100644 --- a/xllm/core/framework/multimodal/mm_data.h +++ b/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; } diff --git a/xllm/core/framework/multimodal/mm_handler.cpp b/xllm/core/framework/multimodal/mm_handler.cpp index ce6290bcb2..b95259d282 100644 --- a/xllm/core/framework/multimodal/mm_handler.cpp +++ b/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; diff --git a/xllm/core/framework/multimodal/mm_input.cpp b/xllm/core/framework/multimodal/mm_input.cpp index e8926914fd..ac47e926b3 100644 --- a/xllm/core/framework/multimodal/mm_input.cpp +++ b/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& messages, return MMErrCode::SUCCESS; } +MMErrCode MMInputTransfer::collect(const std::vector& messages, + MMInput& inputs, + std::vector& refs) { + inputs.clear(); + refs.clear(); + + for (const Message& message : messages) { + const MMContentVec& mmc = std::get(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& 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& refs, + const std::vector& target_indices, + MMInput& inputs) { + std::vector& items = inputs.mutable_items(); + CHECK_EQ(refs.size(), items.size()) << "materialize refs/items size mismatch"; + + if (target_indices.empty()) { + return MMErrCode::SUCCESS; + } + + std::vector work; + work.reserve(target_indices.size()); + for (int32_t input_index : target_indices) { + CHECK_GE(input_index, 0); + CHECK_LT(static_cast(input_index), items.size()); + if (refs[input_index].needs_materialize) { + work.push_back(input_index); + } + } + if (work.empty()) { + return MMErrCode::SUCCESS; + } + + std::atomic error{MMErrCode::SUCCESS}; + BlockingCounter counter(static_cast(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& inputs, MMPayload& payload) { diff --git a/xllm/core/framework/multimodal/mm_input.h b/xllm/core/framework/multimodal/mm_input.h index 098582cba1..1f210e11c2 100644 --- a/xllm/core/framework/multimodal/mm_input.h +++ b/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 get_decode_data(MMType type_) const { @@ -64,6 +65,7 @@ struct MMInputItem { std::string raw_data; // binary std::optional hash_key; + std::optional 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& items() const { return items_; } + std::vector& mutable_items() { return items_; } void insert(const std::vector& inputs) { items_.insert(items_.end(), inputs.begin(), inputs.end()); } + void insert(MMInputItem input) { items_.push_back(std::move(input)); } + std::vector get_decode_data(MMType type) const { std::vector vec; @@ -176,6 +181,17 @@ 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(); @@ -183,7 +199,18 @@ class MMInputTransfer { MMErrCode trans(const std::vector& messages, MMInput& inputs); + MMErrCode collect(const std::vector& messages, + MMInput& inputs, + std::vector& refs); + MMErrCode materialize(const std::vector& refs, + const std::vector& target_indices, + MMInput& inputs); + private: + MMErrCode collect_content(const MMContent& content, + MMInput& inputs, + std::vector& refs); + MMErrCode trans_parallel(const MMContentVec& mmc, std::vector& inputs, MMPayload& payload); diff --git a/xllm/core/framework/multimodal/mm_visitor.cpp b/xllm/core/framework/multimodal/mm_visitor.cpp index cfe58da2a6..015dbbee4f 100644 --- a/xllm/core/framework/multimodal/mm_visitor.cpp +++ b/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 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); diff --git a/xllm/core/framework/multimodal/mm_visitor.h b/xllm/core/framework/multimodal/mm_visitor.h index 955203b03f..3e97f670db 100644 --- a/xllm/core/framework/multimodal/mm_visitor.h +++ b/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 hit_indices_; + std::vector miss_indices_; + + private: + ProcessorCache& cache_; + int32_t index_ = 0; +}; + class ProcessorCacheInsertVisitor final : public MMDataItem::IVisitor { public: explicit ProcessorCacheInsertVisitor(ProcessorCache& cache); diff --git a/xllm/processors/cacheable_multimodal_processor.cpp b/xllm/processors/cacheable_multimodal_processor.cpp index 33bcd3e496..99203fbe02 100644 --- a/xllm/processors/cacheable_multimodal_processor.cpp +++ b/xllm/processors/cacheable_multimodal_processor.cpp @@ -42,63 +42,88 @@ bool CacheableMultimodalProcessor::process_prompt( bool CacheableMultimodalProcessor::process_multimodal(const MMInput& inputs, MMData& data) const { - ProcessorCacheLookupVisitor cache_lookup_visitor(*cache_, inputs.size()); - CHECK(inputs.foreach (cache_lookup_visitor)); + return inner_->process_multimodal(inputs, data); +} - MMItemVec miss_items; - if (!process_misses(cache_lookup_visitor.miss_inputs_, miss_items)) { +bool CacheableMultimodalProcessor::process_mm_input( + const std::vector& messages, + std::string payload, + MMData& out) { + MMInput mm_inputs(std::move(payload)); + std::vector refs; + if (transfer_.collect(messages, mm_inputs, refs) != MMErrCode::SUCCESS) { return false; } - - assemble(cache_lookup_visitor.cache_hits_, std::move(miss_items), data); - return true; -} - -bool CacheableMultimodalProcessor::process_misses( - const std::vector& miss_inputs, - MMItemVec& miss_items) const { - if (miss_inputs.empty()) { + if (mm_inputs.empty()) { return true; } - MMInput inputs; - inputs.insert(miss_inputs); - MMData miss_data; - if (!inner_->process_multimodal(inputs, miss_data)) { + UuidPrefilterVisitor prefilter(*cache_, mm_inputs.size()); + CHECK(mm_inputs.foreach (prefilter)); + + if (transfer_.materialize(refs, prefilter.miss_indices_, mm_inputs) != + MMErrCode::SUCCESS) { return false; } - CHECK_EQ(miss_data.items().size(), miss_inputs.size()) - << "Multimodal processor returned mismatched item count."; - ProcessorCacheInsertVisitor insert(*cache_); - CHECK(miss_data.foreach (insert)); - miss_items = std::move(miss_data.items()); + + const std::vector& items = mm_inputs.items(); + MMInput uuid_misses; + for (int32_t index : prefilter.miss_indices_) { + uuid_misses.insert(items[index]); + } + ProcessorCacheLookupVisitor raw_hash_lookup(*cache_, uuid_misses.size()); + CHECK(uuid_misses.foreach (raw_hash_lookup)); + CHECK_EQ(raw_hash_lookup.cache_hits_.size(), prefilter.miss_indices_.size()); + + MMData produced_data; + if (!raw_hash_lookup.miss_inputs_.empty()) { + MMInput preprocess_inputs; + preprocess_inputs.insert(raw_hash_lookup.miss_inputs_); + if (!inner_->process_multimodal(preprocess_inputs, produced_data)) { + return false; + } + CHECK_EQ(produced_data.items().size(), + raw_hash_lookup.miss_inputs_.size()); + ProcessorCacheInsertVisitor insert(*cache_); + CHECK(produced_data.foreach (insert)); + } + + MMItemVec fresh_items; + if (produced_data.hold()) { + fresh_items = std::move(produced_data.items()); + } + assemble(prefilter, raw_hash_lookup, fresh_items, out); return true; } void CacheableMultimodalProcessor::assemble( - std::vector>& cache_hits, - MMItemVec miss_items, - MMData& data) const { + UuidPrefilterVisitor& prefilter, + ProcessorCacheLookupVisitor& raw_hash_lookup, + MMItemVec& fresh_items, + MMData& out) const { + const int32_t total = static_cast(prefilter.hit_indices_.size() + + prefilter.miss_indices_.size()); + MMItemVec slots(total, MMDataItem(MMType::NONE)); uint32_t full_type = MMType::NONE; - MMItemVec full_items; - full_items.reserve(cache_hits.size()); - - size_t miss_index = 0; - for (std::optional& cache_hit : cache_hits) { - if (cache_hit.has_value()) { - MMDataItem& item = cache_hit.value(); - full_type |= item.type(); - full_items.emplace_back(std::move(item)); - continue; - } - MMDataItem& produced = miss_items[miss_index++]; - full_type |= produced.type(); - full_items.emplace_back(std::move(produced)); + for (size_t i = 0; i < prefilter.hit_indices_.size(); ++i) { + MMDataItem& item = prefilter.hit_items_[i]; + full_type |= item.type(); + slots[prefilter.hit_indices_[i]] = std::move(item); + } + + size_t fresh_index = 0; + for (size_t j = 0; j < prefilter.miss_indices_.size(); ++j) { + const int32_t global_index = prefilter.miss_indices_[j]; + std::optional& cache_hit = raw_hash_lookup.cache_hits_[j]; + MMDataItem& item = + cache_hit.has_value() ? cache_hit.value() : fresh_items[fresh_index++]; + full_type |= item.type(); + slots[global_index] = std::move(item); } - CHECK_EQ(miss_index, miss_items.size()); + CHECK_EQ(fresh_index, fresh_items.size()); - data.set(full_type, std::move(full_items)); + out = MMData(full_type, std::move(slots)); } } // namespace xllm diff --git a/xllm/processors/cacheable_multimodal_processor.h b/xllm/processors/cacheable_multimodal_processor.h index bc5f344f21..0a92413232 100644 --- a/xllm/processors/cacheable_multimodal_processor.h +++ b/xllm/processors/cacheable_multimodal_processor.h @@ -21,6 +21,7 @@ limitations under the License. #include #include +#include "core/framework/multimodal/mm_visitor.h" #include "processors/multimodal_processor.h" #include "processors/processor_cache.h" @@ -37,12 +38,20 @@ class CacheableMultimodalProcessor final : public MultimodalProcessorBase { std::vector& token_ids) override; bool process_multimodal(const MMInput& inputs, MMData& data) const override; + // UUID pre-filter path: collect (no download) -> uuid pre-filter -> + // materialize only misses -> raw-hash lookup -> preprocess -> insert -> + // scatter by index. + bool process_mm_input(const std::vector& messages, + std::string payload, + MMData& out) override; + private: - bool process_misses(const std::vector& miss_inputs, - MMItemVec& miss_items) const; - void assemble(std::vector>& cache_hits, - MMItemVec miss_items, - MMData& data) const; + // Scatters uuid hits, raw-hash hits and freshly-processed items back to their + // original request positions. + void assemble(UuidPrefilterVisitor& prefilter, + ProcessorCacheLookupVisitor& raw_hash_lookup, + MMItemVec& fresh_items, + MMData& out) const; std::unique_ptr inner_; std::unique_ptr cache_; diff --git a/xllm/processors/multimodal_processor.cpp b/xllm/processors/multimodal_processor.cpp index 2e36465627..1bf80daab8 100644 --- a/xllm/processors/multimodal_processor.cpp +++ b/xllm/processors/multimodal_processor.cpp @@ -34,6 +34,29 @@ MultimodalProcessorBase::MultimodalProcessorBase( MultimodalProcessorBase::~MultimodalProcessorBase() = default; +bool MultimodalProcessorBase::process_mm_input( + const std::vector& messages, + std::string payload, + MMData& out) { + MMInput inputs(std::move(payload)); + std::vector refs; + if (transfer_.collect(messages, inputs, refs) != MMErrCode::SUCCESS) { + return false; + } + if (inputs.empty()) { + return true; + } + std::vector all; + all.reserve(inputs.size()); + for (int32_t i = 0; i < static_cast(inputs.size()); ++i) { + all.push_back(i); + } + if (transfer_.materialize(refs, all, inputs) != MMErrCode::SUCCESS) { + return false; + } + return process_multimodal(inputs, out); +} + bool MultimodalProcessorBase::tokenize(const std::string& prompt, std::vector& token_ids) const { Timer timer; diff --git a/xllm/processors/multimodal_processor.h b/xllm/processors/multimodal_processor.h index 98916cc6f0..d2026a8032 100644 --- a/xllm/processors/multimodal_processor.h +++ b/xllm/processors/multimodal_processor.h @@ -51,6 +51,10 @@ class MultimodalProcessorBase { virtual bool process_multimodal(const MMInput& inputs, MMData& data) const = 0; + virtual bool process_mm_input(const std::vector& messages, + std::string payload, + MMData& out); + protected: MultimodalProcessorBase( std::shared_ptr tokenizer, @@ -63,6 +67,11 @@ class MultimodalProcessorBase { void pad_to_max_length(std::vector& token_ids) const; + // Download/decode helper used by process_mm_input. Stateless and thread-safe; + // each processor owns its own (the decorator's inner one stays idle since + // only the outermost object's process_mm_input is called). + MMInputTransfer transfer_; + private: std::shared_ptr tokenizer_; TokenizerArgs tokenizer_args_; diff --git a/xllm/proto/multimodal.proto b/xllm/proto/multimodal.proto index ff3b56fbcd..948c536f56 100644 --- a/xllm/proto/multimodal.proto +++ b/xllm/proto/multimodal.proto @@ -47,6 +47,9 @@ message MMInputData { optional Embedding image_embedding = 6; optional Embedding video_embedding = 7; optional Embedding audio_embedding = 8; + + // Optional client-supplied stable identifier for a multimodal item. + optional string uuid = 9; } message MMChatMessage {