diff --git a/application/agents/base.py b/application/agents/base.py index b8efeb24..22b808d9 100644 --- a/application/agents/base.py +++ b/application/agents/base.py @@ -136,6 +136,7 @@ class BaseAgent(ABC): tools_dict: Dict, pending_tool_calls: List[Dict], tool_actions: List[Dict], + reasoning_content: str = "", ) -> Generator[Dict, None, None]: """Resume generation after tool actions are resolved. @@ -175,11 +176,14 @@ class BaseAgent(ABC): tc_obj["thought_signature"] = pending["thought_signature"] tc_objects.append(tc_obj) - messages.append({ + resumed_assistant: Dict[str, Any] = { "role": "assistant", "content": None, "tool_calls": tc_objects, - }) + } + if reasoning_content: + resumed_assistant["reasoning_content"] = reasoning_content + messages.append(resumed_assistant) # Now process each pending call and append tool result messages for pending in pending_tool_calls: @@ -433,7 +437,17 @@ class BaseAgent(ABC): for i in working_history: if "prompt" in i and "response" in i: messages.append({"role": "user", "content": i["prompt"]}) - messages.append({"role": "assistant", "content": i["response"]}) + asst_msg: Dict[str, Any] = { + "role": "assistant", + "content": i["response"], + } + # Persisted thought from the prior turn rides along as + # reasoning_content so providers that require it on the + # follow-up call (DeepSeek thinking mode) accept the + # request. Other OpenAI-compatible APIs ignore the field. + if i.get("thought"): + asst_msg["reasoning_content"] = i["thought"] + messages.append(asst_msg) if "tool_calls" in i: for tool_call in i["tool_calls"]: call_id = tool_call.get("call_id") or str(uuid.uuid4()) diff --git a/application/api/answer/routes/answer.py b/application/api/answer/routes/answer.py index 78748e29..f1d25a1c 100644 --- a/application/api/answer/routes/answer.py +++ b/application/api/answer/routes/answer.py @@ -82,6 +82,7 @@ class AnswerResource(Resource, BaseAnswerResource): tools_dict, pending_tool_calls, tool_actions, + reasoning_content, ) = processor.resume_from_tool_actions( data["tool_actions"], data["conversation_id"] ) @@ -104,6 +105,7 @@ class AnswerResource(Resource, BaseAnswerResource): "tool_actions": tool_actions, "reserved_message_id": processor.reserved_message_id, "request_id": processor.request_id, + "reasoning_content": reasoning_content, }, ) else: diff --git a/application/api/answer/routes/base.py b/application/api/answer/routes/base.py index c3aaefd7..32852edf 100644 --- a/application/api/answer/routes/base.py +++ b/application/api/answer/routes/base.py @@ -415,6 +415,7 @@ class BaseAnswerResource: tools_dict=_continuation["tools_dict"], pending_tool_calls=_continuation["pending_tool_calls"], tool_actions=_continuation["tool_actions"], + reasoning_content=_continuation.get("reasoning_content", ""), ) else: gen_iter = agent.gen(query=question) @@ -567,6 +568,14 @@ class BaseAnswerResource: # consistent across token_usage rows. "reserved_message_id": reserved_message_id, "request_id": request_id, + # Persisted in agent_config (rather than + # a new column) so resume rebuilds the + # paused assistant message with the + # reasoning DeepSeek thinking mode + # requires on the follow-up turn. + "reasoning_content": continuation.get( + "reasoning_content", "" + ), }, client_tools=getattr( agent.tool_executor, "client_tools", None @@ -1055,7 +1064,7 @@ class BaseAnswerResource: "pending_tool_calls", [] ) elif event["type"] == "thought": - thought = event["thought"] + thought += event["thought"] elif event["type"] == "error": logger.error(f"Error from stream: {event['error']}") return { diff --git a/application/api/answer/routes/stream.py b/application/api/answer/routes/stream.py index 5410d697..1e7c4c69 100644 --- a/application/api/answer/routes/stream.py +++ b/application/api/answer/routes/stream.py @@ -89,6 +89,7 @@ class StreamResource(Resource, BaseAnswerResource): tools_dict, pending_tool_calls, tool_actions, + reasoning_content, ) = processor.resume_from_tool_actions( data["tool_actions"], data["conversation_id"] ) @@ -117,6 +118,7 @@ class StreamResource(Resource, BaseAnswerResource): "tool_actions": tool_actions, "reserved_message_id": processor.reserved_message_id, "request_id": processor.request_id, + "reasoning_content": reasoning_content, }, ), mimetype="text/event-stream", diff --git a/application/api/answer/services/compression/types.py b/application/api/answer/services/compression/types.py index b71ab9ee..7bdba590 100644 --- a/application/api/answer/services/compression/types.py +++ b/application/api/answer/services/compression/types.py @@ -75,9 +75,17 @@ class CompressionResult: Convert recent queries to history format. Returns: - List of prompt/response dicts + List of prompt/response dicts (with thought when present so + DeepSeek-style providers can re-attach reasoning_content on + replay). """ - return [ - {"prompt": q["prompt"], "response": q["response"]} - for q in self.recent_queries - ] + out: List[Dict[str, str]] = [] + for q in self.recent_queries: + entry: Dict[str, str] = { + "prompt": q["prompt"], + "response": q["response"], + } + if q.get("thought"): + entry["thought"] = q["thought"] + out.append(entry) + return out diff --git a/application/api/answer/services/stream_processor.py b/application/api/answer/services/stream_processor.py index e6bb7549..fd212205 100644 --- a/application/api/answer/services/stream_processor.py +++ b/application/api/answer/services/stream_processor.py @@ -190,6 +190,15 @@ class StreamProcessor: { "prompt": query["prompt"], "response": query["response"], + # Carry the persisted thought so _build_messages + # re-attaches it as reasoning_content on replay — + # DeepSeek thinking mode rejects follow-up turns + # whose prior assistant message dropped it. + **( + {"thought": query["thought"]} + if query.get("thought") + else {} + ), **( {"metadata": query["metadata"]} if "metadata" in query @@ -226,6 +235,11 @@ class StreamProcessor: { "prompt": query["prompt"], "response": query["response"], + **( + {"thought": query["thought"]} + if query.get("thought") + else {} + ), **({"metadata": query["metadata"]} if "metadata" in query else {}), } for query in conversation.get("queries", []) @@ -261,6 +275,11 @@ class StreamProcessor: { "prompt": query["prompt"], "response": query["response"], + **( + {"thought": query["thought"]} + if query.get("thought") + else {} + ), **({"metadata": query["metadata"]} if "metadata" in query else {}), } for query in conversation.get("queries", []) @@ -971,7 +990,11 @@ class StreamProcessor: conversation_id: The conversation being resumed. Returns: - Tuple of (agent, messages, tools_dict, pending_tool_calls, tool_actions). + Tuple of (agent, messages, tools_dict, pending_tool_calls, + tool_actions, reasoning_content). ``reasoning_content`` is + the reasoning text emitted before the pause; round-tripping + it back to the model is required by DeepSeek's thinking + mode and ignored elsewhere. """ from application.api.answer.services.continuation_service import ( ContinuationService, @@ -1100,7 +1123,15 @@ class StreamProcessor: self.reserved_message_id = agent_config.get("reserved_message_id") self.request_id = agent_config.get("request_id") - return agent, messages, tools_dict, pending_tool_calls, tool_actions + reasoning_content = agent_config.get("reasoning_content", "") + return ( + agent, + messages, + tools_dict, + pending_tool_calls, + tool_actions, + reasoning_content, + ) def create_agent( self, diff --git a/application/api/v1/routes.py b/application/api/v1/routes.py index 0c862b17..268aa12b 100644 --- a/application/api/v1/routes.py +++ b/application/api/v1/routes.py @@ -117,6 +117,7 @@ def chat_completions(): tools_dict, pending_tool_calls, tool_actions, + reasoning_content, ) = processor.resume_from_tool_actions( internal_data["tool_actions"], conversation_id ) @@ -125,6 +126,7 @@ def chat_completions(): "tools_dict": tools_dict, "pending_tool_calls": pending_tool_calls, "tool_actions": tool_actions, + "reasoning_content": reasoning_content, } question = "" else: diff --git a/application/llm/google_ai.py b/application/llm/google_ai.py index d8231b22..aca099c5 100644 --- a/application/llm/google_ai.py +++ b/application/llm/google_ai.py @@ -549,7 +549,9 @@ class GoogleLLM(BaseLLM): system_instruction = None if formatting == "openai": messages, system_instruction = self._clean_messages_google(messages) - config = types.GenerateContentConfig() + config = types.GenerateContentConfig( + thinking_config=types.ThinkingConfig(include_thoughts=True), + ) if system_instruction: config.system_instruction = system_instruction if tools: @@ -587,7 +589,13 @@ class GoogleLLM(BaseLLM): system_instruction = None if formatting == "openai": messages, system_instruction = self._clean_messages_google(messages) - config = types.GenerateContentConfig() + # include_thoughts surfaces Gemini's thought-summary parts so the + # same {"type":"thought"} accumulator that DeepSeek uses can + # capture and persist them. Off by default; thinking itself is + # already on for Gemini 3.x flash/pro. + config = types.GenerateContentConfig( + thinking_config=types.ThinkingConfig(include_thoughts=True), + ) if system_instruction: config.system_instruction = system_instruction if tools: diff --git a/application/llm/handlers/base.py b/application/llm/handlers/base.py index bfe8f7d3..45959e21 100644 --- a/application/llm/handlers/base.py +++ b/application/llm/handlers/base.py @@ -51,6 +51,7 @@ class LLMResponse: tool_calls: List[ToolCall] finish_reason: str raw_response: Any + reasoning_content: str = "" @property def requires_tool_call(self) -> bool: @@ -714,7 +715,12 @@ class LLMHandler(ABC): return False, None def handle_tool_calls( - self, agent, tool_calls: List[ToolCall], tools_dict: Dict, messages: List[Dict] + self, + agent, + tool_calls: List[ToolCall], + tools_dict: Dict, + messages: List[Dict], + reasoning_content: str = "", ) -> Generator: """ Execute tool calls and update conversation history. @@ -731,6 +737,11 @@ class LLMHandler(ABC): tool_calls: List of tool calls to execute tools_dict: Available tools dictionary messages: Current conversation history + reasoning_content: Reasoning text emitted by the model + before these tool calls. Attached to the recorded + assistant message so providers that require reasoning + to round-trip (DeepSeek thinking mode) accept the + follow-up request. Returns: Tuple of (updated_messages, pending_actions). @@ -872,11 +883,14 @@ class LLMHandler(ABC): } if getattr(call, "thought_signature", None): tool_call_obj["thought_signature"] = call.thought_signature - updated_messages.append({ + assistant_msg: Dict[str, Any] = { "role": "assistant", "content": None, "tool_calls": [tool_call_obj], - }) + } + if reasoning_content: + assistant_msg["reasoning_content"] = reasoning_content + updated_messages.append(assistant_msg) denial_call = ToolCall( id=pause_info["call_id"], name=call.name, @@ -967,11 +981,19 @@ class LLMHandler(ABC): if call.thought_signature: tool_call_obj["thought_signature"] = call.thought_signature - updated_messages.append({ + assistant_msg: Dict[str, Any] = { "role": "assistant", "content": None, "tool_calls": [tool_call_obj], - }) + } + # Each call in a parallel batch becomes its own + # assistant message here, so the same per-round + # reasoning has to ride on every one — DeepSeek + # thinking mode rejects any assistant message in the + # active turn that's missing reasoning_content. + if reasoning_content: + assistant_msg["reasoning_content"] = reasoning_content + updated_messages.append(assistant_msg) updated_messages.append(self.create_tool_message(call, tool_response)) except Exception as e: @@ -1026,8 +1048,13 @@ class LLMHandler(ABC): iteration = 0 while parsed.requires_tool_call: iteration += 1 + reasoning_for_round = parsed.reasoning_content or "" tool_handler_gen = self.handle_tool_calls( - agent, parsed.tool_calls, tools_dict, messages + agent, + parsed.tool_calls, + tools_dict, + messages, + reasoning_content=reasoning_for_round, ) while True: try: @@ -1042,6 +1069,7 @@ class LLMHandler(ABC): "messages": messages, "pending_tool_calls": pending_actions, "tools_dict": tools_dict, + "reasoning_content": reasoning_for_round, } yield { "type": "tool_calls_pending", @@ -1104,15 +1132,19 @@ class LLMHandler(ABC): """ buffer = "" tool_calls = {} + reasoning_buffer = "" for chunk in self._iterate_stream(response): if isinstance(chunk, dict) and chunk.get("type") == "thought": + reasoning_buffer += chunk.get("thought") or "" yield chunk continue if isinstance(chunk, str): yield chunk continue parsed = self.parse_response(chunk) + if parsed.reasoning_content: + reasoning_buffer += parsed.reasoning_content if parsed.tool_calls: for call in parsed.tool_calls: @@ -1134,7 +1166,11 @@ class LLMHandler(ABC): existing.thought_signature = call.thought_signature if parsed.finish_reason == "tool_calls": tool_handler_gen = self.handle_tool_calls( - agent, list(tool_calls.values()), tools_dict, messages + agent, + list(tool_calls.values()), + tools_dict, + messages, + reasoning_content=reasoning_buffer, ) while True: try: @@ -1143,6 +1179,8 @@ class LLMHandler(ABC): messages, pending_actions = e.value break tool_calls = {} + pause_reasoning = reasoning_buffer + reasoning_buffer = "" # If tools need approval or client execution, pause the loop if pending_actions: @@ -1150,6 +1188,7 @@ class LLMHandler(ABC): "messages": messages, "pending_tool_calls": pending_actions, "tools_dict": tools_dict, + "reasoning_content": pause_reasoning, } yield { "type": "tool_calls_pending", diff --git a/application/llm/handlers/openai.py b/application/llm/handlers/openai.py index a7b2628c..9236b1be 100644 --- a/application/llm/handlers/openai.py +++ b/application/llm/handlers/openai.py @@ -1,6 +1,7 @@ from typing import Any, Dict, Generator from application.llm.handlers.base import LLMHandler, LLMResponse, ToolCall +from application.llm.openai import OpenAILLM class OpenAILLMHandler(LLMHandler): @@ -29,11 +30,16 @@ class OpenAILLMHandler(LLMHandler): ) for tc in message.tool_calls or [] ] + # Reasoning lives on the message object for non-streaming and + # on the delta for streaming. DeepSeek thinking mode requires + # this to be echoed back on the next turn. + reasoning_content = OpenAILLM._extract_reasoning_text(message) return LLMResponse( content=getattr(message, "content", ""), tool_calls=tool_calls, finish_reason=getattr(response, "finish_reason", ""), raw_response=response, + reasoning_content=reasoning_content, ) def create_tool_message(self, tool_call: ToolCall, result: Any) -> Dict: diff --git a/application/llm/openai.py b/application/llm/openai.py index 23d560fd..69228d01 100644 --- a/application/llm/openai.py +++ b/application/llm/openai.py @@ -108,6 +108,10 @@ class OpenAILLM(BaseLLM): for message in messages: role = message.get("role") content = message.get("content") + # Reasoning round-trips for providers that demand it + # (DeepSeek thinking mode). Other OpenAI-compatible APIs + # ignore the extra field. + reasoning_content = message.get("reasoning_content") if role == "model": role = "assistant" @@ -132,11 +136,14 @@ class OpenAILLM(BaseLLM): "type": "function", "function": {"name": func.get("name", ""), "arguments": args}, }) - cleaned_messages.append({ + cleaned_assistant: dict = { "role": "assistant", "content": None, "tool_calls": cleaned_tcs, - }) + } + if reasoning_content: + cleaned_assistant["reasoning_content"] = reasoning_content + cleaned_messages.append(cleaned_assistant) continue # Standard format: tool message with tool_call_id (passthrough) @@ -151,7 +158,10 @@ class OpenAILLM(BaseLLM): if role and content is not None: if isinstance(content, str): - cleaned_messages.append({"role": role, "content": content}) + msg_obj: dict = {"role": role, "content": content} + if reasoning_content and role == "assistant": + msg_obj["reasoning_content"] = reasoning_content + cleaned_messages.append(msg_obj) elif isinstance(content, list): content_parts = [] for item in content: @@ -195,7 +205,10 @@ class OpenAILLM(BaseLLM): elif "text" in item and "type" not in item: content_parts.append({"type": "text", "text": item["text"]}) if content_parts: - cleaned_messages.append({"role": role, "content": content_parts}) + list_msg: dict = {"role": role, "content": content_parts} + if reasoning_content and role == "assistant": + list_msg["reasoning_content"] = reasoning_content + cleaned_messages.append(list_msg) else: raise ValueError(f"Unexpected content type: {type(content)}") return cleaned_messages diff --git a/tests/llm/handlers/test_llm_handlers.py b/tests/llm/handlers/test_llm_handlers.py index 38f7a65c..821eec00 100644 --- a/tests/llm/handlers/test_llm_handlers.py +++ b/tests/llm/handlers/test_llm_handlers.py @@ -963,7 +963,9 @@ class TestHandleStreaming: mock_settings.ENABLE_CONVERSATION_COMPRESSION = False # handle_tool_calls yields skip events and sets context_limit_reached - def fake_handle_tool_calls(agent, calls, tools_dict, messages): + def fake_handle_tool_calls( + agent, calls, tools_dict, messages, reasoning_content="" + ): agent.context_limit_reached = True yield {"type": "tool_call", "data": {"status": "skipped"}} return messages, None diff --git a/tests/llm/test_google_ai.py b/tests/llm/test_google_ai.py index 7d91069c..76e01ec2 100644 --- a/tests/llm/test_google_ai.py +++ b/tests/llm/test_google_ai.py @@ -72,13 +72,18 @@ class FakeTypesModule: Content = _FakeContent class GenerateContentConfig: - def __init__(self): + def __init__(self, thinking_config=None, **_kw): self.system_instruction = None self.tools = None - self.thinking_config = None + self.thinking_config = thinking_config self.response_schema = None self.response_mime_type = None + class ThinkingConfig: + def __init__(self, include_thoughts=False, thinking_level=None): + self.include_thoughts = include_thoughts + self.thinking_level = thinking_level + class Tool: def __init__(self, function_declarations=None): self.function_declarations = function_declarations or [] diff --git a/tests/llm/test_google_llm.py b/tests/llm/test_google_llm.py index b5555047..2a65f399 100644 --- a/tests/llm/test_google_llm.py +++ b/tests/llm/test_google_llm.py @@ -51,10 +51,10 @@ class FakeTypesModule: self.thinking_level = thinking_level class GenerateContentConfig: - def __init__(self): + def __init__(self, thinking_config=None, **_kw): self.system_instruction = None self.tools = None - self.thinking_config = None + self.thinking_config = thinking_config self.response_schema = None self.response_mime_type = None @@ -125,7 +125,9 @@ def test_raw_gen_stream_yields_chunks(): assert list(gen) == ["a", "b"] -def test_raw_gen_stream_does_not_set_thinking_config_by_default(monkeypatch): +def test_raw_gen_stream_opts_into_thought_summaries(monkeypatch): + """Gemini won't surface thought parts unless include_thoughts is + set on thinking_config — see google_ai.py:_raw_gen_stream.""" captured = {} def fake_stream(self, *args, **kwargs): @@ -138,7 +140,8 @@ def test_raw_gen_stream_does_not_set_thinking_config_by_default(monkeypatch): msgs = [{"role": "user", "content": "hello"}] list(llm._raw_gen_stream(llm, model="gemini", messages=msgs, stream=True)) - assert captured["config"].thinking_config is None + assert captured["config"].thinking_config is not None + assert captured["config"].thinking_config.include_thoughts is True def test_raw_gen_stream_emits_thought_events(monkeypatch):