diff --git a/src/llm/io_processing/minicpm5/minicpm5_tool_parser.cpp b/src/llm/io_processing/minicpm5/minicpm5_tool_parser.cpp index 36258dd6ec..81aa7d14c9 100644 --- a/src/llm/io_processing/minicpm5/minicpm5_tool_parser.cpp +++ b/src/llm/io_processing/minicpm5/minicpm5_tool_parser.cpp @@ -51,25 +51,6 @@ Minicpm5ToolParserImpl::Minicpm5ToolParserImpl(const ToolsParameterTypeMap_t& to * tagEnd is the position of the '>' that closes the enclosing tag. * Returns the extracted value, or empty string on failure. */ -std::string Minicpm5ToolParserImpl::extractNameAttribute( - const std::string& content, size_t nameAttrValueStart, size_t tagEnd) { - if (nameAttrValueStart >= tagEnd || nameAttrValueStart >= content.size()) { - return {}; - } - char quote = content[nameAttrValueStart]; - if (quote != '"' && quote != '\'') { - // No quote: read until next whitespace or '>' - size_t end = content.find_first_of(" \t\n\r>/", nameAttrValueStart); - if (end == std::string::npos || end > tagEnd) - end = tagEnd; - return content.substr(nameAttrValueStart, end - nameAttrValueStart); - } - size_t closeQuote = content.find(quote, nameAttrValueStart + 1); - if (closeQuote == std::string::npos || closeQuote > tagEnd) { - return {}; - } - return content.substr(nameAttrValueStart + 1, closeQuote - nameAttrValueStart - 1); -} void Minicpm5ToolParserImpl::addParameterToCurrentFunctionDoc(std::string& parameterValueAsString) { if (this->removeNewlineAroundParameters) @@ -130,35 +111,6 @@ void Minicpm5ToolParserImpl::addParameterToCurrentFunctionDoc(std::string& param } } -Status Minicpm5ToolParserImpl::removeToolCallsFromContentIfNeeded(std::string& outContent) { - if (toolCallPositions.begin.size() != toolCallPositions.end.size()) { - SPDLOG_DEBUG("Minicpm5: mismatched tool tags, begin: {}, end: {}", - toolCallPositions.begin.size(), toolCallPositions.end.size()); - return Status(StatusCode::INTERNAL_ERROR, "Mismatched tool tags"); - } - while (!toolCallPositions.begin.empty() && !toolCallPositions.end.empty()) { - auto posBegin = toolCallPositions.begin.top(); - auto posEnd = toolCallPositions.end.top(); - SPDLOG_TRACE("Minicpm5: removing tool call from outContent begin:{}, end:{}", posBegin, posEnd); - outContent.erase(posBegin, posEnd - posBegin); - toolCallPositions.begin.pop(); - toolCallPositions.end.pop(); - } - - const std::vector tokensToErase = { - Minicpm5ToolParser::SOS_TOKEN_STR, - Minicpm5ToolParser::EOS_TOKEN_STR}; - - for (const auto& token : tokensToErase) { - size_t pos = 0; - while ((pos = outContent.find(token, pos)) != std::string::npos) { - outContent.erase(pos, token.length()); - } - } - - return StatusCode::OK; -} - void Minicpm5ToolParserImpl::handleInsideContentState() { // Look for the next streamContent.find(Minicpm5ToolParser::FUNCTION_START_TAG, this->lastProcessedPosition); @@ -298,29 +250,6 @@ Minicpm5ToolParser::Minicpm5ToolParser(ov::genai::Tokenizer& tokenizer, const To toolsParametersTypes(createToolsParametersTypesMap(toolSchemas)), streamParser(this->toolsParametersTypes) {} -const std::vector Minicpm5ToolParser::removeReasoningTokens(const std::vector& generatedTokens) { - std::vector tokensWithoutReasoning; - tokensWithoutReasoning.reserve(generatedTokens.size()); - auto reasoningStartIt = std::find(generatedTokens.begin(), generatedTokens.end(), reasoningStartTokenId); - auto reasoningEndIt = std::find(generatedTokens.begin(), generatedTokens.end(), reasoningEndTokenId); - if (reasoningStartIt == generatedTokens.end() && reasoningEndIt == generatedTokens.end()) { - SPDLOG_LOGGER_DEBUG(llm_calculator_logger, "Minicpm5ToolParser: Reasoning start or end token not found in the generated tokens. Start token found: {}, End token found: {}, Start position: {}, End position: {}", - reasoningStartIt != generatedTokens.end(), reasoningEndIt != generatedTokens.end(), std::distance(generatedTokens.begin(), reasoningStartIt), std::distance(generatedTokens.begin(), reasoningEndIt)); - tokensWithoutReasoning.insert(tokensWithoutReasoning.end(), generatedTokens.begin(), generatedTokens.end()); - } else { - SPDLOG_LOGGER_DEBUG(llm_calculator_logger, "Minicpm5ToolParser: Reasoning tokens found. Start position: {}, End position: {}", - std::distance(generatedTokens.begin(), reasoningStartIt), std::distance(generatedTokens.begin(), reasoningEndIt)); - if (reasoningStartIt == generatedTokens.end()) { - SPDLOG_LOGGER_DEBUG(llm_calculator_logger, "Minicpm5ToolParser: Reasoning start wasn't found, but reasoning end was found. Start position: {}, End position: {}", - std::distance(generatedTokens.begin(), reasoningStartIt), std::distance(generatedTokens.begin(), reasoningEndIt)); - reasoningStartIt = generatedTokens.begin(); - } - tokensWithoutReasoning.insert(tokensWithoutReasoning.end(), generatedTokens.begin(), reasoningStartIt); - tokensWithoutReasoning.insert(tokensWithoutReasoning.end(), reasoningEndIt + 1, generatedTokens.end()); - } - return tokensWithoutReasoning; -} - std::optional Minicpm5ToolParser::sendFullDelta(const ToolCalls_t& toolCalls) { if (toolCalls.size() != 1) { SPDLOG_ERROR("Minicpm5ToolParser: for streaming expected one tool call, got: {}", toolCalls.size()); diff --git a/src/llm/io_processing/minicpm5/minicpm5_tool_parser.hpp b/src/llm/io_processing/minicpm5/minicpm5_tool_parser.hpp index bbc8e56750..771ca05d71 100644 --- a/src/llm/io_processing/minicpm5/minicpm5_tool_parser.hpp +++ b/src/llm/io_processing/minicpm5/minicpm5_tool_parser.hpp @@ -83,8 +83,6 @@ struct Minicpm5ToolParserImpl { std::optional getCurrentFunctionName() const; - Status removeToolCallsFromContentIfNeeded(std::string& outContent); - void reset() { currentState = State::Content; currentFunction.clear(); @@ -122,8 +120,6 @@ struct Minicpm5ToolParserImpl { void handleInsideParamNameState(); void handleInsideParamState(); void handleInsideAfterFunctionState(ToolCalls_t& toolCalls); - - static std::string extractNameAttribute(const std::string& content, size_t nameAttrValueStart, size_t tagEnd); }; class Minicpm5ToolParser : public BaseOutputParser { @@ -175,7 +171,6 @@ class Minicpm5ToolParser : public BaseOutputParser { std::optional parseChunk(const std::string& chunk, const std::vector& tokens, ov::genai::GenerationFinishReason finishReason) override; private: - const std::vector removeReasoningTokens(const std::vector& generatedTokens); std::optional sendFirstDeltaIfNeeded(const std::string& currentFunctionName); std::optional sendFullDelta(const ToolCalls_t& toolCalls); ToolCallDelta wrapCombinedDelta(const ToolCall& toolCall); diff --git a/src/llm/language_model/legacy/legacy_executor.cpp b/src/llm/language_model/legacy/legacy_executor.cpp index fd4e7370ef..83a9e7ef0d 100644 --- a/src/llm/language_model/legacy/legacy_executor.cpp +++ b/src/llm/language_model/legacy/legacy_executor.cpp @@ -17,6 +17,7 @@ #include "legacy_executor.hpp" #include "../../../logging.hpp" +#include "../../ovms_text_streamer.hpp" #include "servable.hpp" #include @@ -36,6 +37,11 @@ void LegacyExecutor::processRequest() { SPDLOG_LOGGER_TRACE(llm_executor_logger, "Generation started"); try { requestExecutionContext->results = pipe->generate(requestExecutionContext->inputRequest.inputIds, requestExecutionContext->inputRequest.generationConfig, requestExecutionContext->textStreamer); + auto streamer = std::dynamic_pointer_cast(requestExecutionContext->textStreamer); + if (streamer != nullptr && streamer->hadParserError()) { + requestExecutionContext->success = false; + SPDLOG_LOGGER_ERROR(llm_executor_logger, "LLM pipeline generation failed: output parser reported an error."); + } } catch (std::exception& e) { requestExecutionContext->success = false; SPDLOG_LOGGER_ERROR(llm_executor_logger, "LLM pipeline generation failed: {}.", e.what()); diff --git a/src/llm/omni_model/legacy/legacy_executor.cpp b/src/llm/omni_model/legacy/legacy_executor.cpp index 18df52bb8a..7efb8b87d6 100644 --- a/src/llm/omni_model/legacy/legacy_executor.cpp +++ b/src/llm/omni_model/legacy/legacy_executor.cpp @@ -20,6 +20,7 @@ #include #include +#include "../../ovms_text_streamer.hpp" #include "servable.hpp" namespace ovms { @@ -55,6 +56,11 @@ void OmniModelLegacyExecutor::processRequest() { requestExecutionContext->speechConfig, requestExecutionContext->textStreamer, requestExecutionContext->speechStreamer); + auto streamer = std::dynamic_pointer_cast(requestExecutionContext->textStreamer); + if (streamer != nullptr && streamer->hadParserError()) { + requestExecutionContext->success = false; + SPDLOG_LOGGER_ERROR(llm_executor_logger, "Omni pipeline generation failed: output parser reported an error."); + } } catch (std::exception& e) { requestExecutionContext->success = false; SPDLOG_LOGGER_ERROR(llm_executor_logger, "Omni pipeline generation failed: {}.", e.what()); diff --git a/src/llm/ovms_text_streamer.cpp b/src/llm/ovms_text_streamer.cpp index 5764c6e290..12f0ad9ee7 100644 --- a/src/llm/ovms_text_streamer.cpp +++ b/src/llm/ovms_text_streamer.cpp @@ -289,7 +289,18 @@ ov::genai::StreamingStatus OVMSTextStreamer::flush_chunk( std::optional delta; if (m_output_parser != nullptr) { - delta = m_output_parser->parseChunk(chunk, tokens, m_tools_available, finish_reason); + try { + delta = m_output_parser->parseChunk(chunk, tokens, m_tools_available, finish_reason); + } catch (const std::exception& e) { + // Do not let parser exceptions unwind GenAI's internal generate() call: that + // leaves the underlying InferRequest stuck "busy" for the next request. + // Cancel generation gracefully instead; hadParserError() is checked by the + // caller once generate() returns to report the failure. + SPDLOG_LOGGER_DEBUG(llm_calculator_logger, "Output parser failed, cancelling generation: {}", e.what()); + m_had_parser_error = true; + m_callback(FinishDelta{}, true); + return ov::genai::StreamingStatus::CANCEL; + } } else if (!chunk.empty()) { delta = ContentDelta{chunk}; } diff --git a/src/llm/ovms_text_streamer.hpp b/src/llm/ovms_text_streamer.hpp index a41be270bb..3d2930c52d 100644 --- a/src/llm/ovms_text_streamer.hpp +++ b/src/llm/ovms_text_streamer.hpp @@ -84,10 +84,19 @@ class OVMSTextStreamer : public ov::genai::TextStreamer { ov::genai::StreamingStatus write(const std::vector& tokens) override; void end() override; + // True when the output parser threw during a flush. Generation is cancelled + // gracefully (StreamingStatus::CANCEL) rather than letting the exception unwind + // GenAI's internal generate() loop, which would leave the underlying InferRequest + // stuck "busy" for the next request. Callers must check this after generate() + // returns to report the failure (e.g. set their own success flag). + bool hadParserError() const { return m_had_parser_error; } + private: std::shared_ptr m_output_parser; bool m_tools_available; Callback m_callback; + bool m_had_parser_error = false; + // Whether the user's request specified skip_special_tokens=false. bool m_user_wants_special = false; // Whether the current decode pass should include special tokens (skip_special_tokens=false). diff --git a/src/llm/visual_language_model/legacy/legacy_executor.cpp b/src/llm/visual_language_model/legacy/legacy_executor.cpp index 795cedccd4..65a4b61b6b 100644 --- a/src/llm/visual_language_model/legacy/legacy_executor.cpp +++ b/src/llm/visual_language_model/legacy/legacy_executor.cpp @@ -15,6 +15,7 @@ //***************************************************************************** #include "legacy_executor.hpp" +#include "../../ovms_text_streamer.hpp" #include "servable.hpp" #include @@ -41,6 +42,11 @@ void VisualLanguageModelLegacyExecutor::processRequest() { SPDLOG_LOGGER_TRACE(llm_executor_logger, "Generation started"); try { requestExecutionContext->results = pipe->generate(requestExecutionContext->inputRequest.promptText, requestExecutionContext->inputRequest.inputImages, requestExecutionContext->inputRequest.generationConfig, requestExecutionContext->textStreamer); + auto streamer = std::dynamic_pointer_cast(requestExecutionContext->textStreamer); + if (streamer != nullptr && streamer->hadParserError()) { + requestExecutionContext->success = false; + SPDLOG_LOGGER_ERROR(llm_executor_logger, "VLM pipeline generation failed: output parser reported an error."); + } } catch (std::exception& e) { requestExecutionContext->success = false; SPDLOG_LOGGER_ERROR(llm_executor_logger, "VLM pipeline generation failed: {}.", e.what()); diff --git a/src/test/http_openai_handler_test.cpp b/src/test/http_openai_handler_test.cpp index 65ea56336c..ac612cc743 100644 --- a/src/test/http_openai_handler_test.cpp +++ b/src/test/http_openai_handler_test.cpp @@ -28,7 +28,9 @@ #include "../filesystem/filesystem.hpp" #include "../llm/apis/openai_completions.hpp" #include "../llm/apis/openai_responses.hpp" +#include "../llm/io_processing/output_parser.hpp" #include "../llm/language_model/legacy/servable.hpp" +#include "../llm/ovms_text_streamer.hpp" #include "../llm/visual_language_model/legacy/servable.hpp" #include "../client_connection.hpp" #include @@ -4790,6 +4792,46 @@ TEST_F(HttpOpenAIHandlerParsingTest, legacyServablePreparePartialResponseRespons << "input_tokens must equal num_input_tokens from perf_metrics: " << response; } +TEST_F(HttpOpenAIHandlerParsingTest, legacyServableParserExceptionCancelsGenerationAndReportsFailure) { + ovms::ToolsSchemas_t emptyToolsSchema{}; + auto outputParser = std::make_shared(*tokenizer, "mistral", "", emptyToolsSchema); + + int finalCallbackCount = 0; + auto callback = [&](rapidjson::Document /*delta*/, bool isLast) -> ov::genai::StreamingStatus { + if (isLast) { + finalCallbackCount++; + } + return ov::genai::StreamingStatus::RUNNING; + }; + auto streamer = std::make_shared(*tokenizer, outputParser, /*tools_available=*/true, callback, ov::AnyMap{}); + + // "arguments" appears before "name" is known - MistralToolParser::parseChunk throws for this shape. + const std::string malformedToolCall = "[{\"arguments\": {\"x\": 1}, \"name\": \"foo\"}]"; + auto inputIds = tokenizer->encode(malformedToolCall, ov::genai::add_special_tokens(false)).input_ids; + std::vector tokens(inputIds.data(), inputIds.data() + inputIds.get_size()); + ASSERT_FALSE(tokens.empty()); + + ov::genai::StreamingStatus status = ov::genai::StreamingStatus::RUNNING; + ASSERT_NO_THROW(status = streamer->write(tokens)); + if (status == ov::genai::StreamingStatus::RUNNING) { + // Offending text may still be sitting in the delay buffer - end() flushes it. + ASSERT_NO_THROW(streamer->end()); + } + ASSERT_TRUE(streamer->hadParserError()); + EXPECT_EQ(finalCallbackCount, 1); + + // Mirrors what LegacyExecutor::processRequest does with the streamer after pipe->generate() returns. + auto ctx = makeLegacyResponsesContext(tokenizer, /*numInputTokens=*/10, /*numGeneratedTokens=*/5); + ASSERT_NE(ctx, nullptr); + ctx->textStreamer = streamer; + ctx->success = !streamer->hadParserError(); + + std::shared_ptr ctxBase = ctx; + ovms::LegacyServable servable; + EXPECT_EQ(servable.preparePartialResponse(ctxBase), + absl::InvalidArgumentError("Request processing failed, check its correctness.")); +} + TEST_F(HttpOpenAIHandlerParsingTest, vlmLegacyServablePreparePartialResponseResponsesEndpointHasCorrectUsageInCompletedEvent) { auto ctx = std::make_shared(); diff --git a/src/test/llm/output_parsers/minicpm5_output_parser_test.cpp b/src/test/llm/output_parsers/minicpm5_output_parser_test.cpp index ed223b1010..a075fdf66b 100644 --- a/src/test/llm/output_parsers/minicpm5_output_parser_test.cpp +++ b/src/test/llm/output_parsers/minicpm5_output_parser_test.cpp @@ -448,21 +448,6 @@ TEST(Minicpm5ToolParserImplTest, NewlinesAroundParamValue) { EXPECT_EQ(callsOpt.value()[0].arguments, R"({"city":"Beijing"})"); } -TEST(Minicpm5ToolParserImplTest, RemoveToolCallsFromContent) { - const std::string input = - "Before. " - R"(intel)" - " After."; - auto content = input; - Minicpm5ToolParserImpl parser(minicpm5TypeMap); - auto callsOpt = parser.parseChunk(content); - ASSERT_TRUE(callsOpt.has_value()); - auto status = parser.removeToolCallsFromContentIfNeeded(content); - EXPECT_TRUE(status.ok()) << status.string(); - EXPECT_EQ(content.find(""), std::string::npos); -} - static ToolCalls_t streamInFragments(const std::string& input, size_t fragmentSize, const ToolsParameterTypeMap_t& typeMap) { Minicpm5ToolParserImpl parser(typeMap); diff --git a/src/test/llm/output_parsers/mistral_output_parser_test.cpp b/src/test/llm/output_parsers/mistral_output_parser_test.cpp index 5307383b97..5ccae3d1e9 100644 --- a/src/test/llm/output_parsers/mistral_output_parser_test.cpp +++ b/src/test/llm/output_parsers/mistral_output_parser_test.cpp @@ -453,3 +453,28 @@ TEST_F(MistralOutputParserTest, ToolCallsWithoutToolsInTheRequestStreaming) { } } } + +TEST_F(MistralOutputParserTest, StreamingToolCallArgumentsBeforeNameThrows) { + std::vector>> chunkToDeltaVec{ + {"[{\"", ov::genai::GenerationFinishReason::NONE, std::nullopt}, + {"arguments", ov::genai::GenerationFinishReason::NONE, std::nullopt}, + // At this point "arguments" key has just appeared in the accumulated JSON, but "name" was never provided. + {"\":", ov::genai::GenerationFinishReason::NONE, std::nullopt}, + }; + + for (const auto& [chunk, finishReason, expectedDelta] : chunkToDeltaVec) { + if (expectedDelta.has_value()) { + FAIL() << "Expected delta should be nullopt for this test case."; + } + try { + std::optional doc = outputParserWithRegularToolParsing->parseChunk(chunk, {}, true, finishReason); + if (doc.has_value()) { + FAIL() << "Expected an exception to be thrown for chunk: " << chunk; + } + } catch (const std::runtime_error& e) { + EXPECT_STREQ(e.what(), "Tool call name is missing in generated output"); + } catch (...) { + FAIL() << "Expected a std::runtime_error to be thrown for chunk: " << chunk; + } + } +}