""" 选题点整体推导 Agent（增强版） 参考 examples/how/run.py，提供： 1. 命令行交互：输入 'p' 暂停、'q' 退出 2. 暂停后可插入干预消息、触发经验总结、查看 GoalTree、手动压缩上下文 3. 支持 --trace 恢复已有 Trace 继续执行 4. 使用 SimplePrompt 加载 derivation_main.md，支持评估子 agent（agent_type=evaluate_derivation） """ import argparse import os import sys import select import asyncio from datetime import datetime from pathlib import Path # 与 examples/how/run.py 一致：禁止 httpx/urllib 自动检测系统 HTTP 代理 # os.environ.setdefault("no_proxy", "*") # 添加项目根目录到 Python 路径 sys.path.insert(0, str(Path(__file__).parent.parent.parent)) from dotenv import 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 agent.llm import create_openrouter_llm_call from agent.trace.compaction import build_reflect_prompt load_dotenv() # ===== 非阻塞 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(): ch = msvcrt.getwch().lower() if ch == 'p': return 'pause' if ch == 'q': return 'quit' return None else: 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 _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, ): """显示交互式菜单，让用户选择操作。""" 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("\n将插入干预消息并继续执行...") 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 return { "action": "continue", "messages": [{"role": "user", "content": text}], "after_sequence": actual_sequence, } elif choice == "2": print("\n触发经验总结...") focus = input("请输入反思重点（可选，直接回车跳过）: ").strip() trace = await store.get_trace(trace_id) 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: await store.update_trace(trace_id, head_sequence=saved_head) if reflection_text: from datetime import datetime experiences_path = runner.experiences_path or "./.cache/experiences_overall_derivation.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" with open(experiences_path, "a", encoding="utf-8") as f: f.write(header + reflection_text + "\n") print(f"\n反思已保存到: {experiences_path}") print("\n--- 反思内容 ---") print(reflection_text) print("--- 结束 ---\n") else: print("未生成反思内容") 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 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("无效选项，请重新输入") def _replace_prompt_placeholders( messages: list, account_name: str, post_id: str, log_id: str, post_point_count: int, ) -> None: """在 messages 的 content 中用 replace 替换 {account_name}, {帖子ID}, {log_id}, {post_point_count}。""" post_point_count_str = str(post_point_count) for m in messages: content = m.get("content") if isinstance(content, str): m["content"] = ( content.replace("{account_name}", account_name) .replace("{帖子ID}", post_id) .replace("{log_id}", log_id) .replace("{post_point_count}", post_point_count_str) ) elif isinstance(content, list): for part in content: if isinstance(part, dict) and part.get("type") == "text": part["text"] = ( (part.get("text") or "") .replace("{account_name}", account_name) .replace("{帖子ID}", post_id) .replace("{log_id}", log_id) .replace("{post_point_count}", post_point_count_str) ) async def main(account_name, post_id): 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 prompt_path = base_dir / "derivation_main.md" output_dir = base_dir / "output" output_dir.mkdir(exist_ok=True) # 加载项目级 presets（evaluate_derivation、derivation_search 等） 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())}") # 注册选题点推导专用工具（主 agent 与评估子 agent 会调用） import importlib.util tools_dir = base_dir / "tools" for mod_name, file_name in [ ("find_tree_node", "find_tree_node.py"), ("find_pattern", "find_pattern.py"), ("point_match", "point_match.py"), ]: path = tools_dir / file_name if path.is_file(): spec = importlib.util.spec_from_file_location(f"overall_derivation.{mod_name}", path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) print(f" - 已注册推导工具: {mod_name}") skills_dir = str(base_dir / "skills") print("=" * 60) print("选题点整体推导 Agent（交互增强）") print("=" * 60) print() print("💡 交互提示：") print(" - 执行过程中输入 'p' 或 'pause' 暂停并进入交互模式") print(" - 执行过程中输入 'q' 或 'quit' 停止执行") print("=" * 60) print() # 在读取 prompt 前生成 log_id（格式 yyyyMMddHHmmss），保证每次运行使用同一 log_id（用于推导日志输出路径） log_id = datetime.now().strftime("%Y%m%d%H%M%S") print(f" - 本次运行 log_id: {log_id}") print(f" - account_name: {account_name}") print(f" - post_id: {post_id}") # 读取选题点列表，得到 post_point_count（用于 prompt 占位符） input_dir = base_dir / "input" / account_name / "post_topic" post_topic_path = input_dir / f"{post_id}.json" post_point_count = 0 if post_topic_path.exists(): import json with open(post_topic_path, "r", encoding="utf-8") as f: post_topics = json.load(f) post_point_count = len(post_topics) if isinstance(post_topics, list) else 0 print(f" - 选题点数量 post_point_count: {post_point_count} (来自 {post_topic_path.relative_to(base_dir)})") else: print(f" - 未找到选题点文件: {post_topic_path}，post_point_count 使用 0") print("1. 加载 prompt 配置...") prompt = SimplePrompt(prompt_path) print("2. 构建任务消息...") messages = prompt.build_messages() _replace_prompt_placeholders(messages, account_name, post_id, log_id, post_point_count) print("3. 创建 Agent Runner...") print(f" - Skills 目录: {skills_dir}") model_key = prompt.config.get("model", "google/gemini-3-flash-preview") # model_id = f"google/{model_key}" if not model_key.startswith("google/") else model_key model_id = model_key print(f" - 模型: {model_id}") store = FileSystemTraceStore(base_path=".trace") runner = AgentRunner( trace_store=store, llm_call=create_openrouter_llm_call(model=model_id), skills_dir=skills_dir, # experiences_path="./.cache/experiences_overall_derivation.md", debug=True, ) resume_trace_id = args.trace if resume_trace_id: 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("4. 启动新 Agent 模式...") print() final_response = "" current_trace_id = resume_trace_id current_sequence = 0 should_exit = False try: if resume_trace_id: initial_messages = None config = RunConfig( model=model_id, temperature=float(prompt.config.get("temperature", 0.3)), max_iterations=200, trace_id=resume_trace_id, ) else: initial_messages = messages config = RunConfig( model=model_id, temperature=float(prompt.config.get("temperature", 0.3)), max_iterations=200, name="选题点整体推导任务", ) while not should_exit: if current_trace_id: config.trace_id = current_trace_id final_response = "" 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("\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: initial_messages = [] config.after_sequence = None continue break initial_messages = [] print(f"{'▶️ 开始执行...' if not current_trace_id else '▶️ 继续执行...'}") 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) 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 if isinstance(item, Trace): current_trace_id = item.trace_id if item.status == "running": print(f"[Trace] 开始: {item.trace_id[:100]}...") elif item.status == "completed": print("\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("\n[Trace] ⏸️ 已停止") elif isinstance(item, Message): current_sequence = item.sequence 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 print("\n[Response] Agent 回复：") print(text) elif text: preview = text[:500] + "..." if len(text) > 500 else text print(f"[Assistant] {preview}") if tool_calls: for tc in tool_calls: tool_name = tc.get("function", {}).get("name", "unknown") tool_args = tc.get("function", {}).get("arguments", "") print(f"[Tool Call] 🛠️ {tool_name}") print(f" params: {tool_args}") elif item.role == "tool": content = item.content if isinstance(content, dict): tool_name = content.get("tool_name", "unknown") tool_result = content.get("result", content) print(f"[Tool Result] ✅ {tool_name}") print(f" result: {tool_result}") if item.description: desc = item.description[:500] if len(item.description) > 500 else item.description print(f" {desc}...") except Exception as e: print(f"\n执行出错: {e}") import traceback traceback.print_exc() if paused: if should_exit: break continue if should_exit: break 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) if final_response: print() print("=" * 60) print("Agent 响应:") print("=" * 60) print(final_response) print("=" * 60) print() output_file = output_dir / account_name / "推导日志" / current_trace_id / log_id / "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: print("=" * 60) print("可视化 Step Tree:") print("=" * 60) print("1. 启动 API Server:") print(" python3 api_server.py") print() print("2. 浏览器访问:") print(" http://localhost:8000/api/traces") print() print(f"3. Trace ID: {current_trace_id}") print(f"4. Log ID（推导日志目录）: {log_id}") print("=" * 60) if __name__ == "__main__": # anthropic/claude-sonnet-4.6 # google/gemini-3-flash-preview asyncio.run(main(account_name="家有大志", post_id="68fb6a5c000000000302e5de"))