从零手写AI Agent：深入理解核心架构与Python实现

1. 项目概述：为什么我们要“徒手”造一个AI Agent？

最近AI Agent这个概念火得不行，好像不提Agent就落伍了。但打开各种教程和框架，动辄就是LangChain、AutoGen、CrewAI，配置复杂，概念一堆，对新手来说，光是理解这些框架的抽象层就得花上好几天。这让我想起早年学编程，从理解指针到写出第一个链表，那种“从无到有”的掌控感，是直接调用现成库无法替代的。

所以，我决定做一次“返璞归真”的尝试：不使用任何现成的AI Agent框架，仅用纯Python，从零开始手写一个具备基础能力的AI Agent。这个Agent的核心目标很简单：它能理解我的自然语言指令，调用我预先赋予它的工具（比如计算器、网络搜索模拟），并基于结果进行简单的推理和决策，最后用自然语言回复我。

你可能会问，有轮子为什么不用？原因有三：第一，为了彻底理解。框架封装了太多细节，就像开自动挡车，虽然快，但你不明白离合器、变速箱是如何协同工作的。亲手实现一遍，你会对Agent的“感知-规划-执行-学习”循环有刻骨铭心的理解。第二，为了极致的定制与控制。当你的需求非常特殊，或者需要对Agent的每一个决策环节进行审计和干预时，一个 stripped-down（精简到核心）的自建系统是无可替代的。第三，为了学习与教学。这是理解AI Agent架构最直观、最深刻的方式，没有之一。

这个项目适合所有对AI感兴趣，具备基础Python语法知识，想窥探AI Agent内部奥秘的开发者。我们不需要GPU，不需要复杂的深度学习知识，核心是逻辑与架构设计。接下来，我将带你一步步拆解、构建，并分享其中每一个“坑”与“闪光点”。

2. 核心架构设计：一个AI Agent的“五脏六腑”

在动手写代码之前，我们必须先想清楚，一个最简单的AI Agent应该由哪些核心部件构成。抛开那些花哨的名词，我们可以将其抽象为四个基本模块，它们构成了Agent的“大脑”和“四肢”。

2.1 大脑核心：大语言模型接口层

这是Agent的“认知中心”。虽然我们说“从零手写”，但并非要自己训练一个LLM，那是另一个维度的工程。我们这里指的是如何与一个现成的LLM API进行交互。我们将封装一个统一的类，来处理与LLM的对话、管理上下文（记忆）、解析返回结果。关键在于设计一个灵活的提示词模板系统，让LLM能按照我们设定的角色和格式进行思考与输出。

注意：选择LLM API时，优先考虑其响应的结构化能力（如支持JSON格式输出）和上下文长度。对于实验和学习，OpenAI的GPT-3.5-Turbo或 Anthropic 的 Claude Haiku 都是成本与效果平衡的好选择。国内的一些平台如百度文心、阿里通义千问也提供了类似的API。

2.2 技能仓库：工具调用系统

Agent的“四肢”就是它能使用的各种工具。一个只会聊天的AI不是Agent，能调用工具完成任务才是。我们需要设计一个工具注册与发现机制。每个工具都是一个Python函数，我们需要用一种方式（比如装饰器）来告诉Agent：“嗨，我这里有一个新工具，它的功能是XXX，调用时需要A、B两个参数”。当LLM大脑决定使用某个工具时，我们的系统要能准确地找到对应的函数，传入正确的参数，并执行它。

2.3 决策循环：推理与执行引擎

这是Agent的“中枢神经系统”，负责调度大脑和四肢。它的工作流程是一个经典循环：

1. 感知：接收用户输入。
2. 规划：LLM大脑分析输入，决定是否需要调用工具、调用哪个工具、参数是什么。
3. 执行：工具调用系统执行对应的函数。
4. 观察：获取工具执行的结果。
5. 再规划/输出：LLM大脑结合工具结果和对话历史，决定是继续调用工具，还是已经可以生成最终答案回复用户。

