130 lines
3.9 KiB
Python
130 lines
3.9 KiB
Python
"""
|
||
带流式输出的调试脚本 - 实时显示Agent的执行情况
|
||
|
||
使用方法：
|
||
export PYTHONIOENCODING=utf-8 && python tests/debug_with_stream.py
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
from datetime import datetime
|
||
|
||
# 添加项目根目录到Python路径
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from src.agents.coordinator import create_research_coordinator
|
||
from src.config import Config
|
||
|
||
|
||
def stream_research(question: str, depth: str = "quick"):
|
||
"""
|
||
调试研究流程，实时显示执行情况
|
||
|
||
Args:
|
||
question: 研究问题
|
||
depth: 深度模式
|
||
"""
|
||
print("\n" + "🔬 " * 40)
|
||
print("智能深度研究系统 - 流式调试模式")
|
||
print("🔬 " * 40)
|
||
|
||
print(f"\n研究问题: {question}")
|
||
print(f"深度模式: {depth}")
|
||
print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
|
||
# 创建Agent
|
||
print("\n" + "="*80)
|
||
print("创建Agent...")
|
||
print("="*80)
|
||
|
||
try:
|
||
agent = create_research_coordinator(
|
||
question=question,
|
||
depth=depth,
|
||
format="technical",
|
||
min_tier=3
|
||
)
|
||
print("✅ Agent创建成功")
|
||
except Exception as e:
|
||
print(f"❌ Agent创建失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return
|
||
|
||
# 执行研究（使用stream模式）
|
||
print("\n" + "="*80)
|
||
print("开始执行（流式模式）...")
|
||
print("="*80)
|
||
|
||
try:
|
||
start_time = datetime.now()
|
||
|
||
# 使用stream方法实时显示
|
||
step_count = 0
|
||
for chunk in agent.stream({
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": f"请开始研究这个问题：{question}"
|
||
}
|
||
]
|
||
}):
|
||
step_count += 1
|
||
print(f"\n{'='*60}")
|
||
print(f"步骤 #{step_count} - {datetime.now().strftime('%H:%M:%S')}")
|
||
print('='*60)
|
||
|
||
# 显示当前chunk的内容
|
||
if isinstance(chunk, dict):
|
||
# 检查是否有新消息
|
||
if 'messages' in chunk:
|
||
messages = chunk['messages']
|
||
if messages:
|
||
last_msg = messages[-1]
|
||
msg_type = type(last_msg).__name__
|
||
print(f"消息类型: {msg_type}")
|
||
|
||
if hasattr(last_msg, 'content'):
|
||
content = last_msg.content
|
||
if content:
|
||
print(f"内容: {content[:200]}")
|
||
|
||
if hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
|
||
print(f"工具调用:")
|
||
for tc in last_msg.tool_calls:
|
||
print(f" - {tc.get('name', 'unknown')}")
|
||
|
||
# 检查是否有文件更新
|
||
if 'files' in chunk:
|
||
files = chunk['files']
|
||
print(f"文件系统: {len(files)} 个文件")
|
||
for path in list(files.keys())[:5]:
|
||
print(f" - {path}")
|
||
|
||
# 超时保护
|
||
elapsed = (datetime.now() - start_time).total_seconds()
|
||
if elapsed > 120: # 2分钟
|
||
print("\n⚠️ 超过2分钟，停止...")
|
||
break
|
||
|
||
end_time = datetime.now()
|
||
duration = (end_time - start_time).total_seconds()
|
||
|
||
print("\n" + "="*80)
|
||
print("执行完成")
|
||
print("="*80)
|
||
print(f"总步骤数: {step_count}")
|
||
print(f"总耗时: {duration:.2f}秒")
|
||
|
||
except KeyboardInterrupt:
|
||
print("\n\n⚠️ 用户中断")
|
||
except Exception as e:
|
||
print(f"\n\n❌ 执行失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
question = "Python asyncio最佳实践"
|
||
stream_research(question, depth="quick")
|