mirror of
https://gitcode.com/JianFeeeee/TrulyMEM-TrueHumanMEM.git
synced 2026-10-15 12:15:32 +00:00
Fix: Properly handle tool calls with results returned to API
This commit is contained in:
@ -1,8 +1,9 @@
|
|||||||
"""聊天服务"""
|
"""聊天服务"""
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import json
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import AsyncIterator, TYPE_CHECKING
|
from typing import AsyncIterator, TYPE_CHECKING, List, Dict, Any
|
||||||
from ..core.imports import GraphMemoryClient
|
from ..core.imports import GraphMemoryClient
|
||||||
from ..models.message import ToolCall, ToolResult
|
from ..models.message import ToolCall, ToolResult
|
||||||
from .tool_service import ToolService
|
from .tool_service import ToolService
|
||||||
@ -23,7 +24,7 @@ class ChatService:
|
|||||||
self._graph = graph
|
self._graph = graph
|
||||||
self._client = client
|
self._client = client
|
||||||
self._tool_service = tool_service
|
self._tool_service = tool_service
|
||||||
self._messages: list[dict] = []
|
self._messages: List[Dict[str, Any]] = []
|
||||||
|
|
||||||
async def send_message(self, user_input: str) -> AsyncIterator[dict]:
|
async def send_message(self, user_input: str) -> AsyncIterator[dict]:
|
||||||
"""发送消息并流式返回事件"""
|
"""发送消息并流式返回事件"""
|
||||||
@ -34,13 +35,12 @@ class ChatService:
|
|||||||
}
|
}
|
||||||
|
|
||||||
try:
|
try:
|
||||||
# 2. 使用流式API调用
|
# 2. 第一次API调用
|
||||||
accumulated_content = ""
|
accumulated_content = ""
|
||||||
tool_calls_data = []
|
tool_calls_data = []
|
||||||
|
|
||||||
# 流式处理响应
|
# 流式处理响应
|
||||||
async for chunk in self._call_api_stream_async(user_input):
|
async for chunk in self._call_api_stream_async(user_input):
|
||||||
# 处理内容增量
|
|
||||||
if chunk.get("content_delta"):
|
if chunk.get("content_delta"):
|
||||||
accumulated_content += chunk["content_delta"]
|
accumulated_content += chunk["content_delta"]
|
||||||
yield {
|
yield {
|
||||||
@ -48,11 +48,10 @@ class ChatService:
|
|||||||
"content": accumulated_content
|
"content": accumulated_content
|
||||||
}
|
}
|
||||||
|
|
||||||
# 处理工具调用
|
|
||||||
if chunk.get("tool_calls"):
|
if chunk.get("tool_calls"):
|
||||||
tool_calls_data = chunk["tool_calls"]
|
tool_calls_data = chunk["tool_calls"]
|
||||||
|
|
||||||
# 3. 处理工具调用
|
# 3. 如果有工具调用,执行并继续调用API
|
||||||
tool_calls = None
|
tool_calls = None
|
||||||
tool_results = None
|
tool_results = None
|
||||||
|
|
||||||
@ -60,26 +59,23 @@ class ChatService:
|
|||||||
tool_calls = []
|
tool_calls = []
|
||||||
tool_results = []
|
tool_results = []
|
||||||
|
|
||||||
|
# 执行所有工具
|
||||||
for tool_call_data in tool_calls_data:
|
for tool_call_data in tool_calls_data:
|
||||||
# 创建工具调用对象
|
|
||||||
tool_call = ToolCall(
|
tool_call = ToolCall(
|
||||||
id=tool_call_data.id,
|
id=tool_call_data["id"],
|
||||||
name=tool_call_data.function.name,
|
name=tool_call_data["function"]["name"],
|
||||||
arguments=tool_call_data.function.arguments
|
arguments=tool_call_data["function"]["arguments"]
|
||||||
)
|
)
|
||||||
tool_calls.append(tool_call)
|
tool_calls.append(tool_call)
|
||||||
|
|
||||||
# 发送工具调用事件
|
|
||||||
yield {
|
yield {
|
||||||
"type": "tool_call",
|
"type": "tool_call",
|
||||||
"tool_call": tool_call
|
"tool_call": tool_call
|
||||||
}
|
}
|
||||||
|
|
||||||
# 执行工具
|
|
||||||
result = await self._tool_service.execute(tool_call)
|
result = await self._tool_service.execute(tool_call)
|
||||||
tool_results.append(result)
|
tool_results.append(result)
|
||||||
|
|
||||||
# 发送工具结果事件
|
|
||||||
log_entry = ToolService._create_log_entry(tool_call, result)
|
log_entry = ToolService._create_log_entry(tool_call, result)
|
||||||
yield {
|
yield {
|
||||||
"type": "tool_result",
|
"type": "tool_result",
|
||||||
@ -87,6 +83,47 @@ class ChatService:
|
|||||||
"log_entry": log_entry
|
"log_entry": log_entry
|
||||||
}
|
}
|
||||||
|
|
||||||
|
# 构建工具结果消息
|
||||||
|
tool_messages = []
|
||||||
|
for tc, tr in zip(tool_calls, tool_results):
|
||||||
|
tool_messages.append({
|
||||||
|
"role": "tool",
|
||||||
|
"tool_call_id": tc.id,
|
||||||
|
"content": tr.content
|
||||||
|
})
|
||||||
|
|
||||||
|
# 构建assistant消息(包含tool_calls)
|
||||||
|
assistant_message = {
|
||||||
|
"role": "assistant",
|
||||||
|
"content": accumulated_content,
|
||||||
|
"tool_calls": [
|
||||||
|
{
|
||||||
|
"id": tc.id,
|
||||||
|
"type": "function",
|
||||||
|
"function": {
|
||||||
|
"name": tc.name,
|
||||||
|
"arguments": tc.arguments
|
||||||
|
}
|
||||||
|
} for tc in tool_calls
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
# 第二次API调用,传入工具结果
|
||||||
|
final_content = ""
|
||||||
|
async for chunk in self._call_api_stream_with_tools(
|
||||||
|
user_input,
|
||||||
|
assistant_message,
|
||||||
|
tool_messages
|
||||||
|
):
|
||||||
|
if chunk.get("content_delta"):
|
||||||
|
final_content += chunk["content_delta"]
|
||||||
|
yield {
|
||||||
|
"type": "content_delta",
|
||||||
|
"content": final_content
|
||||||
|
}
|
||||||
|
|
||||||
|
accumulated_content = final_content
|
||||||
|
|
||||||
# 4. 返回最终回复
|
# 4. 返回最终回复
|
||||||
yield {
|
yield {
|
||||||
"type": "assistant_message",
|
"type": "assistant_message",
|
||||||
@ -96,7 +133,6 @@ class ChatService:
|
|||||||
}
|
}
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
# 错误处理
|
|
||||||
yield {
|
yield {
|
||||||
"type": "error",
|
"type": "error",
|
||||||
"error": str(e)
|
"error": str(e)
|
||||||
@ -106,7 +142,6 @@ class ChatService:
|
|||||||
"""异步流式调用 API"""
|
"""异步流式调用 API"""
|
||||||
loop = asyncio.get_event_loop()
|
loop = asyncio.get_event_loop()
|
||||||
|
|
||||||
# 在executor中运行同步流式API
|
|
||||||
def process_stream():
|
def process_stream():
|
||||||
stream = self._client.send_message_stream(message)
|
stream = self._client.send_message_stream(message)
|
||||||
tool_calls_accumulated = []
|
tool_calls_accumulated = []
|
||||||
@ -114,14 +149,11 @@ class ChatService:
|
|||||||
for chunk in stream:
|
for chunk in stream:
|
||||||
delta = chunk.choices[0].delta
|
delta = chunk.choices[0].delta
|
||||||
|
|
||||||
# 处理内容增量
|
|
||||||
if delta.content:
|
if delta.content:
|
||||||
yield {"content_delta": delta.content}
|
yield {"content_delta": delta.content}
|
||||||
|
|
||||||
# 处理工具调用
|
|
||||||
if delta.tool_calls:
|
if delta.tool_calls:
|
||||||
for tc in delta.tool_calls:
|
for tc in delta.tool_calls:
|
||||||
# 累积工具调用数据
|
|
||||||
if tc.index >= len(tool_calls_accumulated):
|
if tc.index >= len(tool_calls_accumulated):
|
||||||
tool_calls_accumulated.append({
|
tool_calls_accumulated.append({
|
||||||
"id": tc.id,
|
"id": tc.id,
|
||||||
@ -138,11 +170,41 @@ class ChatService:
|
|||||||
if tc.function.arguments:
|
if tc.function.arguments:
|
||||||
tool_calls_accumulated[tc.index]["function"]["arguments"] += tc.function.arguments
|
tool_calls_accumulated[tc.index]["function"]["arguments"] += tc.function.arguments
|
||||||
|
|
||||||
# 返回完整的工具调用
|
|
||||||
if tool_calls_accumulated:
|
if tool_calls_accumulated:
|
||||||
yield {"tool_calls": tool_calls_accumulated}
|
yield {"tool_calls": tool_calls_accumulated}
|
||||||
|
|
||||||
# 使用run_in_executor处理生成器
|
for result in await loop.run_in_executor(None, lambda: list(process_stream())):
|
||||||
|
yield result
|
||||||
|
|
||||||
|
async def _call_api_stream_with_tools(
|
||||||
|
self,
|
||||||
|
user_input: str,
|
||||||
|
assistant_message: dict,
|
||||||
|
tool_messages: list
|
||||||
|
) -> AsyncIterator[dict]:
|
||||||
|
"""带工具结果的流式调用"""
|
||||||
|
loop = asyncio.get_event_loop()
|
||||||
|
|
||||||
|
def process_stream():
|
||||||
|
# 构建完整的消息列表
|
||||||
|
messages = [
|
||||||
|
{"role": "user", "content": user_input},
|
||||||
|
assistant_message
|
||||||
|
]
|
||||||
|
messages.extend(tool_messages)
|
||||||
|
|
||||||
|
# 调用API
|
||||||
|
response = self._client.client.chat.completions.create(
|
||||||
|
model=self._client.tools[0]["function"]["name"] if self._client.tools else "deepseek-chat",
|
||||||
|
messages=messages,
|
||||||
|
stream=True
|
||||||
|
)
|
||||||
|
|
||||||
|
for chunk in response:
|
||||||
|
delta = chunk.choices[0].delta
|
||||||
|
if delta.content:
|
||||||
|
yield {"content_delta": delta.content}
|
||||||
|
|
||||||
for result in await loop.run_in_executor(None, lambda: list(process_stream())):
|
for result in await loop.run_in_executor(None, lambda: list(process_stream())):
|
||||||
yield result
|
yield result
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user