1. 项目概述为什么“最小”的 Agent Loop 如此重要最近和几个做AI应用的朋友聊天发现一个挺有意思的现象大家一提到“智能体”或者“Agent”脑子里蹦出来的要么是AutoGPT那种能自己上网查资料、写长文的庞然大物要么就是ReAct、COT这些听起来就很高大上的框架。但当我们真正想把手头的一个具体小需求自动化时比如每天自动整理邮件里的会议纪要或者根据产品更新日志自动生成社交媒体文案反而有点无从下手。框架太复杂依赖太多调试起来像在迷宫里打转。这就是我想聊聊“从零搭建一个最小的 Agent Loop”的原因。这个“最小”不是功能上的阉割而是指概念上的纯粹和架构上的简洁。它剥离了所有非必要的组件只保留最核心的“感知-思考-行动”循环。你可以把它理解为一个乐高基础颗粒而不是一个已经拼好的宇宙飞船。掌握了这个基础颗粒你就能清晰地理解任何复杂Agent是如何一步步构建起来的更能根据自己业务的实际需要快速拼装出真正有用、可控、成本合理的自动化工具。我自己的体会是跳过这个“最小循环”的搭建直接使用重型框架就像还没学会走就去跑马拉松很容易被框架本身的复杂性带偏忽略了问题本质。今天我就带你亲手搭一个这个“乐高基础颗粒”我们会用最少的代码核心逻辑可能不到50行讲清楚最核心的机制。你会发现Agent的内核其实非常优雅和强大。2. 核心架构拆解一个Agent Loop到底在循环什么在开始写代码之前我们必须像设计一台精密仪器一样在脑子里把它的工作原理和每个部件的作用想清楚。一个有效的Agent Loop无论后续变得多复杂其最核心的骨架都可以归结为以下三个步骤的无限循环2.1 感知获取与理解当前状态这是循环的起点。Agent不是活在真空里的它必须知道自己所处的“环境”是什么样子。这个“环境”可以是一个网页的HTML内容、一份文档的文本、数据库里最新的几条记录或者用户刚刚输入的一句话。关键点在于感知模块的输出必须是一个结构化或半结构化的“观察”而不仅仅是原始数据。例如面对用户提问“今天北京的天气怎么样”感知模块不能只输出这句原话而应该解析出其中的关键信息实体{“意图” “查询天气” “地点” “北京” “时间” “今天”}。这一步通常由一个大语言模型来完成我们称之为“理解”或“解析”阶段。在实际的最小化实现中为了极致简化我们有时会让“思考”环节直接处理原始输入但严格来说一个健壮的感知环节是必不可少的。它决定了Agent对世界的理解精度。2.2 思考基于观察进行决策这是Agent的“大脑”。它接收来自感知环节的“观察”并结合内置的“记忆”可能是之前几轮的对话历史也可能是长期的知识进行推理最终决定下一步要做什么。这个决策通常体现为两个方面内部思考分析现状可能分解复杂任务或得出一些中间结论。这部分思考过程可以对外隐藏也可以展示出来以增加可解释性这就是ReAct框架中的“Thought”部分。生成动作指令决定调用哪个工具函数以及调用时传入什么参数。例如思考环节的输出可能是我需要调用‘天气查询API’参数为{‘city’ ‘北京’}。在最小实现中我们会要求大语言模型严格按照我们定义的格式比如JSON来输出这个决策以便程序能够可靠地解析。2.3 行动执行决策并影响环境思考环节输出了动作指令行动环节就是执行它。这通常意味着调用一个预定义好的函数工具比如执行一段Python代码、调用一个外部API、在数据库中查询等。行动会产生一个“结果”比如API返回了“北京晴25℃”。这个结果连同最新的“观察”将被送入下一轮循环的“感知”或直接作为“思考”的输入从而开启下一个迭代。这个循环何时结束由“思考”环节决定。当模型认为任务已经完成或者无法继续时它会输出一个特殊的动作指令如final_answer并附带最终答案循环随即终止。用一个简单的流程图来概括这个永不停止直到任务完成的引擎[感知获取用户问题/环境状态] | v [思考分析并决定下一步行动] | v [行动执行工具调用] | v [获取行动结果作为新的观察] | --- 循环回到【思考】或【感知】...3. 环境准备与工具定义搭建舞台与准备道具在让我们的Agent演员上台表演之前我们需要先搭建好舞台运行环境和准备好它可能用到的道具工具集。这里我们选择Python作为实现语言因为它有极其丰富的AI生态库。3.1 基础依赖安装打开你的终端创建一个新的项目目录并安装最核心的包。我们这里以使用OpenAI的API为例但你完全可以替换成任何其他兼容OpenAI接口的模型服务。# 创建项目目录并进入 mkdir minimal-agent cd minimal-agent # 创建虚拟环境推荐避免包冲突 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 安装核心依赖 pip install openai python-dotenvopenai库是与大模型交互的核心。python-dotenv用于管理你的API密钥等敏感信息避免硬编码在代码里。接下来在项目根目录创建一个.env文件存放你的密钥OPENAI_API_KEY你的实际api密钥 OPENAI_BASE_URL你的api基础地址如果使用第三方代理然后在代码中通过os.getenv来读取。3.2 设计并封装核心工具工具是Agent延伸的手脚。一个“最小”的Agent至少应该有一个工具可用。我们来定义两个最经典的工具一个用于计算一个用于获取当前时间。# tools.py import math from datetime import datetime class Calculator: 一个简单的计算器工具能执行基础运算。 staticmethod def add(a: float, b: float) - float: 返回两个数字的和。 return a b staticmethod def subtract(a: float, b: float) - float: 返回两个数字的差 (a - b)。 return a - b staticmethod def multiply(a: float, b: float) - float: 返回两个数字的乘积。 return a * b staticmethod def divide(a: float, b: float) - float: 返回两个数字的商 (a / b)。如果除数为零则返回错误信息。 if b 0: return 错误除数不能为零。 return a / b staticmethod def sqrt(a: float) - float: 返回一个数字的平方根。 if a 0: return 错误不能对负数开平方根。 return math.sqrt(a) class TimeKeeper: 一个获取当前时间的工具。 staticmethod def get_current_time(format: str %Y-%m-%d %H:%M:%S) - str: 返回当前时间。 参数: format: 时间格式字符串默认为%Y-%m-%d %H:%M:%S return datetime.now().strftime(format)为什么要把工具封装成类这不仅仅是代码组织的问题。清晰的封装有利于我们后续通过反射机制自动发现和描述所有可用的工具这是构建可扩展Agent系统的关键一步。每个工具方法都应该有清晰的文档字符串这些字符串稍后会被用来自动生成给大模型看的“工具使用说明书”。3.3 构建工具注册与调用机制有了工具类我们需要一个中心化的“工具箱”来管理它们并能根据名称动态调用。# tool_registry.py import inspect from typing import Dict, Any, Callable class ToolRegistry: 工具注册表负责管理所有可用工具及其描述。 def __init__(self): self._tools: Dict[str, Dict] {} # 工具名 - {函数 描述} def register(self, tool_class: Any): 注册一个工具类中的所有静态方法。 for name, method in inspect.getmembers(tool_class, predicateinspect.isfunction): if not name.startswith(_): # 跳过私有方法 self._tools[name] { function: method, description: method.__doc__.strip() if method.__doc__ else 无描述 } print(f已注册工具类: {tool_class.__name__}) def get_tool(self, name: str) - Callable: 根据名称获取工具函数。 tool_info self._tools.get(name) if not tool_info: raise ValueError(f工具 {name} 未注册。) return tool_info[function] def get_tools_description(self) - str: 生成给LLM看的工具描述文本。 description_lines [你可以使用以下工具] for tool_name, tool_info in self._tools.items(): desc tool_info[description] # 简单解析函数签名让模型知道参数 func tool_info[function] sig inspect.signature(func) params list(sig.parameters.keys()) description_lines.append(f- {tool_name}: {desc} 参数: {params}) return \n.join(description_lines) property def available_tools(self): 返回所有已注册的工具名称列表。 return list(self._tools.keys())这个注册表是我们Agent系统的“装备库”。它做了三件关键事1. 自动收集工具信息2. 提供查询和调用接口3. 生成格式化的工具描述这是后续引导大模型正确使用工具的关键。4. 核心循环实现组装大脑与引擎现在舞台和道具都已就位是时候请出我们的大脑LLM并组装运行引擎了。这部分代码是整个项目的灵魂但它的核心逻辑可能比你想象的要简洁。4.1 与大语言模型的交互封装首先我们封装一个简单的LLM客户端用于发送提示词和接收回复。这里我们采用OpenAI的ChatCompletion格式。# llm_client.py import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() # 加载.env文件中的环境变量 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 def generate_response(self, system_prompt: str, user_prompt: str) - str: 发送请求到LLM并返回回复内容。 为了简化这里省略了复杂的错误处理和流式输出。 try: response self.client.chat.completions.create( modelself.model, messages[ {role: system, content: system_prompt}, {role: user, content: user_prompt} ], temperature0.1, # 低温度保证输出的稳定性对于工具调用至关重要 streamFalse ) return response.choices[0].message.content except Exception as e: return fLLM调用出错: {e}关键参数解析temperature0.1 这个设置非常重要。在工具调用场景下我们需要模型尽可能稳定、确定性地输出格式化的内容如JSON而不是富有“创意”的散文。较低的温度值有助于减少随机性。system_prompt 这是引导模型行为的关键。我们将在这里定义Agent的角色、思考格式以及可用的工具列表。4.2 构建系统提示词给大脑植入“操作系统”系统提示词是Agent的“宪法”它定义了Agent应该如何思考、如何响应。编写一个好的系统提示是Agent能否正确工作的决定性因素。# 这是一个提示词模板我们会在运行时填充具体工具描述 SYSTEM_PROMPT_TEMPLATE 你是一个专业的任务执行助手。你的目标是通过使用合适的工具一步步解决用户的问题。 ## 行动规则 1. 你一次只能执行一个动作。 2. 你必须严格按照以下格式输出你的思考过程和下一步动作思考: [你当前的分析和推理过程] 动作: { name: 工具名称, args: { 参数1: 值1, 参数2: 值2, ... } }3. 如果根据当前信息你认为问题已经解决需要给出最终答案请使用以下格式思考: [总结性思考] 动作: { name: final_answer, args: { answer: 你的最终答案 } }## 可用工具 {tools_description} ## 当前对话历史 {history} 现在请开始处理最新的用户请求或观察结果。 这个提示词模板有几个精妙之处角色定义明确告诉模型“你是什么”。强制格式化输出通过明确的格式要求思考/动作以及动作的JSON结构我们让模型的输出变得可被程序解析。这是实现自动化循环的基石。工具集成{tools_description}占位符将在运行时被替换为具体的工具列表。历史上下文{history}占位符用于注入之前的对话轮次让模型拥有“记忆”。4.3 实现主循环逻辑最后我们把所有部件组装起来形成那个著名的“感知-思考-行动”循环。# main_loop.py import json import re from llm_client import LLMClient from tool_registry import ToolRegistry from tools import Calculator, TimeKeeper class MinimalAgent: def __init__(self, llm_client: LLMClient, tool_registry: ToolRegistry): self.llm llm_client self.tools tool_registry self.conversation_history [] # 用于存储多轮对话 def _parse_model_response(self, response: str) - tuple: 解析模型的响应提取思考内容和动作JSON。 这是整个循环中最脆弱的环节需要健壮的解析逻辑。 thought_match re.search(r思考:\s*(.*?)(?\n动作:|$), response, re.DOTALL) action_match re.search(r动作:\s*(\{.*?\}), response, re.DOTALL) thought thought_match.group(1).strip() if thought_match else 未提供思考过程。 action_json_str action_match.group(1).strip() if action_match else None if not action_json_str: return thought, None try: action_dict json.loads(action_json_str) return thought, action_dict except json.JSONDecodeError as e: print(f解析动作JSON失败: {e}, 原始内容: {action_json_str}) # 可以在这里添加一些修复逻辑比如尝试提取更简单的结构 return thought, None def _execute_action(self, action_dict: dict) - str: 执行动作字典中指定的工具调用。 if not action_dict or name not in action_dict: return 无效的动作指令。 action_name action_dict[name] args action_dict.get(args, {}) # 检查是否为最终答案 if action_name final_answer: print(f最终答案: {args.get(answer, 无答案)}) return 任务完成。 # 执行工具调用 if action_name not in self.tools.available_tools: return f错误未知工具 {action_name}。 try: tool_func self.tools.get_tool(action_name) # 将参数字典展开为关键字参数 result tool_func(**args) return str(result) except TypeError as e: return f工具调用参数错误: {e} except Exception as e: return f工具执行异常: {e} def run(self, user_input: str, max_turns: int 10): 运行Agent主循环。 Args: user_input: 用户的初始请求。 max_turns: 最大循环轮次防止无限循环。 print(f用户: {user_input}) current_observation user_input turn_count 0 while turn_count max_turns: turn_count 1 print(f\n--- 第 {turn_count} 轮 ---) # 1. 构建提示词 (感知思考的触发点) tools_desc self.tools.get_tools_description() # 简化历史只保留最近几轮以避免token过长 history_str \n.join([f{role}: {content} for role, content in self.conversation_history[-4:]]) system_prompt SYSTEM_PROMPT_TEMPLATE.format( tools_descriptiontools_desc, historyhistory_str ) # 2. 调用LLM进行思考 (思考) llm_response self.llm.generate_response(system_prompt, current_observation) print(f模型原始响应:\n{llm_response}) # 3. 解析响应 thought, action_dict self._parse_model_response(llm_response) print(f解析结果 - 思考: {thought}) print(f解析结果 - 动作: {action_dict}) # 4. 记录思考到历史 self.conversation_history.append((助手-思考, thought)) # 5. 执行动作 (行动) if action_dict and action_dict.get(name) final_answer: print(任务完成循环终止。) break action_result self._execute_action(action_dict) print(f动作执行结果: {action_result}) # 6. 将结果作为新的观察准备下一轮循环 # 通常我们会把“动作”和“结果”一起作为观察反馈给模型 current_observation f上次动作的结果是: {action_result} # 记录动作和结果到历史 self.conversation_history.append((助手-动作, json.dumps(action_dict, ensure_asciiFalse))) self.conversation_history.append((系统-结果, action_result)) if turn_count max_turns: print(f达到最大轮次 ({max_turns})强制终止。) # 启动Agent if __name__ __main__: # 初始化组件 client LLMClient(modelgpt-3.5-turbo) # 也可用 gpt-4 registry ToolRegistry() registry.register(Calculator) registry.register(TimeKeeper) # 创建Agent agent MinimalAgent(client, registry) # 运行一个示例任务 task 请计算一下15的平方根然后告诉我现在是什么时间。 agent.run(task)这个MinimalAgent类清晰地体现了核心循环初始化准备好大脑LLMClient和工具箱ToolRegistry。run方法启动循环。构建提示将当前观察用户输入或上轮结果、工具描述、历史记录组合成完整的系统提示。这完成了“感知”的集成。调用LLM发送提示获得包含思考和动作指令的响应。这是“思考”的核心。解析响应使用正则表达式和JSON解析从模型回复中提取结构化信息。这是连接思考与行动的关键桥梁。执行动作根据解析出的动作名称和参数调用对应的工具函数。这是“行动”。更新状态将动作结果作为新的“观察”并更新对话历史开启下一轮循环。运行这段代码你会看到Agent如何一步步地解析你的复杂指令“计算平方根并获取时间”先调用sqrt工具再调用get_current_time工具最后可能还会自动组合两个结果给出一个完整的回答。整个过程完全自动化无需人工干预。5. 关键问题排查与实战优化技巧当你亲手运行起这个最小循环后可能会遇到一些“坑”。别担心这都是宝贵的经验。下面是我在多次实践中总结出的最常见问题和优化技巧。5.1 模型不按格式输出怎么办这是新手搭建Agent时遇到的头号问题。你定义好了JSON输出格式但模型偏偏给你回复一段散文。解决方法有以下几个层次强化系统提示词在提示词中明确强调格式并使用“你必须”、“严格”等词语。可以给出多个清晰的示例。# 在SYSTEM_PROMPT_TEMPLATE的“行动规则”部分加强 你必须严格、精确地使用以下JSON格式输出你的动作不要添加任何额外的解释、标记或注释。 正确示例 动作: {name: add, args: {a: 5, b: 3}} 错误示例 动作: 我想我可以使用加法工具参数是5和3。 {name: add, args: {a: 5, b: 3}} # 前面多了文字降低Temperature如我们之前所做将temperature设为0.1甚至0可以极大提高输出稳定性。使用JSON Mode如果使用的模型支持如GPT-4 Turbo在API调用中设置response_format{“type”: “json_object”}可以强制模型输出合法JSON。但注意这要求你的整个消息内容都是围绕生成JSON的可能需要调整提示词结构。实现一个“修复层”在_parse_model_response函数中增加后处理逻辑。如果JSON解析失败可以尝试用更灵活的方法提取比如寻找第一个“{”和最后一个“}”之间的内容。def _parse_model_response(self, response: str) - tuple: # ... 原有正则匹配 ... if not action_json_str: # 尝试暴力提取JSON对象 start response.find({) end response.rfind(}) 1 if start ! -1 and end ! 0: action_json_str response[start:end] # 再次尝试json.loads # ... 可以记录日志观察模型不守规矩的模式5.2 工具参数类型不匹配导致调用失败模型可能会猜错参数类型。比如计算器工具期望数字参数但模型传递了字符串15。解决方案在工具描述中明确类型在ToolRegistry.get_tools_description方法生成描述时不仅列出参数名也列出其期望的类型。# 改进的描述生成 sig inspect.signature(func) params_desc [] for param_name, param in sig.parameters.items(): param_type param.annotation if param.annotation ! inspect.Parameter.empty else Any params_desc.append(f{param_name}: {param_type}) description_lines.append(f- {tool_name}: {desc} 参数: ({, .join(params_desc)}))在调用前进行参数校验与转换在_execute_action中根据工具函数的签名注解尝试将传入的字符串参数转换为正确的类型。def _execute_action(self, action_dict: dict) - str: # ... 获取 tool_func ... sig inspect.signature(tool_func) bound_args {} for param_name, param in sig.parameters.items(): raw_value args.get(param_name) if raw_value is None: if param.default inspect.Parameter.empty: return f错误缺少必需参数 {param_name}。 else: continue # 尝试类型转换 try: # 这里可以根据 param.annotation 做更精细的转换 # 例如如果注解是 int就尝试转 int if param.annotation is int: bound_args[param_name] int(raw_value) elif param.annotation is float: bound_args[param_name] float(raw_value) else: bound_args[param_name] raw_value except (ValueError, TypeError) as e: return f参数 {param_name} 类型转换失败: {e} result tool_func(**bound_args) return str(result)5.3 如何处理复杂任务与长期记忆我们当前的Agent只有很短的对话历史history[-4:]对于需要多步骤、信息量大的任务它容易“忘记”最初的目标。优化方向任务分解在系统提示词中鼓励模型进行任务分解。例如“如果用户的任务很复杂你可以将其分解为多个子步骤并逐步完成。”向量化记忆对于更复杂的场景可以引入向量数据库。将每轮的关键信息用户目标、关键结果转换为向量存储起来。在每一轮开始时不仅提供最近的对话历史还从向量库中检索与当前观察最相关的历史信息作为上下文注入提示词。这相当于给Agent配备了“长期记忆”。总结性记忆另一种策略是定期让模型自己总结对话的进展和当前状态然后将这个总结作为下一轮历史的一部分而不是罗列所有原始对话。这能有效节省Token并聚焦核心信息。5.4 循环无法终止或陷入死循环有时模型会卡在一个动作上反复执行或者无法识别任务已完成。应对策略设置明确的终止动作我们已经在提示词中定义了final_answer动作这很好。确保模型充分理解何时使用它。添加最大轮次限制我们的max_turns参数就是最后的安全网。检测重复动作在代码中维护一个最近动作的列表如果发现模型在连续几轮中重复执行相同的动作且参数相同可以主动中断循环并将“检测到重复动作可能陷入循环”作为观察反馈给模型让它自我纠正。在提示词中强调目标导向“请始终牢记用户的最终目标。当你认为已经收集到足够的信息或完成了所有必要步骤来回答用户最初的问题时请使用final_answer动作。”5.5 性能与成本考量每次循环都调用一次LLM对于复杂任务成本和延迟可能会累积。优化技巧缓存对于具有确定性的工具调用如查询特定数据如果参数相同可以缓存结果避免重复调用和重复向LLM报告相同结果。并行工具调用如果模型有能力如GPT-4可以在提示词中支持它在一个动作里并行调用多个不相关的工具。这需要更复杂的动作解析逻辑但能显著减少循环轮次。选择更经济的模型对于简单的工具调用和格式控制gpt-3.5-turbo通常足够可靠且成本更低。可以将复杂的规划任务交给更强大的模型如GPT-4而简单的执行步骤交给轻量级模型。搭建这个最小循环的最大收获不是代码本身而是对Agent工作流本质的深刻理解。每一个复杂的AI应用背后都是这个基础循环的扩展和变形。当你再看到那些功能繁多的Agent框架时你就能一眼看穿它的核心结构并判断哪些功能是你真正需要的哪些是你可以自己动手添加的。这个“最小可行产品”是你构建一切更智能、更自动化应用的坚实起点。