1. 项目概述为什么我们需要一个“会思考”的编程智能体最近在折腾AI编程助手从Cursor到Claude Code再到各种开源模型用了一圈下来发现一个问题它们大多还是“问答机”模式。你问它答上下文一长就失忆任务一复杂就乱套。这就像你有一个知识渊博但记性差、不会规划的新手搭档你得不停地给它喂信息、拆任务、纠正方向累得半死。所以我决定自己动手从零开始构建一个更接近“智能体”的东西。它不应该只是一个被动的代码补全工具而应该是一个能主动思考、规划、执行并记忆的编程伙伴。这个项目的核心就是标题里的三个关键词主循环、上下文压缩与Hook设计。主循环是它的大脑负责调度整个思考-行动流程上下文压缩是它的记忆系统解决大模型有限的“工作记忆”问题Hook设计则是它的神经系统让我们能在关键节点注入自定义逻辑实现灵活的控制与扩展。这不仅仅是另一个AI代码生成脚本。我想探讨的是如何将软件工程中成熟的设计模式如事件驱动、中间件、状态机与大语言模型的能力相结合打造一个真正可协作、可演进、能处理复杂编程任务的智能体框架。无论你是想深入AI应用开发还是单纯想提升自己的编程效率理解这套设计思路都会大有裨益。2. 核心架构设计主循环、记忆与控制的三角支撑一个健壮的AI编程智能体其架构必须清晰地区分“思考”、“记忆”和“控制”这三个核心职责。我的设计正是围绕这三个支柱展开的。2.1 主循环智能体的“心脏”与状态机主循环是整个智能体的驱动引擎它定义了一个持续的“感知-思考-行动”循环。我把它设计成一个有限状态机而不是简单的while True循环。这样做的最大好处是状态清晰、逻辑可控便于调试和扩展。一个典型的主循环状态流转可能包括空闲等待用户输入或外部事件。目标解析接收用户指令如“为这个函数添加错误处理”并将其分解为明确的、可执行的任务。规划基于当前代码库上下文和任务制定分步行动计划。例如先定位函数再分析现有逻辑最后生成补丁代码。执行调用具体的工具如代码编辑器API、终端命令、文件读写来执行规划中的一步。观察收集执行结果成功、失败、输出内容。反思根据观察结果评估当前步骤是否达成目标是否需要调整计划。压缩/总结将本轮循环的关键信息提炼后存入长期记忆上下文压缩。这个循环会持续运行直到任务被标记为完成或失败。在实现上我使用一个AgentState枚举来管理状态用一个AgentContext对象来携带循环中所有的数据如当前任务、历史记录、代码上下文等。主循环控制器根据当前状态和上下文决定下一个状态是什么并调用对应的处理器。注意主循环的频率需要仔细设计。无脑地快速循环会大量消耗API额度且效率低下。我通常会引入“节流”机制例如在“观察”和“反思”状态后根据任务复杂度等待几百毫秒模拟人类“想一想”的过程也让系统更有节奏感。2.2 上下文压缩突破令牌限制的“记忆宫殿”这是智能体能否处理大型项目的关键。大语言模型有上下文窗口限制比如128K但一个项目的代码、文档、历史对话很容易超过这个限制。我们不能把整个项目历史都塞给模型这就需要上下文压缩。我的策略是分层管理记忆工作记忆即当前对话轮次的上下文包含最新的用户指令、模型回复和工具执行结果。这部分保持完整确保模型对当前操作有连贯理解。短期记忆存储最近几轮循环的关键信息。我设计了一个“滚动摘要”缓冲区。每完成一个任务步骤就生成一段简短的摘要例如“步骤3在utils.py的validate_input函数中添加了类型检查和对空值的处理。”并将其推入缓冲区。缓冲区有固定长度旧的摘要会被挤出或合并。长期记忆这是智能体的知识库。我采用向量数据库来实现。所有重要的代码片段、文档总结、解决过的问题案例都会被转换成向量嵌入存储起来。当智能体需要背景知识时比如“我之前是怎么处理类似登录逻辑的”就从向量数据库中检索最相关的几条记忆动态注入工作上下文。压缩的具体操作发生在“压缩/总结”状态。这里有一个核心技巧不是简单删除旧消息而是让模型自己来总结。我会给模型一个提示“请将之前关于[某个模块]的讨论浓缩成一段不超过3句话的摘要重点保留核心决策和代码结构。” 这样生成的摘要信息密度高且保留了关键逻辑。2.3 Hook设计赋予智能体“可观测性”与“可控制性”Hook钩子机制是连接智能体核心框架与外部自定义逻辑的桥梁。它借鉴了中间件和事件驱动的思想允许我们在智能体运行的关键生命周期节点插入自定义代码。我定义了以下几类核心Hook生命周期Hook在on_cycle_start,on_cycle_end,on_state_change等时机触发。可以用来打日志、收集性能指标、或强制中断运行。上下文Hook在before_context_sent_to_llm发送给模型前和after_response_received收到回复后触发。这是功能最强大的Hook。发送前我们可以修改或丰富提示词收到回复后我们可以解析、验证甚至重写模型的输出。工具调用Hook在before_tool_execution和after_tool_execution触发。可以用来检查工具调用的安全性比如禁止执行rm -rf /或者对工具执行结果进行预处理。一个实战案例代码风格强制统一。我写了一个before_context_sent_to_llm的Hook它会检查当前编辑的文件如果项目有.clang-format或.prettierrc配置就在系统提示词里动态追加一条“你生成的所有代码必须严格符合项目中已配置的[代码风格]规范。” 这比事后用格式化工具去修更直接从源头保证了代码风格一致。Hook的设计让这个智能体框架不再是黑盒。你可以为它添加“单元测试”通过Hook模拟工具调用、实现“A/B测试”不同提示词策略、或者集成到CI/CD流程中真正做到深度定制。3. 关键技术实现与核心代码解析理论讲完了我们来看看具体怎么实现。我会用Python和LangChain框架作为基础来演示核心模块但设计思想是语言无关的。3.1 主循环的状态机实现首先我们定义状态和上下文数据类。from enum import Enum from typing import Any, Dict, List, Optional from pydantic import BaseModel class AgentState(Enum): IDLE idle PARSING_GOAL parsing_goal PLANNING planning EXECUTING executing OBSERVING observing REFLECTING reflecting SUMMARIZING summarizing DONE done ERROR error class AgentContext(BaseModel): 智能体运行上下文贯穿整个主循环 current_state: AgentState AgentState.IDLE user_input: str parsed_goal: Optional[Dict] None plan: List[str] [] current_step_index: int 0 execution_result: Optional[Any] None observation: Optional[str] None reflection: Optional[str] None working_memory: List[Dict] [] # 原始对话记录 short_term_memory: List[str] [] # 摘要缓冲区 # ... 其他业务相关字段接下来是主循环控制器。它不处理具体业务只负责状态流转和调用对应的处理器。class AgentCore: def __init__(self, llm, tools, memory_store): self.llm llm self.tools tools self.memory memory_store self.context AgentContext() self.state_handlers { AgentState.PARSING_GOAL: self._handle_parsing, AgentState.PLANNING: self._handle_planning, # ... 注册其他状态处理器 } self.hooks [] # Hook注册列表 def run_cycle(self, user_input: str): 运行一个完整的主循环 self.context.user_input user_input self.context.current_state AgentState.PARSING_GOAL while self.context.current_state not in [AgentState.DONE, AgentState.ERROR]: # 1. 触发状态变更Hook self._trigger_hooks(on_state_enter, self.context.current_state) # 2. 执行当前状态对应的业务逻辑 handler self.state_handlers.get(self.context.current_state) if handler: handler() else: # 没有处理器进入错误状态 self.context.current_state AgentState.ERROR break # 3. 状态转移逻辑这里简化实际会更复杂 self._transition_state() # 4. 触发状态退出Hook self._trigger_hooks(on_state_exit, self.context.current_state) return self.context def _handle_parsing(self): 解析用户目标 prompt f 用户指令{self.context.user_input} 请将该指令解析为一个结构化任务包含 1. 主要目标 2. 涉及的文件或模块如果可推断 3. 任务类型如代码生成、代码修复、代码审查、问题诊断 请以JSON格式回复。 # 触发before_context_sent_to_llm Hook可以在这里增强prompt enhanced_prompt self._trigger_hooks_and_modify(before_context_sent_to_llm, prompt) response self.llm.invoke(enhanced_prompt) # 触发after_response_received Hook可以在这里验证和清洗response parsed_response self._trigger_hooks_and_modify(after_response_received, response) try: self.context.parsed_goal json.loads(parsed_response.content) except json.JSONDecodeError: # 解析失败可以重试或进入错误状态 self.context.current_state AgentState.ERROR self.context.working_memory.append({role: user, content: prompt}) self.context.working_memory.append({role: assistant, content: parsed_response.content}) def _transition_state(self): 简单的状态转移逻辑实际项目会更复杂 transition_map { AgentState.PARSING_GOAL: AgentState.PLANNING, AgentState.PLANNING: AgentState.EXECUTING, AgentState.EXECUTING: AgentState.OBSERVING, AgentState.OBSERVING: AgentState.REFLECTING, AgentState.REFLECTING: AgentState.SUMMARIZING, AgentState.SUMMARIZING: AgentState.DONE, } next_state transition_map.get(self.context.current_state) if next_state: self.context.current_state next_state3.2 上下文压缩与向量记忆的实现短期记忆滚动摘要的实现相对直接我们用一个双端队列来维护固定长度的摘要。from collections import deque class ShortTermMemory: def __init__(self, max_length: int 10): self.buffer deque(maxlenmax_length) def add_summary(self, summary: str): 添加一轮摘要 self.buffer.append(summary) def get_recent_summaries(self, n: int 5) - str: 获取最近N条摘要合并成字符串 recent list(self.buffer)[-n:] return \n.join(recent) if recent else 暂无近期记忆。长期记忆向量数据库需要集成像Chroma、Weaviate或Qdrant这样的向量库。这里以Chroma为例。import chromadb from chromadb.config import Settings from langchain.embeddings import OpenAIEmbeddings # 或其他嵌入模型 from langchain.text_splitter import RecursiveCharacterTextSplitter class LongTermMemory: def __init__(self, persist_dir: str ./memory_db): self.client chromadb.PersistentClient(pathpersist_dir, settingsSettings(allow_resetTrue)) self.collection self.client.get_or_create_collection(namecode_memory) self.embedder OpenAIEmbeddings() # 需要替换为你的嵌入模型 self.text_splitter RecursiveCharacterTextSplitter(chunk_size1000, chunk_overlap200) def store(self, code_snippet: str, file_path: str, metadata: dict None): 存储代码片段到长期记忆 # 分割文本避免单个片段过长 chunks self.text_splitter.split_text(code_snippet) ids [] embeddings [] metadatas [] for i, chunk in enumerate(chunks): chunk_id f{file_path}_{i} ids.append(chunk_id) # 生成向量嵌入 embedding self.embedder.embed_query(chunk) embeddings.append(embedding) meta {file_path: file_path, chunk_index: i, ** (metadata or {})} metadatas.append(meta) # 存入向量数据库 self.collection.add( embeddingsembeddings, documentschunks, metadatasmetadatas, idsids ) def retrieve(self, query: str, n_results: int 3) - List[Dict]: 根据查询检索相关记忆 query_embedding self.embedder.embed_query(query) results self.collection.query( query_embeddings[query_embedding], n_resultsn_results ) # 组织返回格式 retrieved [] if results[documents]: for doc, meta in zip(results[documents][0], results[metadatas][0]): retrieved.append({ content: doc, metadata: meta, distance: results[distances][0][results[documents][0].index(doc)] # 假设距离可用 }) return retrieved在智能体的“总结”状态我们需要调用LLM来生成摘要并决定是否存入长期记忆。def _handle_summarizing(self): 总结本轮循环更新记忆 # 从工作记忆中提取本轮关键交互 recent_interactions self.context.working_memory[-6:] # 取最近几轮 summary_prompt f 以下是智能体最近完成的一个任务步骤的对话记录 {recent_interactions} 请用一段话不超过100字总结这一步我们做了什么、结果如何、有什么关键决策或代码变更。 摘要 summary self.llm.invoke(summary_prompt).content # 存入短期记忆 self.short_term_memory.add_summary(summary) # 判断是否值得存入长期记忆例如涉及核心逻辑修改、解决了复杂问题等 if self._is_worth_long_term_memory(summary): # 提取相关的代码片段从执行结果中 code_snippet self._extract_code_from_result(self.context.execution_result) if code_snippet: self.long_term_memory.store( code_snippetcode_snippet, file_pathself.context.parsed_goal.get(file, unknown), metadata{summary: summary, task: self.context.user_input} )3.3 Hook系统的设计与注册Hook系统本质上是一个事件发布-订阅模型。我们定义一个基类然后允许用户注册具体的Hook实现。from abc import ABC, abstractmethod from typing import Callable class BaseHook(ABC): Hook基类 abstractmethod def execute(self, context: AgentContext, data: Any None) - Any: pass class BeforeContextSentHook(BaseHook): 在上下文发送给LLM前执行的Hook def execute(self, context: AgentContext, prompt: str) - str: # 默认实现不做修改 return prompt class CodeStyleHook(BeforeContextSentHook): 强制代码风格统一的Hook def __init__(self, style_guide_path: str None): self.style_guide self._load_style_guide(style_guide_path) def execute(self, context: AgentContext, prompt: str) - str: # 检查当前任务是否涉及代码生成 if context.parsed_goal and context.parsed_goal.get(type) code_generation: style_reminder f\n\n重要提示你生成的所有代码必须严格遵循以下风格规范{self.style_guide} return prompt style_reminder return prompt def _load_style_guide(self, path): # 从文件加载或返回默认指南 if path and os.path.exists(path): with open(path, r) as f: return f.read() return PEP 8 for Python, with 4-space indentation. # 在AgentCore中管理Hook class AgentCore: # ... 其他代码 ... def register_hook(self, hook: BaseHook, hook_point: str): 注册Hook到指定钩子点 if not hasattr(self, _hooks_registry): self._hooks_registry {} if hook_point not in self._hooks_registry: self._hooks_registry[hook_point] [] self._hooks_registry[hook_point].append(hook) def _trigger_hooks_and_modify(self, hook_point: str, initial_data: Any) - Any: 触发特定钩子点的所有Hook并允许它们修改数据 data initial_data hooks getattr(self, _hooks_registry, {}).get(hook_point, []) for hook in hooks: # 将数据传递给Hook并用Hook的返回值更新数据 data hook.execute(self.context, data) return data使用时我们可以这样注册Hookagent AgentCore(llm, tools, memory) # 注册代码风格Hook style_hook CodeStyleHook(style_guide_path.clang-format) agent.register_hook(style_hook, before_context_sent_to_llm) # 注册一个日志Hook class LoggingHook(BaseHook): def execute(self, context, data): print(f[LOG] State: {context.current_state}, Step: {context.current_step_index}) return data # 对于日志Hook通常返回原数据不变 agent.register_hook(LoggingHook(), on_state_enter)4. 实战集成Claude Code与处理复杂任务有了核心框架我们就可以将其与具体的AI服务如Claude Code API和编辑器如VSCode集成打造一个可用的智能体。4.1 与Claude Code API的适配层Claude Code提供了强大的代码生成和理解能力。我们需要创建一个适配器将我们智能体的“思考”转化为Claude能理解的请求。import anthropic class ClaudeCodeAdapter: def __init__(self, api_key: str, model: str claude-3-5-sonnet-20241022): self.client anthropic.Anthropic(api_keyapi_key) self.model model def generate_code(self, prompt: str, context_files: List[Dict] None) - str: 调用Claude Code生成代码。 context_files: [{path: file.py, content: ...}] 提供相关文件上下文 system_prompt 你是一个专业的软件工程师AI助手。请根据用户的请求和提供的代码上下文生成高质量、可运行、符合最佳实践的代码。如果请求不明确请先询问澄清。 messages [{role: user, content: prompt}] # 如果有上下文文件可以以特定格式附加到消息中 if context_files: context_text \n\n相关代码上下文\n for cf in context_files: context_text f--- File: {cf[path]} ---\n{cf[content]}\n\n messages[0][content] context_text prompt try: response self.client.messages.create( modelself.model, max_tokens4096, systemsystem_prompt, messagesmessages ) return response.content[0].text except anthropic.APIError as e: print(fClaude API错误: {e}) return fError: {e}然后在我们的工具集中创建一个“代码生成工具”它内部调用这个适配器。class CodeGenerationTool: def __init__(self, claude_adapter: ClaudeCodeAdapter, workspace_root: str): self.adapter claude_adapter self.workspace_root workspace_root def execute(self, task_description: str, target_file: str None, context_file_paths: List[str] None) - Dict: 执行代码生成。 返回格式{success: bool, code: str, message: str} # 1. 读取上下文文件内容 contexts [] if context_file_paths: for path in context_file_paths: full_path os.path.join(self.workspace_root, path) if os.path.exists(full_path): with open(full_path, r, encodingutf-8) as f: contexts.append({path: path, content: f.read()}) # 2. 构建给Claude的提示词 prompt f 任务{task_description} if target_file: prompt f\n目标文件{target_file} # 3. 调用Claude生成代码 generated_code self.adapter.generate_code(prompt, contexts) # 4. 解析响应这里简单处理实际可能需要解析出代码块 # 假设Claude返回的是纯代码或Markdown代码块 import re code_pattern r(?:\w)?\n([\s\S]*?)\n matches re.findall(code_pattern, generated_code) if matches: final_code matches[0] # 取第一个代码块 else: final_code generated_code # 如果没有代码块假设整个回复就是代码 return { success: True if final_code and len(final_code.strip()) 10 else False, code: final_code, raw_response: generated_code, message: 代码生成成功 if final_code else 未生成有效代码 }4.2 处理一个完整任务为现有函数添加错误处理让我们模拟智能体处理一个真实任务“为项目中的data_loader.py文件的load_csv函数添加全面的错误处理。”目标解析智能体进入PARSING_GOAL状态调用LLM解析指令。输出可能是{ goal: 为load_csv函数添加错误处理, file: data_loader.py, type: code_modification, sub_tasks: [定位函数, 分析现有逻辑, 设计错误处理方案, 生成补丁代码] }规划进入PLANNING状态。智能体可能会先检索长期记忆看看有没有类似的错误处理模式。然后制定计划步骤1读取data_loader.py文件内容。步骤2定位load_csv函数。步骤3分析函数逻辑识别潜在错误点文件不存在、编码错误、数据格式错误等。步骤4生成带有try-catch/异常处理的代码补丁。步骤5应用补丁或提供修改建议。执行与观察EXECUTING调用“文件读取工具”获取data_loader.py内容。OBSERVING观察工具返回的文件内容。EXECUTING调用“代码分析工具”可能是另一个LLM调用来识别风险点。OBSERVING接收分析结果如“第15行pd.read_csv可能抛出FileNotFoundError”。EXECUTING调用我们上面定义的CodeGenerationTool提示词为“基于以下函数代码和识别的风险点为其添加健壮的异常处理记录日志并返回友好的错误信息。函数代码{...} 风险点[...]”。OBSERVING接收生成的代码补丁。反思REFLECTING状态中智能体会评估生成的补丁是否覆盖了所有风险点是否引入了新问题如果满意则进入下一步如果不满意可能调整提示词重新生成或标记需要人工介入。总结SUMMARIZING状态中生成摘要“在data_loader.py的load_csv函数中围绕文件I/O和pandas解析添加了try-except块捕获了FileNotFoundError、pd.errors.EmptyDataError等异常并集成了日志记录。” 这个摘要被存入短期记忆。如果这是一个通用性强的错误处理模式其核心代码片段还可能被存入长期记忆。在整个过程中Hook系统在默默工作在发送给Claude前代码风格Hook追加了格式要求日志Hook记录了每个状态切换安全Hook可能检查生成的代码是否包含危险操作。5. 避坑指南与性能调优心得从零搭建这样一个系统我踩过不少坑。这里分享一些关键的注意事项和调优经验。5.1 上下文管理的陷阱与对策陷阱1摘要失真。让模型做摘要有时它会遗漏关键细节或引入错误。对策提供更结构化的摘要指令。不要只说“总结一下”而是说“请总结关于[函数XXX修改]的讨论必须包含1) 修改动机2) 具体变更代码行号3) 测试结果。” 这能引导模型抓住重点。陷阱2向量检索不准。存了太多琐碎代码导致检索时噪音太大。对策精心设计存储策略。只存储“有价值”的代码片段比如函数定义、类设计、复杂算法、解决特定bug的方案。为每个存储片段添加丰富的元数据如功能描述、所属模块、创建时间、关联任务ID检索时结合元数据过滤。陷阱3令牌数超限。即使有压缩在复杂任务中工作记忆短期记忆检索的记忆仍然可能超窗口。对策实现动态上下文窗口。设定一个令牌预算如模型上限的80%。在组装最终上下文时按优先级加入1) 系统指令和当前任务2) 最近的工作记忆3) 检索到的相关长期记忆按相关性得分排序直到预算用尽。这是一个经典的“缓存淘汰”问题。5.2 主循环的稳定性和效率死循环状态机设计不当可能导致在两个状态间来回跳转。对策为每个状态转移设置守卫条件和最大重试次数。例如从EXECUTING到OBSERVING是自动的但从REFLECTING到PLANNING需要重新规划必须有条件触发比如“反思结果认为当前计划完全不可行”。同时整个主循环应有超时设置。工具调用失败网络问题、权限问题、工具本身bug都会导致执行失败。对策所有工具调用必须放在try-catch中。在after_tool_execution的Hook里可以实现统一的错误处理和重试逻辑。对于非致命错误如临时网络超时可以自动重试1-2次。规划过于宏大或模糊LLM有时会制定不切实际的庞大计划。对策在规划阶段后加入一个“计划评审”步骤可以是一个简单的Hook或独立状态。用另一条更严格的提示词让模型自我评审“这个计划是否分解成了可立即执行的小步骤每一步是否有明确的成功标准” 或者由你来定义计划模板强制模型按“步骤1...步骤2...”的格式输出。5.3 与Claude Code等商业API集成的注意事项成本控制智能体如果频繁调用API成本会快速上升。对策缓存对相同的提示词和上下文缓存LLM的响应。例如对“读取文件X”这种确定性操作结果可以缓存一段时间。降级策略对于简单的、非创造性的任务如代码格式化、简单查找优先使用本地规则或轻量级模型如本地运行的CodeLlama而不是每次都调用Claude。预算监控实现一个简单的令牌计数器设置每日或每任务预算超限后自动暂停或切换模式。速率限制API有调用频率限制。对策在主循环中内置延迟并使用令牌桶等算法进行限流。将非紧急的、可批量处理的操作如批量存储记忆到向量库放到后台线程。处理API错误Claude Code API可能返回各种错误如400 Bad Request,429 Too Many Requests。对策实现一个健壮的API客户端包装器包含指数退避的重试机制。对于400错误要分析是否是提示词问题并尝试简化或重构提示词。5.4 提示词工程的经验之谈智能体的“智商”很大程度上取决于你给LLM的提示词。经过大量实验我总结出几个原则角色扮演要具体不要只说“你是一个助手”要说“你是一个拥有10年Python后端开发经验的专家特别擅长编写可维护、健壮的错误处理代码。你注重代码清晰度和日志可观测性。”思维链Chain-of-Thought强制化在需要复杂推理的步骤如规划、反思明确要求模型“逐步思考”。例如“在给出最终答案前请先分析代码的输入、输出和潜在故障点。”提供结构化输出示例对于需要解析模型输出的地方如目标解析、计划生成在系统提示词里直接给出JSON格式的例子。这能极大提高输出格式的稳定性。上下文标记清晰当在提示词中拼接多个文件或对话历史时使用清晰的标记如 文件data_loader.py 帮助模型区分不同信息源。构建一个真正的AI编程智能体是一个迭代的过程。从最简单的“问答代理”开始逐步加入规划、记忆、工具调用和Hook。每增加一个功能你都会对它能力的边界和脆弱性有更深的理解。这个框架不是一个完美的成品而是一个探索的起点。你可以根据自己的需求替换其中的LLM、工具、记忆存储或者添加更复杂的Hook来实现监控、审计、人机协同审批等高级功能。