主要改进: - 新增 convert_tool_calls_to_content 函数,将消息历史中的 tool_calls 转换为 LLM 可理解的 XML 格式 - 修复 response_parser 支持同时解析多个 tool_calls - 优化响应解析逻辑,支持 content 和 tool_calls 同时存在 - 添加完整的测试覆盖,包括多工具调用、消息转换和混合响应 技术细节: - services.py: 实现工具调用历史到 content 的转换 - response_parser.py: 使用非贪婪匹配支持多个 tool_calls 解析 - main.py: 集成消息转换功能,确保消息历史正确传递给 LLM 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
156 lines
5.0 KiB
Python
156 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
测试 chat 接口同时返回文本内容和 tool_calls
|
|
"""
|
|
import sys
|
|
import os
|
|
import json
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from app.response_parser import ResponseParser
|
|
from app.models import ResponseMessage, ToolCall, ToolCallFunction
|
|
|
|
def test_content_and_tool_calls():
|
|
"""测试同时返回文本内容和 tool_calls 的各种场景"""
|
|
|
|
parser = ResponseParser()
|
|
|
|
print("=" * 70)
|
|
print("测试:同时返回文本内容和 tool_calls")
|
|
print("=" * 70)
|
|
|
|
# 场景 1: 文本在前 + tool_calls
|
|
print("\n场景 1: 先说话,再调用工具")
|
|
print("-" * 70)
|
|
|
|
text1 = """好的,我来帮你查询北京的天气情况。
|
|
|
|
<invoke>{"name": "get_weather", "arguments": {"location": "北京", "unit": "celsius"}}</invoke>"""
|
|
|
|
result1 = parser.parse(text1)
|
|
|
|
print(f"输入文本:\n{text1}\n")
|
|
print(f"解析结果:")
|
|
print(f" - content: {result1.content}")
|
|
print(f" - tool_calls: {len(result1.tool_calls) if result1.tool_calls else 0} 个")
|
|
if result1.tool_calls:
|
|
for tc in result1.tool_calls:
|
|
print(f" * {tc.function.name}: {tc.function.arguments}")
|
|
|
|
# 验证
|
|
assert result1.content is not None, "Content should not be None"
|
|
assert result1.tool_calls is not None, "Tool calls should not be None"
|
|
assert len(result1.tool_calls) == 1, "Should have 1 tool call"
|
|
assert "北京" in result1.content or "查询" in result1.content, "Content should contain original text"
|
|
|
|
print(" ✓ 场景 1 通过")
|
|
|
|
# 场景 2: tool_calls + 文本在后
|
|
print("\n场景 2: 先调用工具,再说话")
|
|
print("-" * 70)
|
|
|
|
text2 = """<invoke>{"name": "search", "arguments": {"query": "今天天气"}}</invoke>
|
|
|
|
我已经帮你查询了,请稍等片刻。"""
|
|
|
|
result2 = parser.parse(text2)
|
|
|
|
print(f"输入文本:\n{text2}\n")
|
|
print(f"解析结果:")
|
|
print(f" - content: {result2.content}")
|
|
print(f" - tool_calls: {len(result2.tool_calls) if result2.tool_calls else 0} 个")
|
|
if result2.tool_calls:
|
|
for tc in result2.tool_calls:
|
|
print(f" * {tc.function.name}: {tc.function.arguments}")
|
|
|
|
assert result2.content is not None
|
|
assert result2.tool_calls is not None
|
|
assert "稍等" in result2.content or "查询" in result2.content
|
|
|
|
print(" ✓ 场景 2 通过")
|
|
|
|
# 场景 3: 文本 - tool_calls - 文本
|
|
print("\n场景 3: 文本 - 工具调用 - 文本(三明治结构)")
|
|
print("-" * 70)
|
|
|
|
text3 = """让我先查一下北京的温度。
|
|
|
|
<invoke>{"name": "get_weather", "arguments": {"location": "北京"}}</invoke>
|
|
|
|
查到了,我再查一下上海的。
|
|
|
|
<invoke>{"name": "get_weather", "arguments": {"location": "上海"}}</invoke>
|
|
|
|
好了,两个城市都查询完毕。"""
|
|
|
|
result3 = parser.parse(text3)
|
|
|
|
print(f"输入文本:\n{text3}\n")
|
|
print(f"解析结果:")
|
|
print(f" - content: {result3.content}")
|
|
print(f" - tool_calls: {len(result3.tool_calls) if result3.tool_calls else 0} 个")
|
|
if result3.tool_calls:
|
|
for i, tc in enumerate(result3.tool_calls, 1):
|
|
print(f" * {tc.function.name}: {tc.function.arguments}")
|
|
|
|
assert result3.content is not None
|
|
assert result3.tool_calls is not None
|
|
assert len(result3.tool_calls) == 2
|
|
assert "先查一下" in result3.content
|
|
assert "查询完毕" in result3.content
|
|
|
|
print(" ✓ 场景 3 通过")
|
|
|
|
# 场景 4: 测试 ResponseMessage 序列化
|
|
print("\n场景 4: 验证 ResponseMessage 可以正确序列化为 JSON")
|
|
print("-" * 70)
|
|
|
|
msg = ResponseMessage(
|
|
role="assistant",
|
|
content="好的,我来帮你查询。",
|
|
tool_calls=[
|
|
ToolCall(
|
|
id="call_123",
|
|
type="function",
|
|
function=ToolCallFunction(
|
|
name="get_weather",
|
|
arguments=json.dumps({"location": "北京"})
|
|
)
|
|
)
|
|
]
|
|
)
|
|
|
|
json_str = msg.model_dump_json(indent=2)
|
|
print("序列化的 JSON 响应:")
|
|
print(json_str)
|
|
|
|
parsed_back = ResponseMessage.model_validate_json(json_str)
|
|
assert parsed_back.content == msg.content
|
|
assert parsed_back.tool_calls is not None
|
|
assert len(parsed_back.tool_calls) == 1
|
|
|
|
print(" ✓ 场景 4 通过 - JSON 序列化/反序列化正常")
|
|
|
|
print("\n" + "=" * 70)
|
|
print("所有测试通过! ✓")
|
|
print("=" * 70)
|
|
|
|
print("\n总结:")
|
|
print("✓ chat 接口支持同时返回文本内容和 tool_calls")
|
|
print("✓ content 和 tool_calls 可以同时存在")
|
|
print("✓ 支持文本在前、在后、或前后都有文本的场景")
|
|
print("✓ 支持多个 tool_calls 与文本内容混合")
|
|
print("✓ JSON 序列化/反序列化正常")
|
|
|
|
print("\n实际应用场景示例:")
|
|
print("""
|
|
Assistant: "好的,我来帮你查询一下。"
|
|
[调用 get_weather 工具]
|
|
[收到工具结果]
|
|
Assistant: "北京今天晴天,气温 25°C。"
|
|
""")
|
|
|
|
if __name__ == "__main__":
|
|
test_content_and_tool_calls()
|