1. 项目缘起为什么我们要“徒手”造一个AI Agent最近AI Agent这个概念火得不行感觉是个技术分享会或者技术文章不提两句Agent就跟不上时代了。市面上也确实涌现了像LangChain、AutoGPT、CrewAI这样优秀的框架它们封装了大量工具调用、记忆管理、任务拆解的复杂逻辑让开发者能快速搭建起一个看起来挺智能的“代理”。但不知道你有没有过这种感觉用框架搭出来的东西跑起来是能跑但一旦出了问题或者想深度定制某个行为就感觉像在碰一个黑盒调试起来无从下手文档翻遍了也找不到那个关键的“开关”。这正是我决定抛开所有现成框架从零开始手写一个AI Agent的核心动机。这绝不是为了标新立异或者重复造轮子而是想通过这个“笨办法”彻底搞明白一个AI Agent到底是怎么“思考”和“行动”的。它的“大脑”大模型是如何被调用的它如何记住之前的对话又如何决定下一步是去查资料、写代码还是直接回答当你亲手用最基础的代码把这些流程串起来之后你对Agent的理解会从“用户”层面深入到“设计者”层面。以后再遇到任何Agent相关的复杂问题你脑子里会自然浮现出它的工作流程图而不是一堆模糊的API调用。这个过程很像学编程时不直接上SpringBoot而是先手写一个简单的IoC容器学数据结构时不满足于调用HashMap而是去琢磨它的哈希冲突解决和扩容机制。它锻炼的是一种“第一性原理”的思维。所以这篇文章就是一次这样的实践记录。我们将只用Python标准库和OpenAI的API作为“大脑”从定义一个最简单的Agent循环开始一步步构建起它的记忆、工具使用和决策能力。你会发现Agent的核心逻辑远没有想象中那么神秘。2. 核心蓝图一个最小可运行AI Agent的解剖图在动手写代码之前我们必须先在大脑里勾勒出这个AI Agent的骨架。一个能独立完成任务的Agent无论框架如何包装都离不开以下几个核心组件它们共同构成了一个经典的“感知-思考-行动”循环。大脑 (The Brain): 这是Agent的智能核心通常是一个大语言模型。它负责理解用户的输入感知结合上下文进行推理思考并生成下一步的指令或回答。在我们的手写版本里它就是通过API调用的OpenAI GPT模型。我们发送给它一段精心设计的提示词它返回文本形式的“想法”。记忆 (Memory): Agent不能是“金鱼”它必须能记住之前发生过的事情。记忆分为短期和长期。短期记忆通常指当前对话的上下文即我们发送给大模型的那一串消息历史。长期记忆则可以更复杂比如一个向量数据库用于存储和检索过去的经验。初始版本我们先实现一个简单的对话历史记忆。工具 (Tools): 这是Agent与外部世界交互的“手脚”。大模型本身无法获取实时信息、无法执行计算、无法操作文件系统。工具就是赋予它这些能力的函数。例如一个搜索工具、一个计算器工具、一个读写文件的工具。Agent在思考后如果认为需要可以“决定”调用某个工具。执行器 (Executor): 也可以称为“运行时”或“循环控制器”。这是驱动整个Agent运转的引擎。它负责管理整个工作流接收用户输入将其与记忆组合成提示词调用大脑解析大脑的输出判断是否需要调用工具如果需要则调用工具并获取结果再将结果反馈给大脑进行下一轮思考直到大脑认为可以给出最终答案再由执行器输出给用户。它们之间的关系可以用一个简单的循环来描述接收输入用户提出请求。组织上下文执行器从记忆模块获取历史对话和当前输入一起组装成发送给大脑的提示词。大脑思考调用大模型API获得模型的回复文本。解析与决策执行器解析模型回复。如果回复中包含调用工具的指令则进入步骤5否则直接进入步骤6。执行工具根据解析出的指令调用对应的工具函数并获得工具执行的结果。更新与循环将工具执行的结果或模型的直接回复作为新的上下文追加到记忆对话历史中。如果需要调用工具则带着工具的结果跳回步骤3开始新一轮思考如果不需要则跳转至步骤7。输出结果将模型的最终回复返回给用户并更新记忆。这个循环就是Agent自主性的来源。下面我们就将这个蓝图转化为具体的代码。3. 从零搭建逐步实现Agent的每一个器官我们将采用自底向上的方式构建先实现基础部件最后组装成完整的Agent。请确保你有一个可用的OpenAI API密钥。3.1 第一步构建“大脑”——与大模型对话大脑的核心功能是接收一段提示词返回模型的思考结果。我们将其封装成一个类便于管理API密钥、模型选择等配置。import openai from typing import List, Dict, Any class OpenAIBrain: Agent的大脑封装与OpenAI API的交互。 为了简化我们暂时只使用ChatCompletion接口。 def __init__(self, api_key: str, model: str gpt-3.5-turbo): openai.api_key api_key self.model model # 系统提示词用于设定Agent的角色和基础行为准则 self.system_prompt 你是一个有帮助的AI助手。你可以思考并且在需要时使用工具来帮助你完成任务。 如果你需要使用工具请严格按照以下格式回复 思考[你的推理过程] 行动工具名称(参数1, 参数2, ...) 例如行动search_web(什么是量子计算) 如果你不需要使用工具或者已经通过工具获得了足够信息请直接给出最终答案。 最终答案[你的回答] def think(self, messages: List[Dict[str, str]]) - str: 核心思考函数。将消息历史发送给大模型并返回它的回复文本。 # 在消息列表开头插入系统提示 conversation [{role: system, content: self.system_prompt}] messages try: response openai.ChatCompletion.create( modelself.model, messagesconversation, temperature0.7, # 控制创造性对于任务执行可以调低 max_tokens500 ) return response.choices[0].message[content].strip() except Exception as e: return f思考过程出错: {e}这里有几个关键点系统提示词这是“调教”Agent行为的关键。我们明确规定了它何时以及如何调用工具行动工具名(...)并要求它在不需要工具或得到答案后以最终答案开头回复。这种结构化的输出约定是后续解析的基础。消息格式遵循OpenAI Chat API的格式每条消息是一个字典包含rolesystem,user,assistant和content。错误处理简单的try-catch在实际项目中需要更健壮的处理比如重试、降级等。3.2 第二步赋予“记忆”——记住对话的历史我们实现一个简单的对话历史记忆它只保存最近的若干轮对话防止上下文过长。class SimpleMemory: 简单的对话记忆使用列表保存消息历史。 可以添加限制最大轮数的功能以防止上下文过长。 def __init__(self, max_turns: int 10): self.messages [] # 格式[{role: user, content: ...}, {role: assistant, content: ...}] self.max_turns max_turns * 2 # 因为一轮包含user和assistant两条消息 def add_message(self, role: str, content: str): 添加一条消息到历史记录 self.messages.append({role: role, content: content}) # 如果超出最大限制从头部移除最旧的消息 if len(self.messages) self.max_turns: self.messages self.messages[-self.max_turns:] def get_context(self) - List[Dict[str, str]]: 获取当前的完整对话上下文 return self.messages.copy() def clear(self): 清空记忆 self.messages []这个记忆模块非常简单就是维护一个列表。add_message方法负责追加并管理长度get_context方法在组织提示词时被大脑调用。在更复杂的Agent中这里可能会接入向量数据库实现基于语义的长期记忆检索。3.3 第三步打造“工具”——Agent的手和脚工具是函数我们需要一种方式让Agent知道有哪些工具可用以及如何调用它们。我们创建一个工具注册表。class ToolRegistry: 工具注册表。管理所有可用的工具并提供调用接口。 def __init__(self): self.tools {} # 工具名 - 工具函数 def register(self, name: str, func: callable, description: str ): 注册一个工具 self.tools[name] { function: func, description: description } def execute(self, tool_name: str, *args, **kwargs) - str: 执行一个工具 if tool_name not in self.tools: return f错误未找到工具 {tool_name}。 try: result self.tools[tool_name][function](*args, **kwargs) return str(result) except Exception as e: return f工具执行出错: {e} def get_tools_description(self) - str: 生成工具描述文本用于拼接到系统提示词中让Agent知道有哪些工具可用 desc [] for name, info in self.tools.items(): desc.append(f- {name}: {info[description]}) return \n.join(desc) # 定义几个示例工具 def calculator(expression: str) - str: 计算一个数学表达式。例如calculator(3 5 * 2) try: # 警告使用eval有安全风险此处仅用于演示。生产环境应用安全计算库。 result eval(expression, {__builtins__: {}}, {}) return f计算结果: {result} except Exception as e: return f计算错误: {e} def get_current_time(*args) - str: 返回当前的日期和时间。 from datetime import datetime return f当前时间是: {datetime.now().strftime(%Y-%m-%d %H:%M:%S)} def search_web(query: str) - str: 模拟网络搜索。在实际应用中这里会调用真正的搜索API。 # 这里只是一个模拟返回 simulated_results { python: Python是一种高级、解释型的通用编程语言。, 天气: 模拟天气北京晴25摄氏度。, 量子计算: 量子计算是利用量子力学原理如叠加和纠缠进行数据处理的计算范式。 } return simulated_results.get(query.lower(), f未找到关于 {query} 的模拟信息。)关键设计解析注册机制ToolRegistry是一个中心化的管理器。任何函数只要通过register方法注册就成为了Agent可用的工具。这比在代码中硬编码工具列表要灵活得多。工具描述get_tools_description方法生成一段文本描述。这个描述必须被动态地插入到大脑的system_prompt中这样Agent在思考时才知道自己“拥有”哪些能力。这是连接工具定义与Agent认知的桥梁。安全警告calculator工具使用了eval这在演示中可行但在真实、开放的环境中极其危险因为它允许执行任意代码。实际项目中必须使用像ast.literal_eval这样的安全解析器或者自己实现一个算术表达式解析器。3.4 第四步组装“躯干”——创建主循环与执行器这是最核心的部分它将大脑、记忆和工具连接起来并实现之前描述的推理循环。import re class SimpleAgent: 简单的AI Agent执行器。 def __init__(self, brain: OpenAIBrain, memory: SimpleMemory, tool_registry: ToolRegistry): self.brain brain self.memory memory self.tools tool_registry # 更新大脑的系统提示词加入具体的工具描述 self.brain.system_prompt f\n\n你可以使用的工具有\n{self.tools.get_tools_description()} def parse_model_response(self, response: str) - tuple: 解析模型的回复。 返回一个元组(类型, 内容)。 类型可以是 thought, action, final_answer。 # 匹配最终答案 final_match re.search(r最终答案\s*(.*), response, re.DOTALL) if final_match: return (final_answer, final_match.group(1).strip()) # 匹配工具调用动作 action_match re.search(r行动\s*(\w)\s*\((.*)\), response) if action_match: tool_name action_match.group(1) # 解析参数这是一个简化版假设参数是逗号分隔的字符串 args_str action_match.group(2).strip() # 更健壮的解析应该处理引号、嵌套等这里简单按逗号分割并去除空格 args [arg.strip().strip(\\) for arg in args_str.split(,)] if args_str else [] return (action, (tool_name, args)) # 如果既不是最终答案也不是行动则视为纯思考或对话 return (thought, response) def run(self, user_input: str) - str: 运行Agent的主循环。 print(f用户: {user_input}) # 1. 将用户输入存入记忆 self.memory.add_message(user, user_input) max_iterations 5 # 防止无限循环 for i in range(max_iterations): # 2. 获取当前对话上下文 context self.memory.get_context() # 3. 让大脑思考 model_response self.brain.think(context) print(f\n[Agent第{i1}轮思考]) print(f大脑原始输出: {model_response}) # 4. 解析大脑的输出 resp_type, content self.parse_model_response(model_response) if resp_type final_answer: # 获得最终答案循环结束 final_answer content self.memory.add_message(assistant, final_answer) print(f最终答案: {final_answer}) return final_answer elif resp_type action: # 需要执行工具 tool_name, tool_args content print(f决定调用工具: {tool_name}, 参数: {tool_args}) # 5. 执行工具 tool_result self.tools.execute(tool_name, *tool_args) print(f工具执行结果: {tool_result}) # 将工具执行结果作为一条“系统”或“用户”消息加入记忆供下一轮思考使用 # 这里我们将其模拟为“用户”提供的新信息 self.memory.add_message(user, f[工具 {tool_name} 的结果] {tool_result}) elif resp_type thought: # 纯思考内容可以将其加入记忆作为助理的“自言自语”也可以不加入。 # 这里我们选择加入让思考过程也形成上下文。 self.memory.add_message(assistant, content) # 如果模型只是思考而没有行动或最终答案我们需要手动提示它继续。 # 更优的做法是在系统提示词中要求它必须输出行动或最终答案。 # 这里简单处理添加一个用户提示推动它继续。 self.memory.add_message(user, 请根据你的思考继续下一步使用工具或给出答案。) else: # 未知响应类型 error_msg 无法解析模型的响应。 self.memory.add_message(user, error_msg) # 如果循环达到最大次数仍未得到最终答案 timeout_msg 抱歉我在处理您的问题时思考过久未能得出最终结论。 self.memory.add_message(assistant, timeout_msg) return timeout_msg循环逻辑深度解析初始化与提示词增强在__init__中我们将工具描述动态添加到了大脑的系统提示词里。这是至关重要的一步它让模型知道了自己可以调用哪些工具及其功能。解析器parse_model_response函数使用正则表达式匹配模型输出。这是整个Agent稳定性的关键。我们约定了行动工具名(参数)和最终答案的格式解析器必须准确识别。在实际应用中可以要求模型输出JSON等更结构化的格式来降低解析难度。主循环run方法实现了完整的“思考-行动”循环。防呆设计设置了max_iterations这里为5防止模型陷入无限循环的“思考-调用-再思考”中。工具结果反馈将工具执行结果以[工具 X 的结果] ...的格式作为新的用户消息加入历史。这模拟了Agent通过工具感知到的世界变化并基于此进行下一轮思考。这个格式可以根据需要调整。纯思考处理如果模型只输出了思考过程thought我们将其记录后主动添加一个用户消息推动它继续。更好的做法是在系统提示词中强制要求模型在每轮输出中必须包含行动或最终答案。状态管理所有的交互状态都保存在memory.messages中。每一轮循环我们都将最新的上下文包含所有历史对话和工具结果发送给模型这使得模型具备了“记忆”能力。3.5 第五步让Agent跑起来——完整的示例代码现在我们把所有部件组装起来并运行一个完整的示例。def main(): # 0. 配置你的OpenAI API Key OPENAI_API_KEY 你的-OpenAI-API-Key # 1. 初始化各个组件 brain OpenAIBrain(api_keyOPENAI_API_KEY, modelgpt-3.5-turbo) memory SimpleMemory(max_turns5) tool_registry ToolRegistry() # 2. 注册工具 tool_registry.register(calculator, calculator, 计算一个数学表达式例如3 5 * 2。) tool_registry.register(get_current_time, get_current_time, 获取当前的日期和时间。) tool_registry.register(search_web, search_web, 模拟网络搜索获取信息。) # 3. 创建Agent agent SimpleAgent(brain, memory, tool_registry) # 4. 运行测试 queries [ 北京现在的天气怎么样, 计算一下 (15 7) * 3 等于多少, 先查一下量子计算是什么然后告诉我它和传统计算的主要区别。 ] for query in queries: print(\n *50) print(f处理查询: {query}) print(*50) final_response agent.run(query) print(*50) # 清空记忆开始下一个独立对话可选 # memory.clear() if __name__ __main__: main()运行这段代码你将在控制台看到Agent完整的思考过程。对于第三个复杂问题它会先调用search_web工具获取“量子计算”的信息然后将搜索结果作为上下文进行第二轮思考最终给出一个结合了搜索结果的答案。这就是一个具备自主工具调用能力的AI Agent的完整工作流程。4. 从玩具到工具手写Agent的进阶思考与优化完成一个基础可运行的Agent只是第一步。要让它在实际项目中发挥作用我们还需要考虑很多工程化和优化问题。这部分才是从“知道原理”到“能用好用”的关键。4.1 解析器的稳健性告别脆弱的正则表达式我们之前用正则表达式r行动\s*(\w)\s*\((.*)\)来解析工具调用。这非常脆弱。如果模型输出的括号不匹配、参数里包含逗号或引号解析就会失败。更可靠的做法是结构化输出在调用大模型时使用function calling如果API支持或要求模型输出严格的JSON格式。例如系统提示词可以改为“请以以下JSON格式回复{thought: ..., action: {name: ..., args: [...]}, final_answer: ...}”。然后在解析时使用json.loads。使用Pydantic模型验证即使输出JSON其结构也可能不对。可以定义一个Pydantic模型来描述期望的响应结构在解析后进行验证和类型转换这能极大提高代码的健壮性。后备与重试当解析失败时不应直接报错退出。可以将解析失败的信息反馈给模型例如“你上次的回复格式不正确请严格按照要求输出”并让其重试。这需要在主循环中增加相应的错误处理逻辑。4.2 记忆系统的演进从列表到向量数据库SimpleMemory只保留了最近的对话这对于长篇幅、多回合的复杂任务远远不够。真正的长期记忆需要解决两个问题存储容量和相关性检索。分窗口记忆可以维护多个记忆窗口如“超短期”最后几轮对话、“短期”当前会话主题、“长期”跨会话的持久化存储。大脑在思考时从不同的窗口中抽取最相关的信息组合成上下文。向量记忆这是目前的主流方案。将对话中的关键信息如用户的问题、助理的结论、工具的结果转换成向量Embedding存储到向量数据库如Chroma、Pinecone、Weaviate中。当新问题到来时将问题也转换成向量并在数据库中搜索最相似的过往记忆片段将其作为上下文注入。这模拟了人类的“联想记忆”。记忆摘要对于非常长的对话可以在每N轮后让大模型自动生成一个对话摘要然后将摘要存入长期记忆并清空或压缩短期记忆列表。这样可以保留核心信息同时节省上下文令牌。4.3 工具调用的高级模式并行、流式与验证我们实现的工具调用是串行、同步的。在实际场景中可能需要更复杂的模式并行工具调用有些任务可以同时进行。例如Agent可以同时调用“获取天气”和“查询航班”两个工具。最新的OpenAI API已经支持在单次请求中要求模型并行调用多个函数。在我们的手写框架中可以通过解析出多个行动指令然后使用asyncio或线程池来并发执行。流式思考为了让用户体验更好可以支持流式输出模型的“思考”过程就像ChatGPT那样一个字一个字地显示。这需要处理OpenAI API的流式响应。参数验证与类型转换在ToolRegistry.execute中我们直接传递字符串参数。更安全的做法是在注册工具时同时注册其参数的类型如int,str,bool。在执行前先进行类型转换和验证。例如calculator工具期望一个字符串表达式但如果模型传入了数字可以尝试转换。工具链一个工具的输出可以作为另一个工具的输入。这需要更复杂的规划能力。可以在系统提示词中说明工具之间的关系或者设计一个更高级的“规划模块”让模型先输出一个步骤计划再逐步执行。4.4 规划与反思让Agent更“智能”基础的Agent是反应式的收到输入思考行动再思考。更高级的Agent应该具备规划和反思能力。任务分解面对复杂任务如“为我制定一个周末旅行计划”Agent应该能先将其分解为子任务查目的地天气、找酒店、排行程。这可以通过在系统提示词中要求模型“先制定一个计划”来实现或者单独调用一个“规划模型”。自我反思Agent在执行完一系列步骤后应该能评估结果是否达到了目标。如果没有它应该能调整策略。这可以通过在循环中引入一个“反思”步骤来实现。例如在得到最终答案前让模型先回答“当前的结果是否已充分解答用户问题如果否还缺少什么信息下一步应该做什么”试错与学习理论上Agent可以将成功和失败的经验存储到长期记忆中在未来遇到类似任务时参考。这涉及到更复杂的强化学习机制但在我们的框架中可以通过精心设计记忆的存储和检索来初步实现。5. 避坑指南手写Agent过程中常见的“坑”与对策在亲手实现和迭代这个Agent的过程中我踩过不少坑。这里分享几个最典型的希望能帮你绕过去。坑一提示词工程是成败的关键但极易失控现象Agent要么不调用工具要么乱调用工具或者输出的格式总是不对导致解析失败。根因系统提示词system_prompt写得不够清晰、具体或者与模型的能力不匹配。不同的模型如GPT-3.5-Turbo和GPT-4对指令的遵循能力有差异。对策分步调试不要一次性写很长的复杂提示词。先让模型能正确识别“需要调用工具”的场景。可以先用一个简单的例子测试“如果我问你一个需要计算的问题请调用计算器工具。”提供大量示例在提示词中使用Few-Shot示例。直接给出2-3个完整的、格式正确的输入输出对。例如用户123乘以456等于多少 助理思考这是一个数学计算问题我需要使用计算器工具。 行动calculator(123 * 456)这比单纯用文字描述格式有效得多。结构化输出如前所述强烈建议使用JSON等结构化输出格式并在提示词中提供JSON Schema示例。迭代优化将提示词单独保存在一个文件中方便修改和版本管理。根据测试结果不断调整措辞和示例。坑二上下文管理不当导致成本激增或信息丢失现象对话轮次一多API调用费用飙升或者Agent“忘记”了很早之前的重要信息。根因无限制地将所有历史对话都塞进上下文。OpenAI的API按Token收费上下文越长越贵。而且模型对上下文中间部分的信息关注度会下降。对策设置合理的记忆窗口像我们的SimpleMemory一样限制保存的对话轮数。关键信息摘要在对话中主动让模型对当前讨论的核心点进行总结并将摘要作为一条独立消息存入记忆替代冗长的原始对话。向量检索作为补充对于需要长期记忆的知识采用“向量数据库摘要”的方式。只将最重要的信息片段向量化存储需要时通过检索召回而不是把全部历史都放进上下文。坑三工具执行的安全性与可靠性隐患现象Agent调用的工具导致系统文件被误删、服务器负载过高或者因为网络问题长时间挂起。根因工具函数没有进行权限控制、输入验证和超时处理。对策最小权限原则工具函数运行在沙箱环境或受限权限下。特别是文件操作、系统命令执行类工具。严格的输入清洗与验证对所有传入工具的参数进行类型检查、长度限制、危险字符过滤。绝对不要相信模型直接输出的参数。超时与重试机制为每个工具调用设置超时时间。对于可能失败的操作如网络请求实现指数退避的重试逻辑。用户确认对于高风险操作如删除文件、发送邮件可以让Agent先征求用户确认再将确认结果作为参数调用工具。坑四循环失控陷入死循环现象Agent在“思考-调用工具-再思考”的循环中出不来永远无法输出最终答案。根因模型可能陷入逻辑怪圈或者工具返回的结果始终无法让它满意。对策强制循环上限像我们代码中的max_iterations一样这是必须的保险丝。超时总结在达到循环上限后不是简单报错。可以让模型基于已有的所有信息强制输出一个当前能给出的最佳答案或阶段性总结。反思中断在每轮循环后可以引入一个简单的“反思”步骤判断是否进展停滞。例如检查最近两轮的工具调用和思考内容是否高度重复。手写一个AI Agent的过程是一个绝佳的深度学习机会。它迫使你去关注每一个细节从提示词设计、API调用、状态管理到错误处理。当你成功运行起第一个自己打造的Agent并看着它调用工具完成任务时那种对技术原理的透彻理解所带来的成就感是单纯使用框架无法比拟的。这个简单的骨架已经包含了智能体最核心的思想。你可以在此基础上根据自己的需求为其添加更强大的记忆、更丰富的工具、更稳健的循环逻辑逐步构建出属于你自己的、功能独特的AI智能体。