mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-13 04:14:42 +00:00
Deepseek (#2499)
* feat: add deepseek * feat: deepseek reasoning tool fix
This commit is contained in:
14 files changed
+175
-31
No files matched your search
@@ -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())
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
@@ -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,
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 []
|
||||
|
||||
@@ -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):
|
||||
|
||||
Reference in new issue
Block a user