""" 示例(增强版) 使用 Agent 模式 + Skills 新增功能: 1. 支持命令行随时打断(输入 'p' 暂停,'q' 退出) 2. 暂停后可插入干预消息 3. 支持触发经验总结 4. 查看当前 GoalTree 5. 框架层自动清理不完整的工具调用 6. 支持通过 --trace 恢复已有 Trace 继续执行 """ import argparse import os import sys import select import asyncio import json from pathlib import Path from typing import Any # Clash Verge TUN 模式兼容:禁止 httpx/urllib 自动检测系统 HTTP 代理 # TUN 虚拟网卡已在网络层接管所有流量,不需要应用层再走 HTTP 代理, # 否则 httpx 检测到 macOS 系统代理 (127.0.0.1:7897) 会导致 ConnectError # os.environ.setdefault("no_proxy", "*") # 添加项目根目录到 Python 路径 sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from dotenv import load_dotenv load_dotenv() from agent.llm.prompts import SimplePrompt from agent.core.runner import AgentRunner, RunConfig from agent.core.presets import AgentPreset, register_preset from agent.trace import ( FileSystemTraceStore, Trace, Message, ) from examples.create.html import trace_to_html from agent.llm import create_openrouter_llm_call from agent.tools import get_tool_registry DEFAULT_MODEL = "anthropic/claude-sonnet-4.5" # DEFAULT_MODEL = "google/gemini-3-flash-preview" # ===== 非阻塞 stdin 检测 ===== if sys.platform == 'win32': import msvcrt def check_stdin() -> str | None: """ 跨平台非阻塞检查 stdin 输入。 Windows: 使用 msvcrt.kbhit() macOS/Linux: 使用 select.select() """ if sys.platform == 'win32': # 检查是否有按键按下 if msvcrt.kbhit(): # 读取按下的字符(msvcrt.getwch 是非阻塞读取宽字符) ch = msvcrt.getwch().lower() if ch == 'p': return 'pause' if ch == 'q': return 'quit' # 如果是其他按键,可以选择消耗掉或者忽略 return None else: # Unix/Mac 逻辑 ready, _, _ = select.select([sys.stdin], [], [], 0) if ready: line = sys.stdin.readline().strip().lower() if line in ('p', 'pause'): return 'pause' if line in ('q', 'quit'): return 'quit' return None # ===== 格式化打印 ===== def _format_json(obj: Any, indent: int = 2) -> str: """格式化 JSON 对象为字符串""" try: return json.dumps(obj, indent=indent, ensure_ascii=False) except (TypeError, ValueError): # 如果无法序列化为 JSON,返回字符串表示 return str(obj) def _print_message_details(message: Message): """完整打印消息的详细信息""" print("\n" + "=" * 80) print(f"[Message #{message.sequence}] {message.role.upper()}") print("=" * 80) # 基本信息 if message.goal_id: print(f"Goal ID: {message.goal_id}") if message.parent_sequence is not None: print(f"Parent Sequence: {message.parent_sequence}") if message.tool_call_id: print(f"Tool Call ID: {message.tool_call_id}") # 内容打印 if message.role == "user": print("\n[输入内容]") print("-" * 80) if isinstance(message.content, str): print(message.content) else: print(_format_json(message.content)) elif message.role == "assistant": content = message.content if isinstance(content, dict): text = content.get("text", "") tool_calls = content.get("tool_calls") if text: print("\n[LLM 文本回复]") print("-" * 80) print(text) if tool_calls: print(f"\n[工具调用] (共 {len(tool_calls)} 个)") print("-" * 80) for idx, tc in enumerate(tool_calls, 1): func = tc.get("function", {}) tool_name = func.get("name", "unknown") tool_id = tc.get("id", "unknown") arguments = func.get("arguments", {}) print(f"\n工具 #{idx}: {tool_name}") print(f" Call ID: {tool_id}") print(f" 参数:") # 尝试解析 arguments(可能是字符串或字典) if isinstance(arguments, str): try: parsed_args = json.loads(arguments) print(_format_json(parsed_args, indent=4)) except json.JSONDecodeError: print(f" {arguments}") else: print(_format_json(arguments, indent=4)) elif isinstance(content, str): print("\n[LLM 文本回复]") print("-" * 80) print(content) else: print("\n[内容]") print("-" * 80) print(_format_json(content)) if message.finish_reason: print(f"\n完成原因: {message.finish_reason}") elif message.role == "tool": content = message.content print("\n[工具执行结果]") print("-" * 80) if isinstance(content, dict): tool_name = content.get("tool_name", "unknown") result = content.get("result", content) print(f"工具名称: {tool_name}") print(f"\n返回结果:") if isinstance(result, str): print(result) elif isinstance(result, list): # 可能是多模态内容(包含图片) for idx, item in enumerate(result, 1): if isinstance(item, dict) and item.get("type") == "image_url": print(f" [{idx}] 图片 (base64, 已省略显示)") else: print(f" [{idx}] {item}") else: print(_format_json(result)) else: print(str(content) if content is not None else "(无内容)") elif message.role == "system": print("\n[系统提示]") print("-" * 80) if isinstance(message.content, str): print(message.content) else: print(_format_json(message.content)) # Token 和成本信息 if message.prompt_tokens is not None or message.completion_tokens is not None: print("\n[Token 使用]") print("-" * 80) if message.prompt_tokens is not None: print(f" 输入 Tokens: {message.prompt_tokens:,}") if message.completion_tokens is not None: print(f" 输出 Tokens: {message.completion_tokens:,}") if message.reasoning_tokens is not None: print(f" 推理 Tokens: {message.reasoning_tokens:,}") if message.cache_creation_tokens is not None: print(f" 缓存创建 Tokens: {message.cache_creation_tokens:,}") if message.cache_read_tokens is not None: print(f" 缓存读取 Tokens: {message.cache_read_tokens:,}") if message.tokens: print(f" 总计 Tokens: {message.tokens:,}") if message.cost is not None: print(f"\n[成本] ${message.cost:.6f}") if message.duration_ms is not None: print(f"[执行时间] {message.duration_ms}ms") print("=" * 80 + "\n") # ===== 交互菜单 ===== async def perform_reflection( runner: AgentRunner, trace_id: str, store: FileSystemTraceStore, focus: str = "", intervention_message: str = "", ) -> str: """ 执行经验总结(反思) Args: runner: AgentRunner 实例 trace_id: Trace ID store: TraceStore 实例 focus: 可选的反思重点 intervention_message: 可选的干预消息,如果提供则一并记录到经验文件中 Returns: 反思文本内容,如果失败则返回空字符串 """ from agent.trace.compaction import build_reflect_prompt from datetime import datetime # 保存当前 head_sequence trace = await store.get_trace(trace_id) if not trace: print("未找到 Trace,无法执行反思") return "" saved_head = trace.head_sequence prompt = build_reflect_prompt() if focus: prompt += f"\n\n请特别关注:{focus}" print("正在生成反思...") reflect_cfg = RunConfig(trace_id=trace_id, max_iterations=1, tools=[]) reflection_text = "" try: result = await runner.run_result( messages=[{"role": "user", "content": prompt}], config=reflect_cfg, ) reflection_text = result.get("summary", "") finally: # 恢复 head_sequence(反思消息成为侧枝) await store.update_trace(trace_id, head_sequence=saved_head) # 追加到 experiences 文件 if reflection_text: experiences_path = runner.experiences_path or "./.cache/experiences.md" os.makedirs(os.path.dirname(experiences_path), exist_ok=True) header = f"\n\n---\n\n## {trace_id} ({datetime.now().strftime('%Y-%m-%d %H:%M')})\n\n" # 如果提供了干预消息,也一并记录 content_parts = [] if intervention_message: content_parts.append(f"**干预消息:**\n{intervention_message}\n") content_parts.append(f"**反思总结:**\n{reflection_text}") with open(experiences_path, "a", encoding="utf-8") as f: f.write(header + "\n".join(content_parts) + "\n") print(f"\n反思已保存到: {experiences_path}") if intervention_message: print("\n--- 干预消息 ---") print(intervention_message) print("\n--- 反思内容 ---") print(reflection_text) print("--- 结束 ---\n") else: print("未生成反思内容") return reflection_text def _read_multiline() -> str: """ 读取多行输入,以连续两次回车(空行)结束。 单次回车只是换行,不会提前终止输入。 """ print("\n请输入干预消息(连续输入两次回车结束):") lines: list[str] = [] blank_count = 0 while True: line = input() if line == "": blank_count += 1 if blank_count >= 2: break lines.append("") # 保留单个空行 else: blank_count = 0 lines.append(line) # 去掉尾部多余空行 while lines and lines[-1] == "": lines.pop() return "\n".join(lines) async def show_interactive_menu( runner: AgentRunner, trace_id: str, current_sequence: int, store: FileSystemTraceStore, ): """ 显示交互式菜单,让用户选择操作。 进入本函数前不再有后台线程占用 stdin,所以 input() 能正常工作。 """ print("\n" + "=" * 60) print(" 执行已暂停") print("=" * 60) print("请选择操作:") print(" 1. 插入干预消息并继续") print(" 2. 触发经验总结(reflect)") print(" 3. 查看当前 GoalTree") print(" 4. 手动压缩上下文(compact)") print(" 5. 继续执行") print(" 6. 停止执行") print("=" * 60) while True: choice = input("请输入选项 (1-6): ").strip() if choice == "1": text = _read_multiline() if not text: print("未输入任何内容,取消操作") continue print(f"\n将插入干预消息并继续执行...") # 从 store 读取实际的 last_sequence,避免本地 current_sequence 过时 live_trace = await store.get_trace(trace_id) actual_sequence = live_trace.last_sequence if live_trace and live_trace.last_sequence else current_sequence # 触发干预后,自动执行一次经验总结(并将干预消息一并记录) print("\n" + "=" * 60) print("自动触发经验总结...") print("=" * 60) await perform_reflection(runner, trace_id, store, focus="", intervention_message=text) return { "action": "continue", "messages": [{"role": "user", "content": text}], "after_sequence": actual_sequence, } elif choice == "2": # 触发经验总结 print("\n触发经验总结...") focus = input("请输入反思重点(可选,直接回车跳过): ").strip() await perform_reflection(runner, trace_id, store, focus=focus) continue elif choice == "3": goal_tree = await store.get_goal_tree(trace_id) if goal_tree and goal_tree.goals: print("\n当前 GoalTree:") print(goal_tree.to_prompt()) else: print("\n当前没有 Goal") continue elif choice == "4": # 手动压缩上下文 print("\n正在执行上下文压缩(compact)...") try: goal_tree = await store.get_goal_tree(trace_id) trace = await store.get_trace(trace_id) if not trace: print("未找到 Trace,无法压缩") continue # 重建当前 history main_path = await store.get_main_path_messages(trace_id, trace.head_sequence) history = [msg.to_llm_dict() for msg in main_path] head_seq = main_path[-1].sequence if main_path else 0 next_seq = head_seq + 1 compact_config = RunConfig(trace_id=trace_id) new_history, new_head, new_seq = await runner._compress_history( trace_id=trace_id, history=history, goal_tree=goal_tree, config=compact_config, sequence=next_seq, head_seq=head_seq, ) print(f"\n✅ 压缩完成: {len(history)} 条消息 → {len(new_history)} 条") except Exception as e: print(f"\n❌ 压缩失败: {e}") continue elif choice == "5": print("\n继续执行...") return {"action": "continue"} elif choice == "6": print("\n停止执行...") return {"action": "stop"} else: print("无效选项,请重新输入") async def main(): # 解析命令行参数 parser = argparse.ArgumentParser(description="任务 (Agent 模式 + 交互增强)") parser.add_argument( "--trace", type=str, default=None, help="已有的 Trace ID,用于恢复继续执行(不指定则新建)", ) args = parser.parse_args() # 路径配置 base_dir = Path(__file__).parent project_root = base_dir.parent.parent prompt_path = base_dir / "create.prompt" output_dir = base_dir / "output_1" output_dir.mkdir(exist_ok=True) # 加载项目级 presets(examples/create/presets.json) presets_path = base_dir / "presets.json" if presets_path.exists(): import json with open(presets_path, "r", encoding="utf-8") as f: project_presets = json.load(f) for name, cfg in project_presets.items(): register_preset(name, AgentPreset(**cfg)) print(f" - 已加载项目 presets: {list(project_presets.keys())}") # Skills 目录(可选:用户自定义 skills) # 注意:内置 skills(agent/memory/skills/)会自动加载 skills_dir = str(base_dir / "skills") print("=" * 60) print("mcp/skills 发现、获取、评价 分析任务 (Agent 模式 + 交互增强)") print("=" * 60) print() print("💡 交互提示:") print(" - 执行过程中输入 'p' 或 'pause' 暂停并进入交互模式") print(" - 执行过程中输入 'q' 或 'quit' 停止执行") print("=" * 60) print() # 1. 加载 prompt print("1. 加载 prompt 配置...") prompt = SimplePrompt(prompt_path) # 读取 system.md 并替换 {system} 占位符 system_md_path = base_dir / "PRD" / "system.md" if system_md_path.exists(): system_content = system_md_path.read_text(encoding='utf-8') if 'system' in prompt._messages and '{system}' in prompt._messages['system']: prompt._messages['system'] = prompt._messages['system'].replace('{system}', system_content) else: print(f" - 警告: system.md 文件不存在: {system_md_path}") # 读取 create_process.md 并替换 {create_process} 占位符 create_process_md_path = base_dir / "PRD" / "create_process.md" if create_process_md_path.exists(): create_process_content = create_process_md_path.read_text(encoding='utf-8') if 'system' in prompt._messages and '{create_process}' in prompt._messages['system']: prompt._messages['system'] = prompt._messages['system'].replace('{create_process}', create_process_content) print(f" - 已替换 create_process.md 内容到 prompt") else: print(f" - 警告: prompt 中未找到 {{create_process}} 占位符") else: print(f" - 警告: create_process.md 文件不存在: {create_process_md_path}") # 读取 user.md 并替换 {input} 占位符 input_md_path = base_dir / "PRD" / "input.md" if input_md_path.exists(): user_content = input_md_path.read_text(encoding='utf-8') if 'user' in prompt._messages and '{input}' in prompt._messages['user']: prompt._messages['user'] = prompt._messages['user'].replace('{input}', user_content) print(f" - 已替换 user.md 内容到 prompt") else: print(f" - 警告: prompt 中未找到 {{input}} 占位符") else: print(f" - 警告: user.md 文件不存在: {input_md_path}") output_md_path = base_dir / "PRD" / "output.md" if output_md_path.exists(): user_content = output_md_path.read_text(encoding='utf-8') if 'user' in prompt._messages and '{output}' in prompt._messages['user']: prompt._messages['user'] = prompt._messages['user'].replace('{output}', user_content) print(f" - 已替换 user.md 内容到 prompt") else: print(f" - 警告: prompt 中未找到 {{output}} 占位符") else: print(f" - 警告: user.md 文件不存在: {output_md_path}") print("\n替换后的prompt:") print("=" * 60) print("System:") print("-" * 60) print(prompt._messages.get('system', '')) print("=" * 60) if 'user' in prompt._messages: print("\nUser:") print("-" * 60) print(prompt._messages['user']) print("=" * 60) print() # 2. 构建消息(仅新建时使用,恢复时消息已在 trace 中) print("2. 构建任务消息...") messages = prompt.build_messages() # 3. 创建 Agent Runner(配置 skills) print("3. 创建 Agent Runner...") print(f" - Skills 目录: {skills_dir}") print(f" - 模型: {prompt.config.get('model', 'sonnet-4.5')}") # 加载自定义工具 print(" - 加载自定义工具: topic_search") import examples.create.tool # 选题检索工具,用于在数据库中匹配已有帖子选题 store = FileSystemTraceStore(base_path=".trace") runner = AgentRunner( trace_store=store, llm_call=create_openrouter_llm_call(model=prompt.config.get('model', DEFAULT_MODEL)), skills_dir=skills_dir, experiences_path="./.cache/experiences.md", debug=True ) # 4. 判断是新建还是恢复 resume_trace_id = args.trace if resume_trace_id: # 验证 trace 存在 existing_trace = await store.get_trace(resume_trace_id) if not existing_trace: print(f"\n错误: Trace 不存在: {resume_trace_id}") sys.exit(1) print(f"4. 恢复已有 Trace: {resume_trace_id[:8]}...") print(f" - 状态: {existing_trace.status}") print(f" - 消息数: {existing_trace.total_messages}") print(f" - 任务: {existing_trace.task}") else: print(f"4. 启动新 Agent 模式...") print() final_response = "" current_trace_id = resume_trace_id current_sequence = 0 should_exit = False try: # 恢复模式:不发送初始消息,只指定 trace_id 续跑 if resume_trace_id: initial_messages = None # None = 未设置,触发早期菜单检查 config = RunConfig( model=prompt.config.get('model', DEFAULT_MODEL), temperature=float(prompt.config.get('temperature', 0.3)), max_iterations=1000, trace_id=resume_trace_id, ) else: initial_messages = messages config = RunConfig( model=prompt.config.get('model', DEFAULT_MODEL), temperature=float(prompt.config.get('temperature', 0.3)), max_iterations=1000, name="社交媒体内容解构、建构、评估任务", ) while not should_exit: # 如果是续跑,需要指定 trace_id if current_trace_id: config.trace_id = current_trace_id # 清理上一轮的响应,避免失败后显示旧内容 final_response = "" # 如果 trace 已完成/失败且没有新消息,直接进入交互菜单 # 注意:initial_messages 为 None 表示未设置(首次加载),[] 表示有意为空(用户选择"继续") if current_trace_id and initial_messages is None: check_trace = await store.get_trace(current_trace_id) if check_trace and check_trace.status in ("completed", "failed"): if check_trace.status == "completed": print(f"\n[Trace] ✅ 已完成") print(f" - Total messages: {check_trace.total_messages}") print(f" - Total cost: ${check_trace.total_cost:.4f}") else: print(f"\n[Trace] ❌ 已失败: {check_trace.error_message}") current_sequence = check_trace.head_sequence menu_result = await show_interactive_menu( runner, current_trace_id, current_sequence, store ) if menu_result["action"] == "stop": break elif menu_result["action"] == "continue": new_messages = menu_result.get("messages", []) if new_messages: initial_messages = new_messages config.after_sequence = menu_result.get("after_sequence") else: # 无新消息:对 failed trace 意味着重试,对 completed 意味着继续 initial_messages = [] config.after_sequence = None continue break # 对 stopped/running 等非终态的 trace,直接续跑 initial_messages = [] print(f"{'▶️ 开始执行...' if not current_trace_id else '▶️ 继续执行...'}") # 执行 Agent paused = False try: async for item in runner.run(messages=initial_messages, config=config): # 检查用户中断 cmd = check_stdin() if cmd == 'pause': # 暂停执行 print("\n⏸️ 正在暂停执行...") if current_trace_id: await runner.stop(current_trace_id) # 等待一小段时间让 runner 处理 stop 信号 await asyncio.sleep(0.5) # 显示交互菜单 menu_result = await show_interactive_menu( runner, current_trace_id, current_sequence, store ) if menu_result["action"] == "stop": should_exit = True paused = True break elif menu_result["action"] == "continue": # 检查是否有新消息需要插入 new_messages = menu_result.get("messages", []) if new_messages: # 有干预消息,需要重新启动循环 initial_messages = new_messages after_seq = menu_result.get("after_sequence") if after_seq is not None: config.after_sequence = after_seq paused = True break else: # 没有新消息,需要重启执行 initial_messages = [] config.after_sequence = None paused = True break elif cmd == 'quit': print("\n🛑 用户请求停止...") if current_trace_id: await runner.stop(current_trace_id) should_exit = True break # 处理 Trace 对象(整体状态变化) if isinstance(item, Trace): current_trace_id = item.trace_id if item.status == "running": print(f"[Trace] 开始: {item.trace_id[:8]}...") elif item.status == "completed": print(f"\n[Trace] ✅ 完成") print(f" - Total messages: {item.total_messages}") print(f" - Total tokens: {item.total_tokens}") print(f" - Total cost: ${item.total_cost:.4f}") elif item.status == "failed": print(f"\n[Trace] ❌ 失败: {item.error_message}") elif item.status == "stopped": print(f"\n[Trace] ⏸️ 已停止") # 处理 Message 对象(执行过程) elif isinstance(item, Message): current_sequence = item.sequence # 完整打印所有消息详情 _print_message_details(item) # 保留原有的简化输出逻辑(用于最终响应) if item.role == "assistant": content = item.content if isinstance(content, dict): text = content.get("text", "") tool_calls = content.get("tool_calls") if text and not tool_calls: # 纯文本回复(最终响应) final_response = text except Exception as e: print(f"\n执行出错: {e}") import traceback traceback.print_exc() # paused → 菜单已在暂停时内联显示过 if paused: if should_exit: break continue # quit → 直接退出 if should_exit: break # Runner 退出(完成/失败/停止/异常)→ 显示交互菜单 if current_trace_id: menu_result = await show_interactive_menu( runner, current_trace_id, current_sequence, store ) if menu_result["action"] == "stop": break elif menu_result["action"] == "continue": new_messages = menu_result.get("messages", []) if new_messages: initial_messages = new_messages config.after_sequence = menu_result.get("after_sequence") else: initial_messages = [] config.after_sequence = None continue break except KeyboardInterrupt: print("\n\n用户中断 (Ctrl+C)") if current_trace_id: await runner.stop(current_trace_id) finally: # 进程退出时自动生成 messages HTML 到 .trace// 目录 if current_trace_id: try: html_path = store.base_path / current_trace_id / "messages.html" await trace_to_html(current_trace_id, html_path, base_path=str(store.base_path)) print(f"\n✓ Messages 可视化已保存: {html_path}") except Exception as e: print(f"\n⚠ 生成 HTML 失败: {e}") # 流程执行完退出时,自动总结一次经验 try: final_trace = await store.get_trace(current_trace_id) if final_trace and final_trace.status in ("completed", "failed"): print("\n" + "=" * 60) print("流程执行完成,自动触发经验总结...") print("=" * 60) await perform_reflection(runner, current_trace_id, store, focus="") except Exception as e: print(f"\n⚠ 自动经验总结失败: {e}") import traceback traceback.print_exc() # 6. 输出结果 if final_response: print() print("=" * 60) print("Agent 响应:") print("=" * 60) print(final_response) print("=" * 60) print() # 7. 保存结果 output_file = output_dir / "result.txt" with open(output_file, 'w', encoding='utf-8') as f: f.write(final_response) print(f"✓ 结果已保存到: {output_file}") print() # 可视化提示 if current_trace_id: html_path = store.base_path / current_trace_id / "messages.html" print("=" * 60) print("可视化:") print("=" * 60) print(f"1. 本地 HTML: {html_path}") print() print("2. API Server:") print(" python3 api_server.py") print(" http://localhost:8000/api/traces") print() print(f"3. Trace ID: {current_trace_id}") print("=" * 60) if __name__ == "__main__": asyncio.run(main())