这个循环可能执行多次，直到任务完成。如何设计这个循环的状态机，如何让LLM的每次输出都能被稳定地解析成“思考”或“行动指令”，是这里的核心挑战。

2.4 记忆模块：对话上下文管理

Agent不能得鱼忘筌，它需要记住之前的对话和工具执行结果。这就是记忆模块。最简单的实现就是一个列表，按顺序存放每轮用户输入、AI思考、工具调用和结果。更高级的可以引入摘要记忆、向量数据库记忆等。在我们的初版中，一个能自动修剪长度的对话历史列表就足够了，重点是理解记忆在推理中的关键作用。

3. 分步实现：从零搭建每一块积木

理论说得再多，不如一行代码。让我们开始动手，我会详细解释每一行关键代码的意图和可能遇到的坑。

3.1 第一步：搭建与LLM对话的桥梁

首先，我们需要一个LLMClient类。这里以OpenAI API为例，但设计上要保持接口通用，以便日后切换模型。

import openai import json from typing import Dict, List, Optional, Any class LLMClient: def __init__(self, api_key: str, model: str = "gpt-3.5-turbo", temperature: float = 0.1): """ 初始化LLM客户端。 :param api_key: API密钥 :param model: 使用的模型名称 :param temperature: 温度参数，越低输出越确定，越高越有创造性。对于工具调用，建议调低。 """ self.client = openai.OpenAI(api_key=api_key) self.model = model self.temperature = temperature self.conversation_history: List[Dict[str, str]] = [] # 存储对话历史 def add_message(self, role: str, content: str): """向对话历史中添加一条消息。""" self.conversation_history.append({"role": role, "content": content}) def get_completion(self, messages: List[Dict[str, str]]) -> str: """调用LLM API获取补全结果。""" try: response = self.client.chat.completions.create( model=self.model, messages=messages, temperature=self.temperature, ) return response.choices[0].message.content except Exception as e: print(f"调用LLM API失败: {e}") return f"Error: {e}" def clear_history(self): """清空对话历史。""" self.conversation_history.clear()

关键点解析：

• temperature=0.1：对于需要稳定解析、执行工具调用的Agent，较低的温度值能减少输出的随机性，让模型更“听话”。
• conversation_history：我们用列表存储所有消息。一个更健壮的实现需要考虑上下文窗口长度，当历史消息太长时，需要智能地摘要或丢弃最早的消息，否则会触发API的token限制错误。

3.2 第二步：构建工具系统

工具系统的核心是一个“工具注册表”和一个统一的“工具执行器”。我们使用装饰器来优雅地注册工具。

class ToolRegistry: def __init__(self): self._tools: Dict[str, Dict] = {} # 工具名 -> 工具信息（函数、描述、参数模式） def register(self, name: str, description: str): """装饰器，用于注册一个工具。""" def decorator(func): # 获取函数的参数信息，用于后续生成提示词和参数验证 import inspect sig = inspect.signature(func) params = list(sig.parameters.keys()) self._tools[name] = { "function": func, "description": description, "parameters": params } return func return decorator def get_tool(self, name: str) -> Optional[Dict]: """根据名称获取工具信息。""" return self._tools.get(name) def list_tools(self) -> List[str]: """列出所有可用的工具名称和描述。""" return [f"{name}: {info['description']}" for name, info in self._tools.items()] # 全局工具注册表实例 tool_registry = ToolRegistry() # 示例：注册一个计算器工具 @tool_registry.register(name="calculator", description="执行简单的数学计算，支持加减乘除。") def calculator(expression: str) -> str: """计算数学表达式。注意：使用eval有安全风险，此处仅用于演示。""" try: # 警告：在生产环境中，直接eval用户输入是极度危险的！ # 这里应使用安全的表达式解析库（如 ast.literal_eval 配合自定义解析器）。 result = eval(expression, {"__builtins__": None}, {}) return f"计算结果: {result}" except Exception as e: return f"计算错误: {e}" # 示例：注册一个获取天气的模拟工具（真实情况需调用API） @tool_registry.register(name="get_weather", description="获取指定城市的模拟天气信息。") def get_weather(city: str) -> str: # 这里模拟一个API调用 weather_data = { "北京": "晴，25°C", "上海": "多云，28°C", "深圳": "雷阵雨，30°C" } return weather_data.get(city, f"未找到{city}的天气信息。")

