Servable framework refactor - phase 2

spelling-whitelist.txt

@@ -16,6 +16,7 @@ src/test/modelconfig_test.cpp:656: OptionA
 src/test/modelconfig_test.cpp:662: OptionA
 src/test/modelconfig_test.cpp:668: OptionA
 src/test/modelinstance_test.cpp:1093: THROUGHTPUT
+src/test/llm/output_parsers/qwen3_output_parser_test.cpp:720: thi
 third_party/aws-sdk-cpp/aws-sdk-cpp.bz
 WORKSPACE:98: thirdparty
 demos/classification_using_paddlepaddle_model/python/utils/imagenet_class_index.json
@@ -38,4 +39,4 @@ windows_parse_tests.bat:136: SEH ==> SHE
 windows_parse_tests.bat:141: SEH ==> SHE
 windows_parse_tests.bat:144: SEH ==> SHE
 src/test/llm/output_parsers/gemma4_output_parser_test.cpp
-src/test/llm/output_parsers/qwen3_output_parser_test.cpp:719: thi ==> the, this
+src/test/llm/output_parsers/qwen3_output_parser_test.cpp:715: thi ==> the, this

src/BUILD

@@ -2922,11 +2922,12 @@ cc_library(
     name = "test_llm_output_parser_tests",
     linkstatic = 1,
     alwayslink = True,
+    hdrs = ["test/llm/output_parsers/output_parser_test_utils.hpp"],
     srcs = glob(["test/llm/output_parsers/*_test.cpp"]),
     deps = [
         "@com_google_googletest//:gtest",
         ":test_platform_utils",
-        "//src/llm:output_parsers",
+        "//src/llm:text_streamer",
     ],
     copts = COPTS_TESTS,
     local_defines = COMMON_LOCAL_DEFINES,

src/llm/BUILD

@@ -272,7 +272,8 @@ ovms_cc_library(
 
 ovms_cc_library(
     name = "io_processing_base_output_parser",
-    hdrs = ["io_processing/base_output_parser.hpp"],
+    hdrs = ["io_processing/base_output_parser.hpp",
+            "io_processing/output_parsing_config.hpp"],
     srcs = ["io_processing/base_output_parser.cpp"],
     deps = [
         "@com_github_tencent_rapidjson//:rapidjson",
@@ -312,9 +313,9 @@ ovms_cc_library(
 )
 
 ovms_cc_library(
-    name = "io_processing_lfm2_utils",
-    hdrs = ["io_processing/lfm2/lfm2_utils.hpp"],
-    srcs = ["io_processing/lfm2/lfm2_utils.cpp"],
+    name = "io_processing_lfm2_tool_parser",
+    hdrs = ["io_processing/lfm2/lfm2_tool_parser.hpp"],
+    srcs = ["io_processing/lfm2/lfm2_tool_parser.cpp"],
     deps = [
         "@com_github_tencent_rapidjson//:rapidjson",
         "//src/port:rapidjson_document",
@@ -327,28 +328,6 @@ ovms_cc_library(
     visibility = ["//visibility:public"],
 )
 
-ovms_cc_library(
-    name = "io_processing_lfm2_tool_parser",
-    hdrs = ["io_processing/lfm2/lfm2_tool_parser.hpp"],
-    srcs = ["io_processing/lfm2/lfm2_tool_parser.cpp"],
-    deps = [
-        ":io_processing_lfm2_utils",
-    ],
-    visibility = ["//visibility:public"],
-)
-
-
-ovms_cc_library(
-    name = "io_processing_lfm25_tool_parser",
-    hdrs = ["io_processing/lfm2/lfm25_tool_parser.hpp"],
-    srcs = ["io_processing/lfm2/lfm25_tool_parser.cpp"],
-    deps = [
-        ":io_processing_lfm2_utils",
-        "//src:libovmslogging",
-    ],
-    visibility = ["//visibility:public"],
-)
-
 ovms_cc_library(
     name = "io_processing_gemma4_tool_parser",
     hdrs = ["io_processing/gemma4/gemma4_tool_parser.hpp", "io_processing/gemma4/gemma4_reasoning_parser.hpp"],
@@ -456,7 +435,6 @@ ovms_cc_library( # TODO split further so we don't have to recompile everything w
         ":io_processing_parser_config_validation",
         ":io_processing_qwen3coder_tool_parser",
         ":io_processing_lfm2_tool_parser",
-        ":io_processing_lfm25_tool_parser",
         ":io_processing_gemma4_tool_parser",
         ":io_processing_minicpm5_tool_parser",
         ":io_processing_qwen3_reasoning_parser",
@@ -487,10 +465,22 @@ ovms_cc_library(
     visibility = ["//visibility:public"],
 )
 
+ovms_cc_library(
+    name = "text_streamer",
+    hdrs = ["ovms_text_streamer.hpp"],
+    srcs = ["ovms_text_streamer.cpp"],
+    deps = [
+        "@com_github_tencent_rapidjson//:rapidjson",
+        "//src:libovmslogging",
+        "//third_party:genai",
+        ":output_parsers",
+    ],
+    visibility = ["//visibility:public"],
+)
+
 ovms_cc_library(
     name = "genai_servables",
     hdrs = ["servable.hpp",
-            "ovms_text_streamer.hpp",
             "servable_initializer.hpp", 
             "language_model/continuous_batching/servable.hpp",
             "language_model/continuous_batching/llm_executor.hpp",
@@ -508,7 +498,6 @@ ovms_cc_library(
             "text_utils.hpp"],
     srcs = ["servable.cpp",
             "servable_initializer.cpp",
-            "ovms_text_streamer.cpp",
             "language_model/continuous_batching/servable.cpp",
             "language_model/continuous_batching/servable_initializer.cpp",
             "visual_language_model/continuous_batching/servable.cpp",
@@ -535,6 +524,7 @@ ovms_cc_library(
         "//src/filesystem:libovmsfilesystem",
         "//src/tokenize:tokenize_parser",
         "llmcalculator_cc_proto",
+        ":text_streamer",
         ":openai_completions_api_handler",
         ":openai_responses_handler",
         ":generation_config_builders",

src/llm/apis/openai_api_handler.cpp

@@ -99,19 +99,6 @@ std::string OpenAIApiHandler::serializeFailedEvent(const std::string& errorMessa
     return "";
 }
 
-std::vector<int64_t> OpenAIApiHandler::encodeTextToTokens(const std::string& text) {
-    auto result = tokenizer.encode(text);
-    auto& input_ids = result.input_ids;
-    if (input_ids.get_shape().size() != 2)
-        throw std::runtime_error("input_ids should have 2 dimensions");
-    if (input_ids.get_shape()[0] != 1)
-        throw std::runtime_error("input_ids should have 1 batch size");
-    if (input_ids.get_element_type() != ov::element::i64)
-        throw std::runtime_error("input_ids should have i64 element type");
-    int64_t* data = reinterpret_cast<int64_t*>(input_ids.data());
-    return std::vector<int64_t>(data, data + input_ids.get_shape()[1]);
-}
-
 absl::Status OpenAIApiHandler::parseResponseFormat() {
     auto it = doc.FindMember("response_format");
     if (it != doc.MemberEnd()) {
@@ -395,15 +382,61 @@ void OpenAIApiHandler::incrementProcessedTokens(size_t numTokens) {
     usage.completionTokens += numTokens;
 }
 
-ParsedOutput OpenAIApiHandler::parseOutputIfNeeded(const std::vector<int64_t>& generatedIds) {
-    OVMS_PROFILE_FUNCTION();
-    ParsedOutput parsedOutput;
-    if ((endpoint != Endpoint::CHAT_COMPLETIONS && endpoint != Endpoint::RESPONSES) || outputParser == nullptr) {
-        parsedOutput.content = this->tokenizer.decode(generatedIds, ov::genai::skip_special_tokens(request.skipSpecialTokens));
-    } else {
-        parsedOutput = outputParser->parse(generatedIds, this->areToolsAvailable());
+std::string OpenAIApiHandler::serializeUnaryResponse(
+    const std::vector<std::vector<rapidjson::Document>>& allDeltas,
+    const std::vector<ov::genai::GenerationFinishReason>& finishReasons) {
+    return serializeUnaryResponse(allDeltas, finishReasons, {});
+}
+
+ParsedOutput OpenAIApiHandler::parsedOutputFromDeltas(const std::vector<rapidjson::Document>& deltas) {
+    ParsedOutput output;
+    // tool calls keyed by index; index values are expected to be dense starting from 0
+    std::vector<ToolCall> toolCalls;
+    for (const auto& doc : deltas) {
+        if (!doc.IsObject() || !doc.HasMember("delta")) {
+            continue;  // empty finish-only chunk
+        }
+        const auto& delta = doc["delta"];
+        if (!delta.IsObject()) {
+            continue;
+        }
+        if (delta.HasMember("content") && delta["content"].IsString()) {
+            output.content += delta["content"].GetString();
+        }
+        if (delta.HasMember("reasoning_content") && delta["reasoning_content"].IsString()) {
+            output.reasoning += delta["reasoning_content"].GetString();
+        }
+        if (delta.HasMember("tool_calls") && delta["tool_calls"].IsArray()) {
+            for (const auto& tcEntry : delta["tool_calls"].GetArray()) {
+                if (!tcEntry.IsObject() || !tcEntry.HasMember("index")) {
+                    continue;
+                }
+                const int rawIdx = tcEntry["index"].GetInt();
+                if (rawIdx < 0) {
+                    continue;
+                }
+                const auto idx = static_cast<size_t>(rawIdx);
+                if (idx >= toolCalls.size()) {
+                    toolCalls.resize(idx + 1);
+                }
+                ToolCall& tc = toolCalls[idx];
+                if (tcEntry.HasMember("id") && tcEntry["id"].IsString()) {
+                    tc.id = tcEntry["id"].GetString();
+                }
+                if (tcEntry.HasMember("function") && tcEntry["function"].IsObject()) {
+                    const auto& fn = tcEntry["function"];
+                    if (fn.HasMember("name") && fn["name"].IsString()) {
+                        tc.name = fn["name"].GetString();
+                    }
+                    if (fn.HasMember("arguments") && fn["arguments"].IsString()) {
+                        tc.arguments += fn["arguments"].GetString();
+                    }
+                }
+            }
+        }
     }
-    return parsedOutput;
+    output.toolCalls = std::move(toolCalls);
+    return output;
 }
 
 // --- Free functions ---

src/llm/apis/openai_api_handler.hpp

@@ -87,6 +87,13 @@ struct CompletionUsageStatistics {
     }
 };
 
+// Per-choice raw token data needed to build logprob objects in unary responses.
+// populated in GenAiServable::prepareCompleteResponse from GenerationOutput.
+struct UnaryChoiceLogprobs {
+    std::vector<int64_t> generatedIds;
+    std::vector<float> logProbs;
+};
+
 // Abstract base class for OpenAI API handlers.
 // Holds common state (request, doc, tokenizer, usage, output parser) and implements
 // shared parsing logic. Endpoint-specific parsing and serialization are pure virtual.
@@ -117,10 +124,10 @@ class OpenAIApiHandler {
     absl::Status parseCommonPart(std::optional<uint32_t> maxTokensLimit, uint32_t bestOfLimit, std::optional<uint32_t> maxModelLength);
     absl::Status parseResponseFormat();
     absl::Status ensureArgumentsInToolCalls(Value& messageObj);
-    ParsedOutput parseOutputIfNeeded(const std::vector<int64_t>& generatedIds);
-
-    // Shared VLM workaround: encode text to tokens using tokenizer, validates shape
-    std::vector<int64_t> encodeTextToTokens(const std::string& text);
+    // Assemble a ParsedOutput from a sequence of streaming delta Documents produced by OVMSTextStreamer.
+    // Each document has the shape {"delta":{...}} as emitted by flush_chunk, or an empty object for
+    // finish-only chunks.  Content, reasoning, and tool-call fragments are accumulated in order.
+    static ParsedOutput parsedOutputFromDeltas(const std::vector<rapidjson::Document>& deltas);
 
 public:
     OpenAIApiHandler(Document& doc, Endpoint endpoint, std::chrono::time_point<std::chrono::system_clock> creationTime,
@@ -198,9 +205,19 @@ class OpenAIApiHandler {
     virtual void incrementProcessedTokens(size_t numTokens = 1);
 
     // Serialization - pure virtual, each handler produces its own response format
-    virtual std::string serializeUnaryResponse(const std::vector<ov::genai::GenerationOutput>& generationOutputs) = 0;
-    virtual std::string serializeUnaryResponse(ov::genai::EncodedResults& results) = 0;
-    virtual std::string serializeUnaryResponse(ov::genai::VLMDecodedResults& results, const std::string& textResponse) = 0;
+    // Phase 2: delta-based unary serialisation — assembles a complete response from streaming
+    // delta Documents collected via deltaChannel after OVMSTextStreamer finishes.
+    // Single-choice variant (used by Legacy servables).
+    virtual std::string serializeUnaryResponse(const std::vector<rapidjson::Document>& deltas, ov::genai::GenerationFinishReason finishReason) = 0;
+    // Multi-choice variant: N delta-vectors (one per sequence) + per-sequence finish reasons.
+    // logprobData may be empty when logprobs are not requested; otherwise its size equals
+    // allDeltas.size(). Used by ContinuousBatchingServable for both n=1 and n>1.
+    virtual std::string serializeUnaryResponse(const std::vector<std::vector<rapidjson::Document>>& allDeltas,
+        const std::vector<ov::genai::GenerationFinishReason>& finishReasons,
+        const std::vector<UnaryChoiceLogprobs>& logprobData) = 0;
+    // Convenience overload: no logprobs (delegates to the virtual above with empty logprobData).
+    std::string serializeUnaryResponse(const std::vector<std::vector<rapidjson::Document>>& allDeltas,
+        const std::vector<ov::genai::GenerationFinishReason>& finishReasons);
     virtual std::string serializeStreamingChunk(rapidjson::Document parsedDelta, ov::genai::GenerationFinishReason finishReason) = 0;
     virtual std::string serializeStreamingUsageChunk() = 0;
     virtual std::string serializeStreamingHandshakeChunk() = 0;