from __future__ import annotations import json from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import List, Optional from agents.react_agent import ReActAgent from core.config import Config from core.llm import HelloAgentsLLM from core.message import Message from context.builder import ContextBuilder, ContextConfig, ContextPacket from tools.registry import ToolRegistry from tools.builtin.note_tool import NoteTool from tools.builtin.terminal_tool import TerminalTool from tools.builtin.plan_tool import PlanTool from tools.builtin.todo_tool import TodoTool from tools.builtin.context_fetch_tool import ContextFetchTool @dataclass class CodeAgentPaths: """CodeAgent 路径配置类，集中管理所有相关目录路径""" repo_root: Path notes_dir: Path memory_dir: Path sessions_dir: Path logs_dir: Path @property def helloagents_dir(self) -> Path: """返回 .helloagents 目录路径""" return self.repo_root / ".helloagents" @property def prompts_dir(self) -> Path: """返回 prompts 目录路径""" return self.repo_root / "code_agent" / "prompts" class CodeAgent: """ 类似 Claude Code/Codex 的 CLI 智能体： - 核心循环使用 ReActAgent。 - ContextBuilder 负责拼接：系统提示词 + 最近对话 + 相关笔记 + 情景记忆。 - 规划能力作为可选工具 (`plan`) 暴露给模型，模型可按需调用。 """ def __init__(self, repo_root: Path, llm: Optional[HelloAgentsLLM] = None, config: Optional[Config] = None): """ 初始化 CodeAgent Args: repo_root: 代码仓库根目录 llm: LLM 实例 config: 配置对象 """ repo_root = repo_root.resolve() self.config = config or Config.from_env() # 初始化目录结构 helloagents_dir = Path(self.config.helloagents_dir) state_root = helloagents_dir if helloagents_dir.is_absolute() else (repo_root / helloagents_dir) self.paths = CodeAgentPaths( repo_root=repo_root, notes_dir=state_root / "notes", memory_dir=state_root / "memory", sessions_dir=state_root / "sessions", logs_dir=state_root / "logs", ) # 确保所有必要目录存在 self.paths.helloagents_dir.mkdir(parents=True, exist_ok=True) self.paths.notes_dir.mkdir(parents=True, exist_ok=True) self.paths.sessions_dir.mkdir(parents=True, exist_ok=True) # memory / logs 仅在需要时创建，这里不再预建 self.session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}" self.llm = llm or HelloAgentsLLM() # 初始化工具 (真实实现) self.note_tool = NoteTool(workspace=str(self.paths.notes_dir)) # 类似 Claude Code：默认允许 Shell 语法 (管道等)，但危险操作需确认 self.terminal_tool = TerminalTool( workspace=str(self.paths.repo_root), timeout=60, confirm_dangerous=True, default_shell_mode=True, ) self.todo_tool = TodoTool(workspace=str(self.paths.helloagents_dir / "todos")) # ReActAgent 的工具注册表 # 核心工具：terminal, note, memory, plan # 扩展上下文工具：context_fetch（让模型按需获取更多证据） self.registry = ToolRegistry() self.registry.register_tool(self.terminal_tool) self.registry.register_tool(self.note_tool) self.registry.register_tool(PlanTool(self.llm, prompt_path=str(self.paths.prompts_dir / "plan.md"))) self.registry.register_tool(self.todo_tool) # 注册上下文获取工具（让模型按需探索） self.context_fetch_tool = ContextFetchTool( workspace=str(self.paths.repo_root), note_tool=self.note_tool, memory_tool=None, max_tokens_per_source=800, context_lines=5, ) self.registry.register_tool(self.context_fetch_tool) # 初始化上下文构建器（lazy_fetch=True：只构建保底上下文） self.context_builder = ContextBuilder( memory_tool=None, rag_tool=None, config=ContextConfig( max_tokens=8000, reserve_ratio=0.15, max_history_turns=10, enable_compression=True, include_output_format=False, lazy_fetch=True, # 按需探索模式 ), llm=self.llm, ) # 加载自定义 Prompt 并初始化 ReActAgent react_prompt = (self.paths.prompts_dir / "react.md").read_text(encoding="utf-8") summarize_prompt = (self.paths.prompts_dir / "summarize_observation.md").read_text(encoding="utf-8") def _summarize_observation(tool_name: str, tool_input: str, observation: str) -> str: """ 使用 LLM 压缩工具输出 (避免将巨大的原始输出放入 Prompt) """ truncated = observation if len(truncated) > 8000: truncated = truncated[:8000] + "\n...truncated...\n" user_msg = ( f"Tool: {tool_name}\n" f"Input: {tool_input}\n\n" f"Output:\n{truncated}" ) return self.llm.invoke( [ {"role": "system", "content": summarize_prompt}, {"role": "user", "content": user_msg}, ], max_tokens=400, ) or "" self.react = ReActAgent( name="code_agent", llm=self.llm, tool_registry=self.registry, max_steps=20, custom_prompt=react_prompt, observation_summarizer=_summarize_observation, summarize_threshold_chars=1800, ) base_system = (self.paths.prompts_dir / "system.md").read_text(encoding="utf-8") self.tools_reference_path = self.paths.prompts_dir / "tools.md" self.system_prompt = base_system self.history: List[Message] = [] self.recent_tool_packets: List[ContextPacket] = [] self.last_direct_reply: bool = False def _is_chitchat(self, text: str) -> bool: """判断是否为闲聊，避免不必要的工具调用""" t = (text or "").strip().lower() return t in {"hi", "hello", "hey", "yo", "你好", "您好", "在吗", "嗨", "哈喽"} def _is_history_query(self, text: str) -> bool: """判断是否为'回顾刚才说了什么'的元请求""" t = (text or "").strip().lower() patterns = [ "说了什么", "刚才说了什么", "之前说了什么", "what did i say", "what did we say", "recap", "summary of conversation", ] return any(p in t for p in patterns) def _reply_with_recent_history(self, limit: int = 6) -> str: """生成最近对话的简要回顾""" # 只取用户/助手消息（跳过系统等） items = [m for m in self.history if m.role in {"user", "assistant"}][-limit * 2 :] if not items: return "目前还没有可回顾的对话历史。" lines = [] for m in items: role = "你" if m.role == "user" else "助手" lines.append(f"- {role}: {m.content}") return "下面是最近的对话回顾：\n" + "\n".join(lines) # 以下两个方法在 lazy_fetch 模式下不再主动调用， # 扩展上下文改由模型通过 context_fetch 工具按需获取。 # 保留这些方法以支持 lazy_fetch=False 的传统模式。 def _note_packets(self, query: str) -> List[ContextPacket]: """检索相关笔记并封装为 ContextPacket""" packets: List[ContextPacket] = [] if self._is_chitchat(query): return packets try: # 获取最近的阻碍 (Blocker) blockers = self.note_tool.run({"action": "list", "note_type": "blocker", "limit": 2}) if blockers and isinstance(blockers, str) and "暂无" not in blockers: packets.append(ContextPacket(content=f"[Notes:blocker]\n{blockers}", metadata={"source": "note"})) # 搜索相关笔记 hits = self.note_tool.run({"action": "search", "query": query, "limit": 3}) if hits and isinstance(hits, str) and "未找到" not in hits: packets.append(ContextPacket(content=f"[Notes:search]\n{hits}", metadata={"source": "note"})) except Exception: pass return packets def _memory_packets(self, query: str) -> List[ContextPacket]: """检索相关记忆并封装为 ContextPacket""" packets: List[ContextPacket] = [] if self._is_chitchat(query): return packets try: hits = self.memory_tool.run( {"action": "search", "query": query, "memory_types": self.memory_tool.memory_types, "limit": 5, "min_importance": 0.0} ) if hits and isinstance(hits, str) and "未找到" not in hits: packets.append(ContextPacket(content=f"[Memory]\n{hits}", metadata={"source": "memory"})) except Exception: pass return packets def _persist_session(self) -> None: """持久化当前会话到 JSON 文件""" p = self.paths.sessions_dir / f"{self.session_id}.json" data = { "session_id": self.session_id, "updated_at": datetime.now().isoformat(), "history": [ {"role": m.role, "content": m.content, "timestamp": m.timestamp.isoformat()} for m in self.history[-50:] ], } p.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def run_turn(self, user_input: str) -> str: """ 执行一轮对话： 1. 收集上下文 (笔记、记忆、最近工具输出) 2. 构建完整 Prompt 3. 运行 ReAct 循环 4. 更新历史并持久化 """ # 空输入：提示而不进入 ReAct if not user_input.strip(): return "请提供具体指令或问题。" # 闲聊/问候：直接回复，避免 ReAct 的严格格式解析失败，也避免无谓的工具调用。 if self._is_chitchat(user_input): self.last_direct_reply = True reply = "你好！我是 Code Agent，可以帮你按需探索代码仓库、生成补丁并在确认后落盘。你想做什么？（例如：分析项目结构 / 搜索某个类 / 修复一个报错）" self.history.append(Message(content=user_input, role="user", timestamp=datetime.now())) self.history.append(Message(content=reply, role="assistant", timestamp=datetime.now())) if len(self.history) > 50: self.history = self.history[-50:] self._persist_session() return reply self.last_direct_reply = False # 元请求：回顾最近对话 if self._is_history_query(user_input): self.last_direct_reply = True reply = self._reply_with_recent_history(limit=6) self.history.append(Message(content=user_input, role="user", timestamp=datetime.now())) self.history.append(Message(content=reply, role="assistant", timestamp=datetime.now())) if len(self.history) > 50: self.history = self.history[-50:] self._persist_session() return reply # 若检测到明显多步骤词汇，向模型追加轻量提示（不强制，只提高倾向） multistep_hint = "" multi_patterns = ["分步", "步骤", "三步", "计划", "改造", "完成后", "多步", "多步骤"] if any(p in user_input for p in multi_patterns): multistep_hint = "提示：本任务包含多个步骤，先用 todo 记录/更新，再执行；收尾用 todo list 汇总。" # 构建保底上下文（系统提示 + 对话历史 + 上次工具摘要 + 可选 hint） # 扩展上下文由模型通过 context_fetch 工具按需获取 tool_summaries = [] for packet in self.recent_tool_packets[-3:]: tool_summaries.append(packet.content) context_text = self.context_builder.build_base( user_query=user_input, conversation_history=self.history, system_instructions=self.system_prompt + ("\n" + multistep_hint if multistep_hint else ""), tool_summaries=tool_summaries if tool_summaries else None, ) # 将拼接好的上下文作为"问题"输入给 ReAct response = self.react.run(context_text, max_tokens=8000) # 收集本轮的工具执行证据 (已在 ReActAgent 内部摘要) try: tool_summaries: List[str] = [] todo_used = False todo_listed = False for item in getattr(self.react, "last_trace", [])[-6:]: summary = item.get("observation_summary") tname = item.get("tool_name") if tname == "todo": todo_used = True if "list" in str(item.get("tool_input", "")): todo_listed = True if summary: tool_summaries.append( f"[{item.get('tool_name')}] {item.get('tool_input')}\n{summary}" ) if tool_summaries: self.recent_tool_packets.append( ContextPacket( content="[Tool Evidence]\n" + "\n\n".join(tool_summaries), metadata={"type": "tool_result", "source": "react"}, ) ) # 保持缓冲区较小 if len(self.recent_tool_packets) > 8: self.recent_tool_packets = self.recent_tool_packets[-8:] except Exception: pass # 更新历史记录 (保留最近 50 条) self.history.append(Message(content=user_input, role="user", timestamp=datetime.now())) self.history.append(Message(content=response, role="assistant", timestamp=datetime.now())) if len(self.history) > 50: self.history = self.history[-50:] self._persist_session() try: if todo_used and not todo_listed: todo_snapshot = self.registry.execute_tool("todo", {"action": "list"}) response = f"{response}\n\nTodo board:\n{todo_snapshot}" except Exception: pass return response