实操心得与避坑指南：

1. 安全！安全！安全！：calculator工具中使用了eval，这在任何生产环境或接收不可信用户输入的场景下都是绝对禁止的！这里仅为了演示工具调用的流程。一个安全的计算器应该使用ast.literal_eval并严格限制可用的操作符，或者使用像numexpr这样的专用库。
2. 工具描述的魔力：给工具写一个清晰、准确的description至关重要。LLM大脑完全依赖这段描述来决定是否以及如何调用该工具。描述应包含功能、输入参数格式和输出示例。
3. 参数验证：当前实现只是简单存储了参数名。一个完善的系统应该在调用前验证参数的类型和值，这能极大减少LLM“幻觉”调用导致的错误。

3.3 第三步：设计Agent的核心推理循环

这是最核心的部分。我们需要设计一个Agent类，它整合LLM客户端和工具系统，并运行思考-行动循环。

class SimpleAgent: def __init__(self, llm_client: LLMClient, max_iterations: int = 5): self.llm = llm_client self.max_iterations = max_iterations # 防止无限循环 self.system_prompt = self._build_system_prompt() def _build_system_prompt(self) -> str: """构建系统提示词，定义Agent的角色和能力。""" tools_list = "\n".join(tool_registry.list_tools()) prompt = f""" 你是一个有帮助的AI助手，可以调用工具来解决问题。 你可以使用的工具如下： {tools_list} 你的思考过程必须遵循以下格式： 思考：[你的推理过程，分析用户问题，决定是否需要以及使用哪个工具] 行动：如果需要工具，则格式为 `工具名:参数1,参数2,...`。如果不需要工具，则直接输出最终答案。 答案：[只有当不需要工具或得到工具结果后，才给出给用户的最终答案] 示例1（需要工具）： 用户：计算一下123乘以456。 思考：用户需要计算乘法，我应该使用计算器工具。 行动：calculator:123*456 ...（系统执行工具并返回结果）... 思考：工具返回了结果56088。我可以将这个结果告知用户。 答案：123乘以456等于56088。 示例2（不需要工具）： 用户：你好！ 思考：这是一个简单的问候，不需要调用工具。 答案：你好！我是你的AI助手，有什么可以帮你的吗？ 现在，请开始你的任务。记住，每次只输出一个“思考”或“行动”或“答案”块。 """ return prompt def run(self, user_input: str) -> str: """运行Agent处理用户输入。""" print(f"\n用户: {user_input}") self.llm.add_message("user", user_input) for iteration in range(self.max_iterations): # 1. 准备对话上下文：系统提示 + 完整历史 messages = [{"role": "system", "content": self.system_prompt}] + self.llm.conversation_history # 2. LLM生成响应 llm_response = self.llm.get_completion(messages) print(f"AI原始响应: {llm_response}") # 3. 解析响应 thought, action, answer = self._parse_response(llm_response) if thought: print(f"思考: {thought}") self.llm.add_message("assistant", f"思考: {thought}") if action: # 解析行动指令 tool_name, *args = action.split(':') tool_info = tool_registry.get_tool(tool_name.strip()) if not tool_info: error_msg = f"错误：未知工具 '{tool_name}'" self.llm.add_message("system", error_msg) print(error_msg) continue # 执行工具 tool_func = tool_info['function'] try: # 简单处理：假设参数是以逗号分隔的字符串 # 更复杂的实现需要根据工具参数定义进行类型转换 args_list = [arg.strip() for arg in args[0].split(',')] if args else [] result = tool_func(*args_list) except Exception as e: result = f"工具执行出错: {e}" print(f"执行工具 `{tool_name}`，结果: {result}") # 将工具执行结果作为系统消息加入历史，供LLM下一轮参考 self.llm.add_message("system", f"工具`{tool_name}`返回: {result}") if answer: print(f"答案: {answer}") self.llm.add_message("assistant", f"答案: {answer}") return answer # 任务完成，返回最终答案 # 循环超过最大次数 timeout_msg = "抱歉，我尝试了多次仍未解决问题。" self.llm.add_message("assistant", timeout_msg) return timeout_msg def _parse_response(self, response: str) -> (str, str, str): """解析LLM的响应，提取思考、行动、答案。""" thought = action = answer = "" lines = response.strip().split('\n') for line in lines: if line.startswith('思考:'): thought = line[3:].strip() elif line.startswith('行动:'): action = line[3:].strip() elif line.startswith('答案:'): answer = line[3:].strip() return thought, action, answer

