从零构建最小Agent循环：理解AI智能体的核心工作流

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的“大脑”。它接收来自感知环节的“观察”，并结合内置的“记忆”（可能是之前几轮的对话历史，也可能是长期的知识）进行推理，最终决定下一步要做什么。

这个决策通常体现为两个方面：

1. 内部思考：分析现状，可能分解复杂任务，或得出一些中间结论。这部分思考过程可以对外隐藏，也可以展示出来以增加可解释性（这就是ReAct框架中的“Thought”部分）。
2. 生成动作指令：决定调用哪个工具（函数），以及调用时传入什么参数。例如，思考环节的输出可能是：我需要调用‘天气查询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-dotenv

openai库是与大模型交互的核心。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, predicate=inspect.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_key=os.getenv("OPENAI_API_KEY"), base_url=os.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( model=self.model, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=0.1, # 低温度保证输出的稳定性，对于工具调用至关重要 stream=False ) return response.choices[0].message.content except Exception as e: return f"LLM调用出错: {e}"

关键参数解析：

• temperature=0.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} 现在，请开始处理最新的用户请求或观察结果。 """

这个提示词模板有几个精妙之处：

1. 角色定义：明确告诉模型“你是什么”。
2. 强制格式化输出：通过明确的格式要求（思考/动作，以及动作的JSON结构），我们让模型的输出变得可被程序解析。这是实现自动化循环的基石。
3. 工具集成：{tools_description}占位符将在运行时被替换为具体的工具列表。
4. 历史上下文：{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_description=tools_desc, history=history_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_ascii=False))) self.conversation_history.append(("系统-结果", action_result)) if turn_count >= max_turns: print(f"达到最大轮次 ({max_turns})，强制终止。") # 启动Agent if __name__ == "__main__": # 初始化组件 client = LLMClient(model="gpt-3.5-turbo") # 也可用 "gpt-4" registry = ToolRegistry() registry.register(Calculator) registry.register(TimeKeeper) # 创建Agent agent = MinimalAgent(client, registry) # 运行一个示例任务 task = "请计算一下15的平方根，然后告诉我现在是什么时间。" agent.run(task)

这个MinimalAgent类清晰地体现了核心循环：

1. 初始化：准备好大脑（LLMClient）和工具箱（ToolRegistry）。
2. run方法：启动循环。
   • 构建提示：将当前观察（用户输入或上轮结果）、工具描述、历史记录组合成完整的系统提示。这完成了“感知”的集成。
   • 调用LLM：发送提示，获得包含思考和动作指令的响应。这是“思考”的核心。
   • 解析响应：使用正则表达式和JSON解析，从模型回复中提取结构化信息。这是连接思考与行动的关键桥梁。
   • 执行动作：根据解析出的动作名称和参数，调用对应的工具函数。这是“行动”。
   • 更新状态：将动作结果作为新的“观察”，并更新对话历史，开启下一轮循环。

运行这段代码，你会看到Agent如何一步步地解析你的复杂指令（“计算平方根并获取时间”），先调用sqrt工具，再调用get_current_time工具，最后可能还会自动组合两个结果给出一个完整的回答。整个过程完全自动化，无需人工干预。

5. 关键问题排查与实战优化技巧

当你亲手运行起这个最小循环后，可能会遇到一些“坑”。别担心，这都是宝贵的经验。下面是我在多次实践中总结出的最常见问题和优化技巧。

5.1 模型不按格式输出怎么办？

这是新手搭建Agent时遇到的头号问题。你定义好了JSON输出格式，但模型偏偏给你回复一段散文。解决方法有以下几个层次：

1. 强化系统提示词：在提示词中明确强调格式，并使用“你必须”、“严格”等词语。可以给出多个清晰的示例。

   # 在SYSTEM_PROMPT_TEMPLATE的“行动规则”部分加强 你必须严格、精确地使用以下JSON格式输出你的动作，不要添加任何额外的解释、标记或注释。 正确示例： 动作: {"name": "add", "args": {"a": 5, "b": 3}} 错误示例： 动作: 我想我可以使用加法工具，参数是5和3。 {"name": "add", "args": {"a": 5, "b": 3}} # 前面多了文字

2. 降低Temperature：如我们之前所做，将temperature设为0.1甚至0，可以极大提高输出稳定性。

3. 使用JSON Mode：如果使用的模型支持（如GPT-4 Turbo），在API调用中设置response_format={“type”: “json_object”}，可以强制模型输出合法JSON。但注意，这要求你的整个消息内容都是围绕生成JSON的，可能需要调整提示词结构。

4. 实现一个“修复层”：在_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"。

解决方案：

1. 在工具描述中明确类型：在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)})")

2. 在调用前进行参数校验与转换：在_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:]），对于需要多步骤、信息量大的任务，它容易“忘记”最初的目标。

优化方向：

1. 任务分解：在系统提示词中，鼓励模型进行任务分解。例如：“如果用户的任务很复杂，你可以将其分解为多个子步骤，并逐步完成。”
2. 向量化记忆：对于更复杂的场景，可以引入向量数据库。将每轮的关键信息（用户目标、关键结果）转换为向量存储起来。在每一轮开始时，不仅提供最近的对话历史，还从向量库中检索与当前观察最相关的历史信息，作为上下文注入提示词。这相当于给Agent配备了“长期记忆”。
3. 总结性记忆：另一种策略是定期让模型自己总结对话的进展和当前状态，然后将这个总结作为下一轮历史的一部分，而不是罗列所有原始对话。这能有效节省Token，并聚焦核心信息。

5.4 循环无法终止或陷入死循环

有时模型会卡在一个动作上反复执行，或者无法识别任务已完成。

应对策略：

1. 设置明确的终止动作：我们已经在提示词中定义了final_answer动作，这很好。确保模型充分理解何时使用它。
2. 添加最大轮次限制：我们的max_turns参数就是最后的安全网。
3. 检测重复动作：在代码中维护一个最近动作的列表，如果发现模型在连续几轮中重复执行相同的动作（且参数相同），可以主动中断循环，并将“检测到重复动作，可能陷入循环”作为观察反馈给模型，让它自我纠正。
4. 在提示词中强调目标导向：“请始终牢记用户的最终目标。当你认为已经收集到足够的信息或完成了所有必要步骤来回答用户最初的问题时，请使用final_answer动作。”

5.5 性能与成本考量

每次循环都调用一次LLM，对于复杂任务，成本和延迟可能会累积。

优化技巧：

1. 缓存：对于具有确定性的工具调用（如查询特定数据），如果参数相同，可以缓存结果，避免重复调用和重复向LLM报告相同结果。
2. 并行工具调用：如果模型有能力（如GPT-4），可以在提示词中支持它在一个动作里并行调用多个不相关的工具。这需要更复杂的动作解析逻辑，但能显著减少循环轮次。
3. 选择更经济的模型：对于简单的工具调用和格式控制，gpt-3.5-turbo通常足够可靠且成本更低。可以将复杂的规划任务交给更强大的模型（如GPT-4），而简单的执行步骤交给轻量级模型。

搭建这个最小循环的最大收获，不是代码本身，而是对Agent工作流本质的深刻理解。每一个复杂的AI应用，背后都是这个基础循环的扩展和变形。当你再看到那些功能繁多的Agent框架时，你就能一眼看穿它的核心结构，并判断哪些功能是你真正需要的，哪些是你可以自己动手添加的。这个“最小可行产品”是你构建一切更智能、更自动化应用的坚实起点。