200 lines
5.5 KiB
Python
200 lines
5.5 KiB
Python
"""
|
||
最小化测试 - 理解DeepAgents的工作机制
|
||
|
||
使用方法：
|
||
export PYTHONIOENCODING=utf-8 && python tests/test_minimal_agent.py
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
|
||
# 添加项目根目录到Python路径
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from deepagents import create_deep_agent
|
||
from src.config import Config
|
||
|
||
|
||
def test_minimal_agent():
|
||
"""测试最简单的Agent执行"""
|
||
|
||
print("\n" + "="*80)
|
||
print("最小化测试 - 主Agent写文件")
|
||
print("="*80)
|
||
|
||
# 创建一个最简单的主Agent
|
||
main_system_prompt = """你是一个简单的测试Agent。
|
||
|
||
你的任务：
|
||
1. 使用 write_file 工具写入一个文件到 `/test.txt`，内容为 "Hello World"
|
||
2. 使用 read_file 工具读取 `/test.txt`
|
||
3. 告诉用户文件内容
|
||
|
||
**重要**：完成后明确说"任务完成"。
|
||
"""
|
||
|
||
agent = create_deep_agent(
|
||
model=Config.get_llm(),
|
||
subagents=[], # 不使用SubAgent
|
||
system_prompt=main_system_prompt,
|
||
)
|
||
|
||
print("✅ Agent创建成功")
|
||
print("\n开始执行...")
|
||
|
||
try:
|
||
result = agent.invoke({
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": "请开始执行任务"
|
||
}
|
||
]
|
||
})
|
||
|
||
print("\n" + "="*80)
|
||
print("执行结果")
|
||
print("="*80)
|
||
|
||
# 检查消息
|
||
if 'messages' in result:
|
||
print(f"\n消息数量: {len(result['messages'])}")
|
||
|
||
# 显示最后一条消息
|
||
last_msg = result['messages'][-1]
|
||
print(f"\n最后一条消息:")
|
||
if hasattr(last_msg, 'content'):
|
||
print(last_msg.content)
|
||
|
||
# 检查文件系统
|
||
if 'files' in result:
|
||
print(f"\n文件数量: {len(result['files'])}")
|
||
|
||
for path, info in result['files'].items():
|
||
print(f"\n文件: {path}")
|
||
if isinstance(info, dict) and 'content' in info:
|
||
print(f"内容: {info['content']}")
|
||
else:
|
||
print(f"内容: {info}")
|
||
|
||
print("\n✅ 测试完成")
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ 测试失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
def test_agent_with_subagent():
|
||
"""测试主Agent和SubAgent的文件共享"""
|
||
|
||
print("\n" + "="*80)
|
||
print("测试主Agent和SubAgent的文件共享")
|
||
print("="*80)
|
||
|
||
# 定义一个简单的SubAgent
|
||
subagent_config = {
|
||
"name": "file-reader",
|
||
"description": "读取文件并返回内容",
|
||
"system_prompt": """你是一个文件读取Agent。
|
||
|
||
你的任务：
|
||
1. 使用 read_file 工具读取 `/test.txt` 文件
|
||
2. 告诉用户文件内容
|
||
|
||
**重要**：
|
||
- 如果文件不存在，明确说"文件不存在"
|
||
- 如果文件存在，告诉用户文件内容
|
||
- 完成后明确说"任务完成"
|
||
""",
|
||
"tools": [],
|
||
}
|
||
|
||
# 主Agent
|
||
main_system_prompt = """你是一个测试协调Agent。
|
||
|
||
你的任务：
|
||
1. 使用 write_file 工具写入一个文件到 `/test.txt`，内容为 "Hello from Main Agent"
|
||
2. 使用 task 工具调用 file-reader SubAgent：task(description="读取测试文件", subagent_type="file-reader")
|
||
3. 等待SubAgent返回结果
|
||
4. 告诉用户SubAgent读取的内容
|
||
|
||
**重要**：完成后明确说"所有任务完成"。
|
||
"""
|
||
|
||
agent = create_deep_agent(
|
||
model=Config.get_llm(),
|
||
subagents=[subagent_config],
|
||
system_prompt=main_system_prompt,
|
||
)
|
||
|
||
print("✅ Agent创建成功（1主 + 1子）")
|
||
print("\n开始执行...")
|
||
|
||
try:
|
||
result = agent.invoke({
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": "请开始执行任务"
|
||
}
|
||
]
|
||
})
|
||
|
||
print("\n" + "="*80)
|
||
print("执行结果")
|
||
print("="*80)
|
||
|
||
# 检查消息
|
||
if 'messages' in result:
|
||
print(f"\n消息数量: {len(result['messages'])}")
|
||
|
||
# 显示所有消息内容
|
||
print("\n所有消息:")
|
||
for i, msg in enumerate(result['messages'], 1):
|
||
print(f"\n--- 消息 #{i} ---")
|
||
msg_type = type(msg).__name__
|
||
print(f"类型: {msg_type}")
|
||
|
||
if hasattr(msg, 'content'):
|
||
content = msg.content
|
||
if len(content) > 200:
|
||
print(f"内容: {content[:200]}...")
|
||
else:
|
||
print(f"内容: {content}")
|
||
|
||
if hasattr(msg, 'tool_calls') and msg.tool_calls:
|
||
print(f"工具调用: {msg.tool_calls}")
|
||
|
||
# 检查文件系统
|
||
if 'files' in result:
|
||
print(f"\n文件系统:")
|
||
print(f"文件数量: {len(result['files'])}")
|
||
|
||
for path, info in result['files'].items():
|
||
print(f"\n 文件: {path}")
|
||
if isinstance(info, dict) and 'content' in info:
|
||
print(f" 内容: {info['content']}")
|
||
else:
|
||
print(f" 内容: {info}")
|
||
|
||
print("\n✅ 测试完成")
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ 测试失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("\n🧪 DeepAgents最小化测试")
|
||
print("="*80)
|
||
|
||
# 测试1：单个Agent的文件操作
|
||
test_minimal_agent()
|
||
|
||
print("\n\n")
|
||
|
||
# 测试2：主Agent和SubAgent的文件共享
|
||
test_agent_with_subagent()
|