核心逻辑深度解析：

1. 系统提示词工程：_build_system_prompt方法是Agent的“灵魂注入”。它定义了Agent的角色、可用工具、最重要的输出格式以及示例。清晰的格式要求（思考/行动/答案）是让LLM稳定输出的关键，这比让它自由发挥可靠得多。
2. 循环与状态机：run方法实现了一个简单的状态机。只要没有输出答案:，并且迭代次数未超限，就会持续进行“思考-行动-观察”的循环。每次循环都将之前的对话和工具结果作为上下文喂给LLM。
3. 解析器的脆弱性：_parse_response函数非常简陋，它依赖于LLM严格遵守格式。在实际中，LLM偶尔会“不听话”，输出格式混乱的文本。更健壮的做法是使用LLM的“函数调用”功能（如果API支持），或者要求LLM输出严格的JSON格式，然后用json.loads解析，容错性会高很多。

3.4 第四步：组装并运行你的第一个Agent

现在，让我们把所有部件组装起来，看看这个“手搓”的Agent能否工作。

def main(): # 1. 初始化LLM客户端（请替换为你的真实API Key） API_KEY = "your-openai-api-key-here" # 务必替换！ llm_client = LLMClient(api_key=API_KEY, model="gpt-3.5-turbo") # 2. 创建Agent agent = SimpleAgent(llm_client=llm_client) # 3. 运行测试 test_queries = [ "你好，请介绍一下你自己。", "请问北京今天的天气怎么样？", "帮我计算一下(15 + 27) * 3 等于多少？", "先查一下深圳的天气，然后计算如果温度是30度，相当于多少华氏度？公式是 F = C * 9/5 + 32" ] for query in test_queries: final_answer = agent.run(query) print(f"最终回复: {final_answer}\n{'-'*50}") if __name__ == "__main__": main()

运行这段代码，你会看到类似以下的输出（具体内容因模型随机性略有不同）：

用户: 帮我计算一下(15 + 27) * 3 等于多少？ AI原始响应: 思考：用户需要一个数学计算，我可以使用计算器工具。 行动：calculator:(15 + 27) * 3 思考: 用户需要一个数学计算，我可以使用计算器工具。 执行工具 `calculator`，结果: 计算结果: 126 AI原始响应: 思考：工具返回了结果126。我可以直接给出答案。 答案: (15 + 27) * 3 的计算结果是126。 答案: (15 + 27) * 3 的计算结果是126。 最终回复: (15 + 27) * 3 的计算结果是126。

看，它成功地识别了计算需求，调用了正确的工具，并给出了答案！对于最后一个需要多步推理（先查天气，再换算温度）的复杂问题，一个设计良好的Agent应该能通过多次循环来完成。

4. 进阶优化与问题深度排查

一个能跑起来的Demo只是起点。要让这个手写Agent真正可用、健壮，我们还需要解决一系列工程问题。

4.1 如何让LLM的输出更稳定？—— 结构化输出的艺术

我们之前依赖文本解析，这很脆弱。更优解是使用LLM API的结构化输出功能（如OpenAI的JSON Mode）或函数调用功能。

方案升级：使用JSON Mode修改LLMClient.get_completion和系统提示词，要求LLM始终返回一个JSON对象。

