1. 项目概述为什么我们要亲手写一个AI智能体最近“AI智能体”这个概念火得不行感觉一夜之间从技术论坛到产品发布会人人都在谈Agent。但说实话很多讨论都飘在天上什么“自主智能”、“任务分解”、“工具调用”一堆高大上的术语砸下来新手听完更懵了。作为一个写过不少代码、也踩过不少坑的老手我始终相信理解一个复杂系统最好的方式就是亲手把它最核心的部分实现一遍。所以我决定用大约100行Python代码带大家从零开始构建一个AI智能体的核心循环。这个核心循环你可以把它想象成智能体的“大脑”或“发动机”。它不涉及复杂的大模型微调也不依赖庞大的外部工具链就是最纯粹、最本质的那部分逻辑感知、思考、决策、行动、再学习。通过实现这个循环你会彻底明白一个智能体是如何理解你的指令的它怎么知道自己下一步该做什么所谓的“自主”究竟体现在哪里以及我们该如何设计它的“记忆”和“反思”能力这不仅仅是一个教学演示。当你掌握了这个核心骨架再去学习LangChain、AutoGPT、Dify这些成熟的智能体平台时你会一眼看穿它们华丽外衣下的本质知道每个模块对应着我们今天代码里的哪一部分。这对于想自己动手开发智能应用或者想在团队里主导AI项目落地的开发者来说是至关重要的一课。接下来我们就从一行代码开始把这个“大脑”搭建起来。2. 智能体核心循环的架构设计在动手写代码之前我们必须先把架构想清楚。一个能“自主”工作的智能体绝对不是简单地把用户问题丢给大模型然后回复就完事了。它需要有一个持续运转的内部控制流程。经过对主流框架的拆解和提炼我将其核心循环抽象为以下五个关键组件它们共同构成了智能体的“思维闭环”。2.1 五大核心组件解析1. 记忆模块这是智能体的“经验库”。它需要记住两样东西一是与用户的对话历史这是短期记忆用于理解当前对话的上下文二是从过往任务中总结出的经验、知识或教训这是长期记忆。在我们的简易实现中我们会用一个列表来存储对话历史这就是它的短期记忆。更复杂的系统可能会引入向量数据库来存储和检索长期记忆。2. 规划模块这是智能体的“思考系统”。当它接收到一个任务比如“帮我查一下北京明天天气然后告诉我是否需要带伞”时它不能直接去执行。规划模块负责将这个复杂任务分解成一系列可执行的子步骤Step 1: 确定城市“北京”和时间“明天”Step 2: 调用天气查询工具Step 3: 分析查询结果判断降水概率Step 4: 生成最终建议。这个分解过程通常由大模型根据预设的提示词来完成。3. 工具模块这是智能体的“双手”。智能体本身不会凭空变出天气数据它需要调用外部工具。工具可以是一个简单的函数比如计算器也可以是一个复杂的API比如搜索引擎、数据库查询。在我们的代码里我们会预先定义几个简单的工具函数并告诉大模型这些工具的存在和用法。4. 执行模块这是智能体的“行动单元”。它根据规划模块产出的当前步骤去工具模块里找到对应的工具并传入正确的参数来执行它。比如执行模块会调用“天气查询工具”并传入参数city“北京”。执行后它会得到结果如“明天北京小雨降水概率60%”。5. 反思与循环控制模块这是智能体的“监督与调度中心”是整个循环的灵魂。它负责检查执行结果任务完成了吗如果完成了就整理最终答案返回给用户。如果没完成或者执行出错了怎么办反思模块会评估当前状况将最新的结果和历史对话一起再次交给规划模块去生成下一个步骤。如此循环直到任务完成或达到最大尝试次数。这个过程模拟了人类“尝试-观察-调整”的思考方式。注意这个架构是一个高度简化的模型真实项目中的模块边界可能更模糊交互更复杂。但万变不离其宗理解这个基础模型是驾驭所有复杂框架的钥匙。2.2 技术选型与依赖为了聚焦于核心逻辑我们做最轻量的技术选型编程语言Python。生态丰富语法简洁是AI领域的事实标准。核心依赖openai库。我们将使用OpenAI的GPT系列模型如gpt-3.5-turbo作为智能体的“思考引擎”。它负责理解指令、分解任务和生成决策。你需要一个OpenAI的API Key。辅助库无。我们尽可能用Python标准库实现保持纯净。整个项目的代码文件将只有一个大约100行。我们不会引入任何额外的“智能体框架”因为我们的目的正是要理解那些框架在背后替我们做了什么。3. 分步实现核心模块代码理论讲完了现在打开你的代码编辑器我们开始一行一行地构建我们的智能体。我会先给出每个模块的代码片段并详细解释其作用和设计考量。3.1 记忆模块的实现维护对话上下文记忆模块的核心是维护一个对话历史列表。每轮交互用户输入、智能体思考、工具调用结果都会被记录下来作为后续决策的上下文。class AgentMemory: def __init__(self): # 初始化一个空列表用于存储对话历史 self.history [] def add(self, role, content): 向历史记录中添加一条消息。 参数: role (str): 消息角色如 user, assistant, system, tool。 content (str): 消息内容。 self.history.append({role: role, content: content}) def get_context(self): 获取当前的完整对话上下文用于发送给大模型。 return self.history.copy() # 返回副本避免外部修改内部数据 def clear(self): 清空历史记录例如开始一个新会话时。 self.history.clear()代码解读与注意事项history列表中的每条消息都是一个字典遵循OpenAI Chat Completion API要求的格式role和content。这为我们后续直接调用API做好了准备。add方法是我们与记忆交互的主要接口。无论是用户提问、模型回复还是工具执行结果都通过这个方法存入。get_context方法返回历史记录的副本。这是一个重要的细节直接返回self.history可能会导致外部代码意外修改原始数据引发难以调试的问题。返回副本是一种防御性编程实践。在实际复杂应用中这个模块可能会演变为包含短期记忆当前对话窗口和长期记忆向量数据库检索的混合系统。但今天这个简单的列表足以让我们理解其基础作用。3.2 工具模块的定义赋予智能体“动手能力”工具就是Python函数。我们需要用一种方式告诉大模型“嗨我这里有几个工具你可以用这是它们的名字、描述和参数。” 我们通过一个列表来管理这些工具的定义。# 首先定义几个具体的工具函数 def search_weather(city: str) - str: 查询指定城市的天气信息。 这是一个模拟函数真实场景中会调用天气API。 # 模拟数据 weather_data { 北京: 晴15~25℃微风, 上海: 多云18~28℃东南风3级, 深圳: 阵雨22~30℃南风4级, } return weather_data.get(city, f未找到{city}的天气信息。) def calculator(expression: str) - str: 计算一个数学表达式的结果。 警告使用eval有安全风险此处仅用于演示。 在生产环境中必须使用安全的表达式解析库如ast.literal_eval。 try: # 严重安全提示eval会执行任意代码切勿在接收不可信输入的线上环境使用 result eval(expression) return f{expression} {result} except Exception as e: return f计算错误{e} # 然后创建工具描述列表用于告知大模型 TOOLS [ { name: search_weather, description: 查询某个城市的当前天气情况。, parameters: { type: object, properties: { city: {type: string, description: 城市名称例如北京、上海} }, required: [city] } }, { name: calculator, description: 计算一个数学表达式的结果支持加减乘除和括号。, parameters: { type: object, properties: { expression: {type: string, description: 数学表达式例如(35)*2} }, required: [expression] } } ]代码解读与避坑指南每个工具定义是一个字典包含name函数名、description给模型看的工具说明和parameters参数格式说明。这个格式参考了OpenAI的Function Calling规范是通用做法。calculator函数中使用了eval。这是一个极其重要的安全警告eval函数会执行传入的任意字符串作为代码如果智能体接收了用户输入的恶意表达式如__import__(os).system(rm -rf /)将导致灾难性后果。在演示中我们为了极简而使用它但在任何实际项目中都必须替换为安全的替代方案如ast.literal_eval仅支持字面量或专门的数学表达式解析库。工具的描述 (description) 非常关键。大模型完全依赖这段文本来理解工具的用途和调用方式。描述应清晰、准确并说明参数的意义。模糊的描述会导致模型错误地调用工具。3.3 与大模型交互规划与决策的核心这是智能体“思考”发生的地方。我们将编写一个函数负责将当前的对话上下文和可用工具列表发送给大模型并解析它的回复。模型的回复可能是一段自然语言也可能是一个工具调用请求。import openai import json # 假设你已经设置了环境变量 OPENAI_API_KEY # openai.api_key os.getenv(OPENAI_API_KEY) def call_llm(memory, tools): 调用大语言模型获取下一步的决策。 参数: memory: AgentMemory实例提供对话上下文。 tools: 工具定义列表。 返回: dict: 模型的回复。如果是普通对话包含 type’为 ‘message’和 ‘content’ 如果是工具调用包含 ‘type’为 ‘tool_call’以及工具名和参数。 # 准备发送给API的消息 messages memory.get_context() # 在消息中附加一个系统提示指导模型的行为 system_prompt 你是一个有帮助的AI助手可以调用工具来解决问题。 你可以使用以下工具 {} 如果你需要调用工具请严格按照以下JSON格式回复且只回复这个JSON对象 {{type: tool_call, name: tool_name, arguments: {{arg1: value1}}}} 如果不需要调用工具请直接回复最终答案。 .format(json.dumps(tools, indent2)) # 将系统提示插入到上下文的最前面 messages.insert(0, {role: system, content: system_prompt}) try: response openai.ChatCompletion.create( modelgpt-3.5-turbo, # 也可以使用 gpt-4 messagesmessages, temperature0.1, # 低温度使输出更确定更适合工具调用 ) model_reply response.choices[0].message.content.strip() # 尝试解析回复看是否是工具调用 try: action json.loads(model_reply) if action.get(type) tool_call: return action except json.JSONDecodeError: # 如果不是合法的JSON则认为是普通消息 pass # 返回普通消息 return {type: message, content: model_reply} except Exception as e: # 网络或API错误处理 return {type: message, content: f调用模型时出错{e}}代码解读与实操心得系统提示词 (System Prompt) 是灵魂我们通过插入一条系统消息来“调教”模型的行为。这条提示词清晰地定义了它的角色、可用的工具、以及回复必须遵循的格式。格式约束严格的JSON是让程序能可靠解析模型意图的关键。在复杂场景中提示词工程会变得极其重要。温度参数 (temperature)我们设置为0.1这是一个较低的值。在需要模型进行逻辑决策和工具调用的场景中较低的temperature可以使输出更加稳定和可预测减少随机性带来的不确定性。如果是创意写作则可以调高。回复解析我们尝试将模型的回复解析为JSON。如果成功且包含type: tool_call我们就认为模型决定使用工具。否则我们将其视为直接给用户的文本回复。这种“尝试解析”的模式是一种鲁棒性较强的设计。错误处理网络请求总是可能失败的。用try-except包裹API调用是基本操作返回一个友好的错误信息比让程序直接崩溃要好得多。3.4 执行模块与主循环让智能体运转起来现在我们把所有模块组装起来构建那个最关键的“循环”。主循环的逻辑是接收用户输入然后不断“思考-行动-观察”直到任务完成。class SimpleAgent: def __init__(self): self.memory AgentMemory() self.tools TOOLS # 一个工具名称到实际函数的映射 self.tool_functions { search_weather: search_weather, calculator: calculator, } self.max_steps 10 # 防止无限循环 def run_tool(self, tool_name, arguments): 执行指定的工具。 if tool_name not in self.tool_functions: return f错误未知的工具 {tool_name}。 try: func self.tool_functions[tool_name] # 根据工具定义调用函数这里做了简单适配 if tool_name search_weather: result func(cityarguments[city]) elif tool_name calculator: result func(expressionarguments[expression]) else: result func(**arguments) # 通用解包方式 return str(result) except Exception as e: return f执行工具{tool_name}时出错{e} def chat(self, user_input): 处理用户的一次输入可能触发多轮思考循环。 print(f\n[用户] {user_input}) # 1. 将用户输入存入记忆 self.memory.add(user, user_input) current_step 0 while current_step self.max_steps: current_step 1 print(f\n[Agent] 思考步骤 {current_step}...) # 2. 调用大模型进行规划/决策 decision call_llm(self.memory, self.tools) if decision[type] message: # 3. 如果是最终消息则返回 final_answer decision[content] self.memory.add(assistant, final_answer) print(f[Agent] {final_answer}) return final_answer elif decision[type] tool_call: # 4. 如果是工具调用则执行 tool_name decision[name] tool_args decision.get(arguments, {}) print(f[Agent] 决定调用工具: {tool_name}, 参数: {tool_args}) # 执行工具 tool_result self.run_tool(tool_name, tool_args) print(f[工具 {tool_name}] 返回: {tool_result}) # 5. 将工具执行结果作为一条新消息存入记忆供下一轮思考使用 # 注意角色是“tool”这有助于模型理解消息来源 self.memory.add(tool, f工具 {tool_name} 返回结果: {tool_result}) # 然后循环继续模型将基于包含工具结果的新上下文进行下一步思考 else: # 未知的决策类型 error_msg 模型返回了无法理解的决策。 self.memory.add(assistant, error_msg) return error_msg # 循环超过最大步数 timeout_msg f经过 {self.max_steps} 步仍未完成任务已终止。 self.memory.add(assistant, timeout_msg) return timeout_msg # 让我们来运行一下 if __name__ __main__: agent SimpleAgent() print(简易AI智能体已启动。输入‘退出’或‘quit’结束。) while True: user_input input(\n请输入您的问题: ) if user_input.lower() in [退出, quit, exit]: print(再见) break agent.chat(user_input)代码解读与核心逻辑工具映射self.tool_functions字典建立了工具名字符串和实际Python函数对象的映射。这是执行的关键。主循环chat方法这是智能体核心循环的具现化。步骤1记录用户输入。步骤2调用call_llm模型根据当前全部记忆包括之前的工具结果进行“思考”。步骤3如果模型认为任务已完成直接返回文本答案循环结束。步骤4如果模型决定调用工具则解析出工具名和参数交给run_tool执行。步骤5这是最关键的一步将工具执行结果以特定格式role: tool添加回记忆。这样在下一轮循环中模型就能“看到”上一步行动的结果并据此做出下一步决策。这就实现了“感知-行动”的闭环。循环终止条件max_steps是一个安全阀防止智能体陷入死循环例如模型不断调用同一个工具而无法推进任务。在实际应用中还需要更复杂的终止判断逻辑。角色扮演注意我们使用了不同的roleuser,assistant,tool,system。这符合Chat API的约定并能帮助模型更好地区分信息来源。4. 运行示例与结果分析让我们用代码实际跑几个例子看看这个100行构建的智能体是如何工作的。将上述所有代码块按顺序保存到一个simple_agent.py文件中并确保你的OPENAI_API_KEY已设置。示例1简单计算请输入您的问题: (12 8) * 2 等于多少 [用户] (12 8) * 2 等于多少 [Agent] 思考步骤 1... [Agent] 决定调用工具: calculator, 参数: {expression: (128)*2} [工具 calculator] 返回: (128)*2 40 [Agent] 思考步骤 2... [Agent] (12 8) * 2 等于 40。分析模型在第一轮思考中识别出这是一个数学计算问题于是决定调用calculator工具。工具返回结果40后这个结果被加入记忆。在第二轮思考中模型看到了工具返回的结果并判断任务已完成于是将结果组织成自然语言回复给用户。示例2多步骤任务请输入您的问题: 我想知道北京和上海的天气然后告诉我哪里更暖和。 [用户] 我想知道北京和上海的天气然后告诉我哪里更暖和。 [Agent] 思考步骤 1... [Agent] 决定调用工具: search_weather, 参数: {city: 北京} [工具 search_weather] 返回: 晴15~25℃微风 [Agent] 思考步骤 2... [Agent] 决定调用工具: search_weather, 参数: {city: 上海} [工具 search_weather] 返回: 多云18~28℃东南风3级 [Agent] 思考步骤 3... [Agent] 根据查询结果 北京晴15~25℃。 上海多云18~28℃。 上海的最高温度和最低温度都高于北京因此上海更暖和。分析这是一个经典的任务分解场景。模型没有试图一次性回答所有问题。它首先规划出第一步查询北京天气。执行后在第二步规划查询上海天气。当两个结果都获取后在第三步进行信息综合与比较给出最终判断。这完美演示了智能体的“规划-执行-再规划”循环。示例3无需工具的直接回答请输入您的问题: 你好请介绍一下你自己。 [用户] 你好请介绍一下你自己。 [Agent] 思考步骤 1... [Agent] 你好我是一个AI助手可以通过调用工具来帮助你解决问题例如查询天气或进行数学计算。请问有什么可以帮您的吗分析对于不需要工具就能回答的寒暄或知识性问题模型在单步思考后直接生成了回复类型为message循环一次即结束。5. 从原型到实战常见问题与进阶思考通过上面的代码和示例我们已经看到了一个最小可行智能体的全貌。但在实际开发中你会遇到更多挑战。下面是我总结的一些关键问题和进阶方向。5.1 调试与问题排查技巧当你发现智能体行为异常时可以按以下步骤排查检查记忆上下文在call_llm函数中打印出即将发送给模型的messages。看看历史对话是否按预期包含了所有必要信息工具执行结果的格式是否正确很多时候问题出在上下文混乱或信息缺失上。审查模型回复打印出模型返回的原始内容 (model_reply)。模型是否严格按照你要求的JSON格式回复如果没有很可能是你的系统提示词指令不够清晰。尝试让指令更严格例如“你必须以JSON格式回复且只包含JSON不要有任何其他文字。”验证工具调用单独测试你的工具函数确保它们在不同输入下都能正常工作并返回字符串。注意工具函数内部的异常捕获不要让工具崩溃导致整个智能体停止。控制循环如果智能体陷入无限循环首先检查max_steps是否生效。然后分析模型的决策它是否在重复调用同一个工具可能是工具结果未能提供推进任务所需的信息或者模型陷入了逻辑死胡同。此时需要优化提示词引导模型进行更有效的规划。5.2 性能、成本与优化考量我们的简易版本忽略了性能和成本但在真实项目中必须考虑上下文长度与成本每一轮循环都会将不断增长的对话历史发送给大模型。GPT-3.5/4的API收费是基于输入和输出的总token数量的。历史越长单次调用越贵速度也可能越慢。解决方案是记忆摘要当历史超过一定长度时不是直接截断而是让模型自己生成一个当前对话的简短摘要然后用摘要替代旧的历史保留核心信息。并行与异步如果智能体需要调用多个不依赖前后顺序的工具比如同时查询三个城市的天气串行执行效率低下。可以考虑将工具调用改为异步并行执行最后再汇总结果。错误重试与降级策略API调用可能失败工具可能出错。一个健壮的智能体应该有重试机制例如对瞬时的网络错误重试2次。如果某个工具持续失败是否有一个备选方案降级比如天气API挂了是否可以回复“目前无法获取实时天气但根据历史数据这个季节通常...”5.3 核心挑战与进阶方向复杂任务规划我们的模型只做一步规划。对于“写一份行业报告”这样的复杂任务需要更强大的规划器可能要先生成一个大纲多个子任务然后递归地对每个子任务进行同样的循环。这涉及到更高级的提示工程或专门的规划模型。工具学习与发现目前工具是硬编码的。一个更智能的体应该能动态学习使用新工具。例如给模型一个工具使用手册让它自己理解在什么场景下该选用哪个工具。这需要将工具描述进行向量化存储和检索。长期记忆与知识库当前的记忆只在一次会话中有效。真正的智能体需要持久化的记忆。这通常通过将重要信息存入向量数据库来实现。当遇到新问题时先到向量库中检索相关记忆再结合当前上下文进行决策。评估与反思我们的智能体只是机械地执行“调用工具-加入历史”的循环。一个更高级的智能体应该具备“反思”能力上一步的行动有效吗有没有更好的方法我是否误解了用户意图这可以通过在循环中引入一个“评估”步骤来实现让模型对自己和工具的结果进行评分和检讨。亲手实现这个百行核心循环的最大价值在于它为你撕开了AI智能体技术的神秘面纱。你不再是被动使用框架的开发者而是理解了其内在脉搏的设计者。下次当你看到那些功能繁多的智能体平台时你就能清晰地辨认出哪里是它的“记忆池”哪里是它的“规划器”又是哪个模块在负责“循环控制”。这份从零构建的理解是你进行定制化开发、解决复杂问题、乃至设计下一代智能体架构时最坚实的底气。