diff --git a/test_unified_format.py b/test_unified_format.py deleted file mode 100644 index 59cdda6..0000000 --- a/test_unified_format.py +++ /dev/null @@ -1,155 +0,0 @@ -#!/usr/bin/env python3 -""" -测试统一使用OpenAI格式的一致性 -""" - -import json -from fastapi_app import convert_messages_to_openai_format -from openai_converter import OpenAIConverter - - -def test_unified_openai_format(): - """测试统一使用OpenAI格式""" - print("=== 测试统一OpenAI格式 ===") - - # 测试数据 - test_messages = [ - { - "role": "assistant", - "reasoning_content": "这是思考过程", - "content": "这是回答内容", - "function_call": { - "name": "test_function", - "arguments": '{"param1": "value1"}' - } - }, - { - "role": "function", - "name": "test_function", - "content": "工具执行结果" - } - ] - - print("原始qwen-agent格式:") - print(json.dumps(test_messages, indent=2, ensure_ascii=False)) - - # 使用新的统一函数 - openai_messages = convert_messages_to_openai_format(test_messages) - - print("\n转换后的OpenAI格式:") - print(json.dumps(openai_messages, indent=2, ensure_ascii=False)) - - # 验证格式 - assert len(openai_messages) == 2 - - # 验证第一个消息 - first_msg = openai_messages[0] - assert first_msg["role"] == "assistant" - assert "reasoning_content" in first_msg - assert "tool_calls" in first_msg - assert first_msg["tool_calls"][0]["function"]["name"] == "test_function" - - # 验证第二个消息 - second_msg = openai_messages[1] - assert second_msg["role"] == "tool" - assert "tool_call_id" in second_msg - - print("✅ 统一OpenAI格式测试通过") - - -def test_stream_and_nonstream_consistency(): - """测试流式和非流式响应的一致性""" - print("\n=== 测试流式和非流式一致性 ===") - - converter = OpenAIConverter() - - # 模拟agent返回的消息 - agent_messages = [ - { - "role": "assistant", - "content": "让我帮你查询文件", - "reasoning_content": "用户想查询文件信息", - "function_call": { - "name": "list_files", - "arguments": '{"path": "/tmp"}' - } - } - ] - - # 非流式处理 - openai_messages = converter.convert_messages_to_openai_format(agent_messages) - - # 创建非流式响应 - non_stream_response = converter.create_openai_response( - "test-model", - openai_messages, - {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15} - ) - - print("非流式响应:") - print(json.dumps(non_stream_response, indent=2, ensure_ascii=False)) - - # 验证响应结构 - assert "choices" in non_stream_response - assert "usage" in non_stream_response - assert non_stream_response["choices"][0]["message"]["role"] == "assistant" - assert "tool_calls" in non_stream_response["choices"][0]["message"] - - print("✅ 流式和非流式一致性测试通过") - - -def test_no_more_string_conversion(): - """测试不再有字符串转换""" - print("\n=== 测试无字符串转换 ===") - - # 测试直接返回OpenAI格式,不转换为字符串 - test_messages = [ - { - "role": "assistant", - "content": "这是一个简单的回答", - "function_call": { - "name": "simple_tool", - "arguments": '{"input": "test"}' - } - } - ] - - result = convert_messages_to_openai_format(test_messages) - - # 验证返回的是结构化数据,不是字符串 - assert isinstance(result, list) - assert isinstance(result[0], dict) - assert "tool_calls" in result[0] - assert "function" in result[0]["tool_calls"][0] - - # 验证没有旧的字符串标记 - msg_str = str(result) - assert "[THINK]" not in msg_str - assert "[TOOL_CALL]" not in msg_str - assert "[ANSWER]" not in msg_str - - print("返回的数据类型:", type(result)) - print("消息结构:", json.dumps(result[0], indent=2, ensure_ascii=False)) - print("✅ 无字符串转换测试通过") - - -def main(): - """运行所有测试""" - print("测试统一使用OpenAI格式") - print("=" * 50) - - test_unified_openai_format() - test_stream_and_nonstream_consistency() - test_no_more_string_conversion() - - print("\n" + "=" * 50) - print("✅ 所有测试通过") - print("\n统一后的优势:") - print("1. 流式和非流式都使用OpenAI格式") - print("2. 不再有字符串转换的开销") - print("3. 完全兼容OpenAI API协议") - print("4. 代码更加简洁和一致") - - -if __name__ == "__main__": - main() \ No newline at end of file