def get_structured_completion(self, messages, response_format={"type": "json_object"}): try: response = self.client.chat.completions.create( model=self.model, messages=messages, temperature=self.temperature, response_format=response_format # 指定JSON格式 ) json_str = response.choices[0].message.content return json.loads(json_str) # 直接解析为字典 except json.JSONDecodeError as e: print(f"JSON解析失败: {e}, 原始响应: {json_str}") return {"error": "Invalid JSON response"}

同时，系统提示词要明确要求输出JSON，例如：

请始终以以下JSON格式回复： { "thought": "你的推理过程", "action": {"tool_name": "工具名", "parameters": ["参数1", "参数2"]}, "answer": "给用户的最终答案" } 其中，`action`和`answer`不会同时存在。

这样，在_parse_response中，我们直接处理字典，彻底告别字符串解析的噩梦。

4.2 工具调用参数如何更智能？—— 从字符串到类型化

目前的工具参数传递是简单的字符串分割，无法处理复杂参数（如列表、字典）。我们可以结合Python的inspect模块和LLM的函数调用描述来实现。

思路：为每个工具生成一个符合OpenAI函数调用规范的描述（包含参数类型）。然后，在调用LLM时，将这些描述传入，并启用function_call功能。LLM会返回一个结构化的函数调用请求，其中参数已经是解析好的。这需要更深入地集成特定LLM API的高级功能，是通往生产级Agent的必经之路。

4.3 记忆管理：如何突破上下文长度限制？

随着对话和工具调用轮次增加，conversation_history会越来越长，最终超过模型的上下文窗口（如GPT-3.5的16K）。解决方案有几种：

1. 滑动窗口：只保留最近N轮对话。简单但会丢失早期关键信息。
2. 智能摘要：当历史达到一定长度，让LLM自己生成一个当前对话的简短摘要，然后用“系统消息：之前的对话摘要：...” + “最近几轮实际对话”来替代冗长的完整历史。这需要额外的LLM调用和提示词设计。
3. 向量数据库记忆：将每轮对话的关键信息（如事实、用户偏好）转化为向量存入数据库（如ChromaDB）。每次需要回忆时，用当前问题去检索最相关的记忆片段。这是实现“长期记忆”的先进方式，但架构复杂度陡增。

对于我们的手写Agent，可以先实现滑动窗口，并设置一个最大历史长度阈值。

4.4 常见问题排查速查表

在实际运行中，你几乎一定会遇到以下问题。这里提供一个快速排查指南：

5. 从玩具到工具：扩展你的手写Agent

掌握了基础架构后，你可以像搭乐高一样，为你的Agent添加更多强大功能：

• 多模态能力：让Agent不仅能处理文本，还能“看”图“听”音。这需要在工具系统中集成图像识别（如CLIP）或语音转文本（如Whisper）的API调用。
• 网络搜索能力：注册一个web_search工具，内部调用Serper API或SearxNG，让Agent能获取实时信息，不再局限于训练数据。
• 持久化与状态管理：将Agent的对话历史、学到的知识（如用户偏好）保存到文件或数据库中，下次启动时可以加载。
• 多Agent协作：创建多个具有不同专长（如研究员、写手、校对员）的Agent实例，让它们通过一个“协调者”Agent来共同完成复杂项目。这本质上就是构建一个微型的CrewAI或AutoGen。

手写一个AI Agent的过程，就像在显微镜下观察一个生命体的运作。每一个循环，每一次工具调用，每一次记忆的存取，都清晰可见。你可能会为它偶尔的“愚蠢”而抓狂，也会为它灵光一现完成复杂任务而欣喜。这种深度的理解和掌控感，是使用高级框架无法给予的。

这个项目的代码只是一个起点，它简陋但完整地揭示了一个AI Agent的核心骨架。当你亲手调试它、扩展它、看着它从踉跄学步到逐渐稳健时，你对AI Agent的理解将不再停留在概念和API文档层面。你会明白，所谓智能体，其内核无非是在确定性的工具世界与概率性的语言模型之间，搭建起一座可靠沟通的桥梁。而这座桥的每一块砖，都由你亲手烧制。