1. 项目概述为什么我们要从零构建一个 AI Agent最近“AI Agent”这个词火得不行感觉身边搞技术的朋友都在聊。但说实话很多讨论都停留在概念层面什么“自主完成任务”、“理解复杂指令”听起来很酷但具体怎么实现的底层逻辑是什么往往一笔带过。这就好比告诉你汽车能跑却不给你看发动机。作为一个喜欢“拆开看”的开发者我决定动手从最核心的代码生成场景切入实现一个我称之为“Mini Claude Code”的简化版 AI Agent。我的目标不是复现一个商业级产品而是通过这个微型项目亲手摸一遍 Agent 的核心骨架理解它到底是怎么“思考”和“行动”的。这个 Mini Claude Code 的核心功能很明确你给它一个自然语言描述的任务比如“写一个 Python 函数计算斐波那契数列”它不仅能生成代码还能像一位有经验的程序员那样进行“思考-规划-执行-检查”的完整循环。它会先拆解任务规划步骤然后尝试写代码如果运行出错它会分析错误信息调整思路重新尝试直到任务完成或达到尝试上限。这个过程就是 AI Agent 区别于单纯“问答模型”或“代码补全工具”的本质。为什么选择代码生成作为场景因为它的反馈是确定性的。代码写得好不好运行一下就知道。这为我们观察 Agent 的决策过程、调试其内部状态提供了清晰的“标尺”。通过构建这个 Mini Agent我希望你能和我一样穿透“智能体”、“自主性”这些高大上的术语看到背后实实在在的循环、状态管理和工具调用机制。这不仅是学习一个新框架更是理解一种新的软件范式。2. 核心架构设计拆解 AI Agent 的“五脏六腑”一个功能完整的 AI Agent无论大小其架构都可以抽象为几个核心组件。我们的 Mini Claude Code 将围绕这些组件进行构建理解它们各自的责任和协作方式。2.1 大脑大语言模型 (LLM) 作为推理核心Agent 的“大脑”毫无疑问是大语言模型。但这里有一个关键认知我们不是让 LLM 一次性吐出最终答案而是驱动它进行多轮、结构化的“思考”。对于 Mini Claude CodeLLM 需要承担以下角色任务理解与规划师将模糊的用户指令“写个爬虫”分解为具体的、可执行的步骤“1. 导入 requests 库2. 分析目标网页结构3. 编写解析函数...”。代码生成器根据当前步骤和上下文生成符合语法的代码片段。调试分析员当代码执行出错时分析错误信息Traceback诊断问题根源并提出修改方案。决策者判断当前步骤是否完成决定下一步是继续生成、执行测试还是向用户请求澄清。注意我们通常使用模型的 Chat Completion API而不是 Completion API。因为我们需要与模型进行多轮、带角色system, user, assistant的对话来模拟其“思考链”。选择模型时不需要追求最大参数量的模型像 GPT-3.5-Turbo、Claude Haiku 或开源的 DeepSeek-Coder 等代码能力较强的中小模型在成本、速度和效果上对这个项目更友好。2.2 记忆与状态维持对话上下文与执行历史Agent 必须有“记忆”否则每一轮交互都是孤立的无法进行连贯的任务处理。记忆系统通常分为两部分短期记忆/工作记忆即当前对话的上下文。它包含了最新的用户请求、模型的“思考”过程、工具调用的结果等。这部分通常由我们传递给 LLM 的messages列表来维护。长期记忆/执行历史记录整个任务执行过程中的所有关键事件如每一步的计划、生成的代码、执行的结果成功/失败及输出、错误信息等。这对于复杂任务中的回溯、学习和总结至关重要。在我们的实现中会用一个 Python 列表或专门的状态对象来存储这个历史。2.3 手脚工具 (Tools) 与执行环境这是 Agent 与外部世界交互的桥梁。对于代码生成 Agent最核心的工具就是代码执行器。一个安全的、沙盒化的代码执行环境是必须的。我们可以使用docker容器或者在严格限制下使用subprocess调用隔离的 Python 解释器。除了执行器工具还可以包括文件读写工具让 Agent 能创建、读取、修改项目文件。网络请求工具用于测试 API 或爬虫代码。Linter/Formatter 工具检查代码风格。Git 操作工具进行版本控制。每个工具都需要被定义成 LLM 能够理解和调用的格式通常是包含名称、描述和参数模式的函数。2.4 控制循环驱动一切的“主循环”这是 Agent 的“心脏”一个循环往复的流程。一个典型的 ReAct (Reasoning Acting) 风格的主循环逻辑如下观察整合当前的用户目标、对话历史、执行历史形成当前的“状态”或“观察”。思考将“观察”和系统指令扮演的角色、需要遵循的流程一起提交给 LLM要求它输出下一步的“思考”和“行动”。思考是内部推理“用户要爬虫我需要先检查 requests 库是否可用”行动是对工具的调用“调用execute_code工具运行import requests”。行动解析 LLM 输出的行动指令调用对应的工具函数并获取执行结果。反思将工具执行的结果成功或失败作为新的信息追加到记忆历史和上下文中。判断检查任务是否完成如代码成功运行并输出预期结果或是否达到循环上限避免死循环。如果未完成则回到第1步。这个循环将持续进行直到任务达成或终止。我们的 Mini Claude Code 将具体实现这个循环。3. 分步实现手把手搭建 Mini Claude Code理论讲完了我们开始动手。项目将使用 Python因为它生态丰富且是 LLM 应用开发的主流语言。我们将从环境搭建开始逐步实现上述每个组件。3.1 环境准备与基础依赖首先创建一个新的项目目录并初始化虚拟环境这是保持依赖清洁的好习惯。mkdir mini_claude_code cd mini_claude_code python -m venv venv # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate接下来安装核心依赖。我们主要需要三部分LLM 的 SDK、代码执行的安全沙箱、以及辅助工具。pip install openai # 或其他你选择的 LLM 提供商 SDK如 anthropic, litellm pip install docker # 用于创建安全的代码执行容器 pip install python-dotenv # 管理 API 密钥等环境变量这里我选择 OpenAI API 作为示例因为它普及度高。你需要在项目根目录创建一个.env文件来存放密钥并确保它在.gitignore中避免泄露。# .env OPENAI_API_KEYyour_api_key_here OPENAI_BASE_URLyour_base_url_if_needed # 可选如果你使用代理或特定端点3.2 定义核心状态与记忆管理我们创建一个agent_state.py文件来定义 Agent 的状态。状态对象将贯穿整个主循环。# agent_state.py from dataclasses import dataclass, field from typing import List, Dict, Any dataclass class AgentState: Agent 的核心状态容器。 # 任务目标 objective: str # 与LLM交互的完整消息历史 message_history: List[Dict[str, str]] field(default_factorylist) # 长期执行历史记录每一步的思考、行动和结果 execution_history: List[Dict[str, Any]] field(default_factorylist) # 当前步骤的索引或描述 current_step: str # 任务是否完成 is_complete: bool False # 失败次数用于防止无限循环 failure_count: int 0 def add_to_history(self, thought: str, action: str, result: str): 向执行历史添加一条记录。 self.execution_history.append({ step: len(self.execution_history) 1, thought: thought, action: action, result: result }) # 简单的失败检测如果结果包含错误增加失败计数 if error in result.lower() or traceback in result.lower(): self.failure_count 1 else: self.failure_count 0 # 连续成功则重置 def should_continue(self) - bool: 判断是否应该继续执行循环。 MAX_FAILURES 5 return (not self.is_complete) and (self.failure_count MAX_FAILURES)这个AgentState类封装了 Agent 运行所需的所有关键信息。execution_history是我们的“长期记忆”而message_history是即将发送给 LLM 的“短期记忆”。3.3 实现安全的代码执行工具这是最具技术挑战也是最重要的一环。绝对不能在主机上直接执行来自 LLM 的未知代码。我们将使用 Docker 来创建一个一次性的、隔离的执行环境。创建一个tools.py文件# tools.py import docker import tempfile import os import time class CodeExecutor: 一个基于 Docker 的安全代码执行工具。 def __init__(self, image_name: str python:3.11-slim): self.client docker.from_env() self.image_name image_name # 预先拉取镜像避免第一次执行时等待 try: self.client.images.get(image_name) except docker.errors.ImageNotFound: print(f正在拉取镜像 {image_name}...) self.client.images.pull(image_name) def execute(self, code: str, timeout: int 10) - dict: 在 Docker 容器中执行一段 Python 代码。 返回包含输出、错误和执行状态的字典。 # 创建一个临时目录来挂载代码文件更接近真实项目 with tempfile.TemporaryDirectory() as tmpdir: code_file_path os.path.join(tmpdir, script.py) with open(code_file_path, w, encodingutf-8) as f: f.write(code) # 准备容器执行命令运行我们的脚本 container_cmd fpython /workspace/script.py volumes {tmpdir: {bind: /workspace, mode: ro}} # 只读挂载 try: container self.client.containers.run( imageself.image_name, command[sh, -c, container_cmd], volumesvolumes, working_dir/workspace, detachTrue, # 后台运行 mem_limit100m, # 内存限制 cpu_period100000, cpu_quota50000, # CPU限制 (50%) network_disabledTrue, # 禁用网络更安全 removeTrue, # 运行后自动删除容器 ) # 等待容器执行完成或超时 try: result container.wait(timeouttimeout) exit_code result[StatusCode] logs container.logs(stdoutTrue, stderrTrue).decode(utf-8) # 根据退出码和日志判断是正常输出还是错误 if exit_code 0: return {status: success, output: logs.strip()} else: return {status: error, output: logs.strip()} except Exception as e: container.kill() # 超时则终止容器 return {status: timeout, output: fExecution timed out after {timeout} seconds.} except docker.errors.ContainerError as e: return {status: container_error, output: str(e)} except Exception as e: return {status: system_error, output: fSystem error: {str(e)}} # 工具函数字典方便 LLM 通过名称调用 TOOLS { execute_python_code: { function: None, # 稍后绑定实例方法 description: Execute a piece of Python code in a secure sandbox and return the output or error., parameters: { code: {type: string, description: The Python code to execute.} } } }实操心得这里有几个关键安全点。第一使用network_disabledTrue防止代码进行网络访问除非你的任务需要。第二设置内存 (mem_limit) 和 CPU (cpu_period/cpu_quota) 限制防止资源耗尽攻击。第三使用removeTrue让容器执行后自动清理避免积累垃圾容器。第四将主机目录以只读 (mode: ro) 方式挂载防止代码篡改主机文件。3.4 构建 LLM 交互与提示工程LLM 是我们的“大脑”如何与它有效沟通至关重要。我们创建一个llm_client.py文件来处理交互并精心设计系统提示词。# llm_client.py import os from openai import OpenAI from dotenv import load_dotenv import json load_dotenv() class LLMClient: def __init__(self, model: str gpt-3.5-turbo): self.client OpenAI( api_keyos.getenv(OPENAI_API_KEY), base_urlos.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) ) self.model model # 核心系统提示词定义了 Agent 的角色和行为规范 self.system_prompt 你是一个专业的 AI 编程助手Code Agent。你的任务是帮助用户完成代码编写任务。 你必须遵循以下严格的流程 1. **理解与规划**首先分析用户请求将其分解为具体的、可执行的步骤。 2. **逐步执行**一次只执行一个步骤。对于每个步骤 a. **思考**先进行内部推理说明你打算做什么以及为什么。 b. **行动**如果需要运行代码来验证想法或完成任务调用 execute_python_code 工具。请确保生成的代码是完整、可执行的。 c. **观察**分析工具返回的结果。如果成功总结学到了什么如果失败分析错误原因并规划下一步。 3. **迭代与调试**如果代码运行出错你必须分析错误信息Traceback修正代码并重新执行。不要轻易放弃。 4. **任务完成**当所有步骤完成且最终代码运行成功并输出符合要求的結果时明确宣布任务完成。 你的输出必须是严格的 JSON 格式包含两个字段 - thought: (字符串) 你的内部推理过程。 - action: (字典 或 null) 如果不需要或无法调用工具则为 null。如果需要调用工具则是一个包含 name 和 arguments 键的字典。例如 {name: execute_python_code, arguments: {code: print(hello)}} 请确保 arguments 中的 code 字段是完整的、语法正确的 Python 代码块。 现在开始处理第一个任务。 def get_completion(self, messages: list) - str: 调用 LLM API 并返回回复内容。 try: response self.client.chat.completions.create( modelself.model, messagesmessages, temperature0.1, # 低温度保证输出的稳定性尤其是 JSON 格式 response_format{type: json_object} # 强制要求返回 JSON ) return response.choices[0].message.content except Exception as e: print(fLLM API 调用失败: {e}) return json.dumps({thought: fAPI调用错误: {e}, action: None}) def parse_response(self, response_text: str) - dict: 解析 LLM 返回的 JSON 字符串。 try: return json.loads(response_text) except json.JSONDecodeError: print(f无法解析 LLM 响应为 JSON: {response_text}) # 尝试容错处理或者返回一个安全的默认响应 return {thought: 我的响应格式有误我将重新尝试。, action: None}这个系统提示词是 Agent 行为的“宪法”。它强制了 ReAct 的思考-行动模式并规定了严格的 JSON 输出格式这使我们能可靠地解析出“思考”和“行动”指令。response_format{type: json_object}是 GPT-3.5-turbo-1106 及以上模型支持的特性能极大提高 JSON 输出的稳定性。3.5 组装主控制循环最后我们把所有组件串联起来在main.py中实现核心的主循环逻辑。# main.py from agent_state import AgentState from llm_client import LLMClient from tools import CodeExecutor, TOOLS import json def main(): print( Mini Claude Code Agent 启动 ) # 1. 初始化组件 llm_client LLMClient(modelgpt-3.5-turbo) # 可根据需要更换模型 executor CodeExecutor() # 将工具函数绑定到字典 TOOLS[execute_python_code][function] executor.execute # 2. 获取用户任务 user_objective input(请输入你的编程任务描述: ).strip() # 示例任务: “写一个Python函数它接受一个整数n返回前n个斐波那契数列的列表。” # 3. 初始化Agent状态 state AgentState(objectiveuser_objective) # 初始化消息历史加入系统提示 state.message_history [ {role: system, content: llm_client.system_prompt}, {role: user, content: f任务目标: {user_objective}\n请开始你的第一步。} ] # 4. 主控制循环 step_counter 0 while state.should_continue(): step_counter 1 print(f\n--- 第 {step_counter} 步 ---) # a. 观察 思考调用LLM print([Agent] 正在思考...) llm_response_text llm_client.get_completion(state.message_history) llm_response llm_client.parse_response(llm_response_text) thought llm_response.get(thought, No thought provided.) action_spec llm_response.get(action) print(f[Thought] {thought}) # b. 行动执行工具调用如果有 action_result No action taken. if action_spec and isinstance(action_spec, dict): tool_name action_spec.get(name) tool_args action_spec.get(arguments, {}) if tool_name in TOOLS: print(f[Action] 调用工具: {tool_name}, 参数: {tool_args}) tool_func TOOLS[tool_name][function] try: # 实际调用工具 if tool_name execute_python_code: code tool_args.get(code, ) print(f[Code Execution]\npython\n{code}\n) result tool_func(code) action_result f工具 {tool_name} 执行结果:\n状态: {result[status]}\n输出:\n{result[output]} else: action_result f工具 {tool_name} 暂未实现完整功能。 except Exception as e: action_result f工具调用异常: {str(e)} else: action_result f未知工具: {tool_name} else: print([Action] 本轮无需调用工具。) print(f[Result] {action_result}) # c. 更新状态记忆 state.add_to_history(thought, str(action_spec), action_result) # d. 准备下一轮对话消息 # 将本轮的“助手”回复思考行动指令加入历史 state.message_history.append({role: assistant, content: llm_response_text}) # 将工具执行的结果作为“用户”的新输入加入历史模拟环境反馈 state.message_history.append({role: user, content: f上一步的结果是{action_result}\n请根据结果和原始任务目标决定下一步。如果任务已完成请明确说明。}) # e. 简单判断任务是否完成可根据结果内容做更智能的判断 if 任务完成 in thought or completed in thought.lower(): state.is_complete True print(\n[Agent] 报告任务已完成) break if step_counter 15: # 防止无限循环的安全阀 print(\n[Agent] 已达到最大步数限制停止执行。) break # 5. 输出最终总结 print(f\n 执行结束 ) print(f任务目标: {state.objective}) print(f完成状态: {成功 if state.is_complete else 未完成}) print(f总执行步数: {step_counter}) print(f最终失败计数: {state.failure_count}) print(\n执行历史摘要:) for record in state.execution_history[-5:]: # 打印最后5步 print(f 步骤{record[step]}: {record[thought][:100]}...) if __name__ __main__: main()这个主循环清晰地体现了 ReAct 范式。每一轮迭代Agent 都基于完整的对话历史包含之前的思考和结果进行“观察”然后输出“思考”和“行动”。我们解析行动并调用工具将结果反馈回对话历史从而开启下一轮。循环的终止条件由状态对象的should_continue()方法控制它检查任务完成标志和连续失败次数。4. 运行示例与深度解析让我们用一个具体的任务来驱动这个 Mini Agent并观察其内部运作。假设我们输入的任务是“写一个Python函数它接受一个整数n返回前n个斐波那契数列的列表。”运行python main.py并输入上述任务。以下是可能发生的交互流程的解析第一轮循环观察消息历史包含系统提示和用户任务。思考LLM 输出 JSONthought字段可能是“用户需要斐波那契数列函数。这是一个经典问题。我应该先规划1. 理解斐波那契定义F(0)0, F(1)1, F(n)F(n-1)F(n-2)。2. 编写函数框架。3. 先写一个简单版本并测试。第一步我先测试一下环境并生成一个基础函数。”行动action字段指示调用execute_python_code参数code包含一个简单的测试代码比如def fib(n): return [0, 1, 1, 2, 3][:n]或者甚至只是print(Hello from Code Agent)来验证执行环境。结果执行器返回成功输出Hello from Code Agent。更新这个结果被添加到历史和上下文中。第二轮循环观察现在历史包含了第一轮的思考和成功结果。思考“环境测试成功。现在开始实现真正的斐波那契函数。我需要处理边界情况n0。我将编写一个使用循环的函数。”行动调用工具执行类似以下的代码def fibonacci(n): if n 0: return [] elif n 1: return [0] fib_list [0, 1] for i in range(2, n): fib_list.append(fib_list[-1] fib_list[-2]) return fib_list[:n] # 确保长度精确为n print(fibonacci(10)) # 测试输出结果执行成功输出[0, 1, 1, 2, 3, 5, 8, 13, 21, 34]。更新成功结果被记录。第三轮循环观察历史显示函数已编写并测试了 n10 的情况。思考“函数运行成功输出了前10个斐波那契数看起来正确。但我还应该测试一些边界情况比如 n0, n1, n2以及一个稍大的数以确保健壮性。然后我可以宣布任务完成。”行动调用工具执行另一段测试代码# 假设上面的 fibonacci 函数已定义 test_cases [0, 1, 2, 5, 10] for tc in test_cases: print(ffibonacci({tc}) {fibonacci(tc)})结果执行成功输出所有测试用例的结果。更新成功结果被记录。LLM 在下一轮的思考中可能会判断所有测试通过任务目标已达成。第四轮循环观察历史显示核心函数和多种测试都已通过。思考“所有测试用例包括边界情况都通过了。函数逻辑正确代码完整。任务目标已实现。我可以提供最终的函数定义并结束任务。”行动action为null因为不需要再执行代码。结果无工具调用结果。终止thought中包含“任务完成”的表述主循环中的条件判断会捕获到这一点将state.is_complete设为True循环结束。通过这个流程你可以看到 Agent 并非一次性生成最终答案而是进行了一个计划、编码、测试、调试、再测试的完整软件工程微循环。这正是 AI Agent 自主性的体现。5. 关键问题排查与进阶优化在实际构建和运行过程中你肯定会遇到各种问题。这里我总结了一些常见坑点和对应的解决方案。5.1 LLM 输出格式不稳定问题即使使用了response_format{type: json_object}模型偶尔还是会返回非 JSON 或格式错误的文本。解决方案强化系统提示在系统提示词开头和结尾都强调“必须输出 JSON”。可以加入示例。实现解析重试机制在parse_response函数中如果解析失败可以将错误信息连同原响应一起送回给 LLM要求它纠正。这需要额外一轮 API 调用但能提高鲁棒性。使用输出解析库考虑使用像LangChain的OutputParser或Pydantic来定义严格的响应模型它们有更好的错误处理和重试逻辑。5.2 代码执行环境问题问题Docker 容器执行慢或某些库无法安装。解决方案使用预构建的镜像不要每次都从python:3.11-slim开始。可以预先构建一个包含常用数据科学库如 numpy, pandas的镜像并推送到你的私有仓库加速容器启动。实现代码缓存对于相同的代码片段可以哈希后缓存执行结果避免重复运行。超时与资源管理我们的CodeExecutor已经设置了超时和资源限制。根据任务调整timeout和mem_limit。对于长时间运行的任务需要考虑异步执行和状态轮询。5.3 Agent 陷入死循环或无效行动问题Agent 可能卡在某个步骤不断重复相似但错误的操作。解决方案丰富系统提示的约束明确告诉 Agent “如果连续三次尝试解决同一个错误都失败应该总结问题并向用户求助”。在状态中实现更复杂的终止逻辑除了连续失败计数还可以检查执行历史是否出现循环模式例如最近三步的“思考”内容高度相似。引入“反思”步骤在每 N 步之后或者在连续失败后强制插入一个“反思”阶段。让 LLM 回顾整个历史总结当前进展、遇到的障碍并重新规划后续步骤。这相当于给 Agent 一个“暂停并重新评估”的机会。5.4 扩展更多工具与能力基础版只实现了代码执行。一个实用的 Code Agent 还需要更多工具文件系统工具让 Agent 能创建main.py、utils.py等多文件项目。代码分析工具集成pylint、black进行代码质量和格式检查。网络搜索工具允许 Agent 在遇到未知 API 或库时自行搜索文档需谨慎并做好安全限制。Git 工具实现git add,git commit等让 Agent 能管理代码版本。添加新工具的关键是在TOOLS字典中注册提供清晰的描述和参数模式。在系统提示词中更新工具列表和说明。在主循环的action解析部分添加对新工具函数的调用。5.5 提升任务完成判断的智能度目前我们仅通过thought中是否包含“完成”字样来判断这非常脆弱。进阶方案最终输出验证在任务开始时让用户提供一个“验证标准”。例如对于斐波那契函数标准可以是“对于输入 7函数必须返回[0,1,1,2,3,5,8]”。Agent 在认为自己完成后必须运行这个验证测试只有通过才算真正完成。LLM 评估让另一个 LLM或同一 LLM 的不同调用作为“评审员”根据原始任务描述和最终生成的代码/输出判断任务是否被满意地完成。构建这个 Mini Claude Code 的过程就像在亲手搭建一个微型机器人。你定义了它的大脑LLM提示词、记忆状态管理、手脚工具集和控制逻辑主循环。每一个环节的调整都会直接影响它的行为和能力。通过这个项目你获得的不再是对 AI Agent 概念的模糊认知而是对其内部运作机制清晰、具象的理解。这为你后续学习更复杂的 Agent 框架如 LangChain、AutoGen甚至设计面向特定领域的生产级 Agent打下了最坚实的地基。