在实际项目中将AI能力从简单的对话或生成工具转变为能够自主执行任务、管理流程的“自动化代理”是当前技术落地的一个关键跃迁。很多开发者学习了AI模型的基础调用但面对“构建一个能独立工作的AI代理”这个目标时却不知从何入手陷入工具堆砌或设计复杂的困境。本文旨在为初学者提供一条清晰的路径从核心概念理解到环境搭建再到实现一个具备基础感知、决策和执行能力的AI代理原型并最终探讨将其转化为可持续业务所需考虑的生产级要素。本文适合已经了解Python基础、对主流大模型API如OpenAI、智谱、DeepSeek等有初步调用经验并希望将这些能力系统化、自动化的开发者。我们将避开空泛的理论聚焦于可运行的代码、可复现的配置以及从开发到部署的完整闭环。读完本文你将能够搭建一个可以接收目标、分解任务、调用工具并持续运行的AI代理骨架并理解其扩展为复杂业务系统的核心模式。1. 理解AI自动化代理的核心从工具到智能体在开始写代码之前必须厘清几个关键概念这决定了我们构建的系统的设计方向和复杂度。1.1 AI代理与简单API调用的本质区别简单的AI API调用例如让模型生成一段文本或分析一张图片是一个“请求-响应”的瞬时过程。用户提供输入模型返回输出过程结束。而AI自动化代理则是一个具有状态持续性、目标导向性和自主行动能力的系统。状态持续性代理能记住之前的交互历史、执行结果和环境上下文并基于此做出后续决策。这通常通过维护一个“记忆”或“会话历史”模块来实现。目标导向性代理接收一个高级目标如“帮我分析本周销售数据并写一份报告”而非具体的指令。代理需要自己规划如何达成这个目标。自主行动能力代理不仅会“思考”生成文本还会“行动”。它可以自主调用外部工具如执行代码、查询数据库、调用第三方API、操作文件系统等。这是自动化得以实现的关键。一个典型的AI代理工作流遵循“感知-规划-执行-反思”的循环感知接收用户输入或环境信号。规划分析当前状态和目标决定下一步做什么思考可能分解为子任务。执行调用合适的工具或能力来完成规划出的动作。反思评估执行结果更新内部状态决定是继续下一个子任务、调整规划还是向用户请求澄清。1.2 关键组件与架构模式要构建这样一个系统我们需要设计几个核心组件大脑LLM Core负责所有的推理、规划和决策。通常由一个大语言模型担任它处理自然语言理解目标并生成下一步的行动计划通常以结构化格式如JSON。记忆Memory存储代理的历史对话、工具执行结果、学到的知识等。可以是简单的列表、向量数据库用于语义检索长期记忆或更复杂的图结构。工具Tools代理可以调用的函数或API集合。这是代理与外部世界交互的“手”和“脚”。例如search_web,execute_python_code,read_file,send_email。执行器Orchestrator控制整个循环流程的组件。它负责初始化代理、管理记忆、调用大脑进行规划、解析大脑的输出、调用对应的工具、处理工具返回结果并决定下一步。目前社区流行的架构模式主要有两种ReAct模式强调将“推理”和“行动”明确分离。代理的输出格式为Thought: ... Action: ... Action Input: ... 执行器解析后执行对应动作再将结果Observation: ...返回给代理进行下一轮思考。结构清晰易于调试。Plan-and-Execute模式代理先制定一个完整的计划任务列表然后按顺序或并行地执行这些任务。适合目标明确、步骤可预先分解的场景。对于初学者从ReAct模式入手更容易理解代理的运作机制。1.3 技术选型框架与模型手动从零搭建所有轮子是一个很好的学习过程但对于快速构建原型使用成熟的框架是更高效的选择。以下是一些常见选择及其特点框架/库语言特点适用场景LangChainPython/JS生态最丰富组件齐全文档庞大。概念较多学习曲线稍陡。快速构建复杂、生产级的AI应用需要大量集成。LlamaIndexPython专注于数据连接和检索RAG在构建知识库代理方面有优势。代理需要深度处理私有文档、知识库的场景。Semantic KernelC#/Python微软出品强于规划、插件技能管理和流程编排。.NET生态或需要复杂工作流编排的项目。AutoGenPython支持多代理对话和协作擅长构建代理团队。需要多个专业代理协同完成任务的场景。简单自建Python完全可控深度定制依赖少。学习原理、构建轻量级或特定场景的代理。对于初学者指南我们将采用“简单自建”结合LangChain的部分核心思想的方式以确保你能理解每一行代码在做什么同时又不至于陷入底层细节。模型方面我们将使用OpenAI GPT-4/3.5-Turbo的API作为“大脑”因为它提供了稳定且强大的推理能力。你也可以替换为其他兼容OpenAI API格式的模型服务。2. 环境准备与项目初始化我们将在Python环境中构建代理。请确保你的开发环境满足以下要求。2.1 基础环境与依赖安装首先确保你已安装Python推荐3.8以上版本。然后创建一个新的项目目录并初始化虚拟环境这能有效隔离依赖。# 创建项目目录 mkdir ai_agent_project cd ai_agent_project # 创建虚拟环境以venv为例 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate接下来安装核心依赖。我们将使用openai库与模型交互langchain库主要借用其优秀的“工具”抽象和调用模板但核心循环我们自己控制。# 安装核心库 pip install openai langchain # 安装可能用到的工具库示例 pip install requests # 用于调用网页API pip install python-dotenv # 用于管理环境变量2.2 配置模型API密钥为了调用大模型你需要一个API密钥。这里以OpenAI为例。永远不要将密钥硬编码在代码中。在项目根目录创建.env文件。在.env文件中填入你的密钥OPENAI_API_KEYsk-your-actual-api-key-here在代码中通过python-dotenv加载。同时创建一个config.py文件来集中管理配置# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) OPENAI_API_BASE os.getenv(OPENAI_API_BASE, https://api.openai.com/v1) # 支持自定义端点 MODEL_NAME os.getenv(MODEL_NAME, gpt-3.5-turbo) # 默认使用 3.5-turbo2.3 项目结构设计一个清晰的项目结构有助于后续扩展。建议如下ai_agent_project/ ├── .env # 环境变量列入.gitignore ├── config.py # 配置文件 ├── main.py # 主程序入口 ├── agent/ │ ├── __init__.py │ ├── core.py # 代理核心循环、大脑 │ ├── memory.py # 记忆模块 │ └── tools/ # 工具集目录 │ ├── __init__.py │ ├── base_tool.py # 工具基类 │ ├── web_search.py # 网页搜索工具示例 │ └── calculator.py # 计算器工具示例 └── utils/ └── __init__.py3. 构建核心组件工具、记忆与大脑现在我们从下往上构建代理的各个部件。3.1 定义工具Tools工具是代理能力的扩展。我们定义一个简单的工具基类所有具体工具都继承它。# agent/tools/base_tool.py from abc import ABC, abstractmethod from typing import Any, Dict class BaseTool(ABC): 工具基类 name: str # 工具名称代理通过这个名称来调用 description: str # 工具描述用于帮助LLM理解何时使用此工具 parameters: Dict[str, Any] # 工具所需的参数定义 def __init__(self, name: str, description: str): self.name name self.description description self.parameters {} abstractmethod def execute(self, **kwargs) - str: 执行工具的核心逻辑返回字符串格式的结果 pass def to_function_call_schema(self) - Dict: 将工具转换为OpenAI Function Calling所需的格式 return { type: function, function: { name: self.name, description: self.description, parameters: { type: object, properties: self.parameters, required: list(self.parameters.keys()) } } }接着实现两个简单的工具一个计算器和一个模拟的网页搜索。# agent/tools/calculator.py from .base_tool import BaseTool import math class CalculatorTool(BaseTool): def __init__(self): super().__init__( namecalculator, description执行数学计算。支持加()、减(-)、乘(*)、除(/)、乘方(**)、平方根(sqrt)等。 ) self.parameters { expression: {type: string, description: 数学表达式例如3 5 * 2 或 sqrt(16)} } def execute(self, expression: str) - str: try: # 警告在生产环境中直接eval用户输入是极度危险的 # 这里仅为演示实际应使用安全的表达式解析库如 ast.literal_eval 限制操作。 # 此处我们做一个极简的安全过滤和计算。 allowed_chars set(0123456789-*/.() sqrt) if not all(c in allowed_chars for c in expression): return 错误表达式中包含不安全字符。 # 替换sqrt为math.sqrt expression expression.replace(sqrt, math.sqrt) result eval(expression, {__builtins__: {}}, {math: math}) return f计算结果{result} except Exception as e: return f计算出错{e} # agent/tools/web_search.py from .base_tool import BaseTool import requests import json class WebSearchTool(BaseTool): def __init__(self, api_key: str None, search_engine: str google): # 注意这里需要接入真实的搜索API如SerpAPI、Google Custom Search等。 # 此处仅为模拟。 super().__init__( nameweb_search, description在互联网上搜索信息。当需要获取最新、未知或事实性信息时使用。 ) self.api_key api_key self.search_engine search_engine self.parameters { query: {type: string, description: 搜索关键词} } def execute(self, query: str) - str: # 模拟搜索返回 # 真实情况下这里应调用第三方API mock_results [ f关于{query}的搜索结果1..., f关于{query}的搜索结果2..., f关于{query}的搜索结果3... ] return f搜索完成。摘要{; .join(mock_results[:2])}在agent/tools/__init__.py中导出工具方便管理。# agent/tools/__init__.py from .calculator import CalculatorTool from .web_search import WebSearchTool __all__ [CalculatorTool, WebSearchTool]3.2 实现记忆Memory我们实现一个简单的对话历史记忆只保存最近的几轮交互。# agent/memory.py from typing import List, Dict, Any class SimpleConversationMemory: 简单的对话记忆保存固定轮数的历史 def __init__(self, max_turns: int 10): self.max_turns max_turns self.history: List[Dict[str, str]] [] # 每个元素是 {role: user/assistant/tool, content: ...} def add_user_message(self, message: str): self.history.append({role: user, content: message}) self._trim() def add_assistant_message(self, message: str): self.history.append({role: assistant, content: message}) self._trim() def add_tool_message(self, tool_name: str, result: str): # 工具执行结果通常以“Observation”的形式加入历史 self.history.append({role: tool, content: f[{tool_name}] 返回: {result}}) self._trim() def get_conversation_history(self) - List[Dict[str, str]]: 获取用于LLM提示的对话历史格式 return self.history.copy() def _trim(self): 保持历史记录不超过最大轮数 if len(self.history) self.max_turns: # 简单策略移除最早的一对用户/助手消息如果可能 # 更复杂的策略可以基于token数裁剪 self.history self.history[-self.max_turns:] def clear(self): self.history []3.3 构建大脑与执行器Core这是最核心的部分我们将实现一个遵循ReAct模式的代理循环。# agent/core.py import json import openai from typing import List, Dict, Any, Optional from .memory import SimpleConversationMemory from .tools.base_tool import BaseTool class SimpleAIAgent: def __init__(self, model_name: str gpt-3.5-turbo, api_key: str None, base_url: str None): self.client openai.OpenAI(api_keyapi_key, base_urlbase_url) self.model_name model_name self.memory SimpleConversationMemory() self.tools: Dict[str, BaseTool] {} # 工具名 - 工具实例的映射 self.max_iterations 10 # 防止无限循环 def register_tool(self, tool: BaseTool): 注册一个工具到代理 self.tools[tool.name] tool def _build_messages(self, user_input: str) - List[Dict[str, str]]: 构建发送给LLM的消息列表 messages [ { role: system, content: 你是一个有帮助的AI助手可以调用工具来解决问题。请遵循以下格式 思考你需要先思考当前情况和目标。 行动调用工具时请严格按照以下JSON格式输出 {action: 工具名称, action_input: {参数名: 参数值}} 如果你已经得到最终答案或者无需使用工具请直接输出最终答案。 } ] # 加入历史记忆 messages.extend(self.memory.get_conversation_history()) # 加入当前用户输入 messages.append({role: user, content: user_input}) return messages def _parse_llm_response(self, response: str) - Dict[str, Any]: 解析LLM的回复判断是最终答案还是工具调用 response response.strip() # 尝试解析JSON格式的工具调用 if response.startswith({) and response.endswith(}): try: data json.loads(response) if action in data and action_input in data: return {type: action, data: data} except json.JSONDecodeError: pass # 否则视为最终答案 return {type: final_answer, data: response} def _execute_action(self, action_name: str, action_input: Dict) - str: 执行工具调用 if action_name not in self.tools: return f错误未知工具 {action_name}。 tool self.tools[action_name] try: # 这里简化处理实际需要根据工具定义的参数进行校验和适配 result tool.execute(**action_input) return result except Exception as e: return f工具执行出错{e} def run(self, user_input: str) - str: 运行代理的主要循环 self.memory.add_user_message(user_input) print(f用户: {user_input}) for i in range(self.max_iterations): # 1. 构建消息并调用LLM messages self._build_messages(user_input) try: response self.client.chat.completions.create( modelself.model_name, messagesmessages, temperature0.1, # 低温度保证输出稳定易于解析 streamFalse, ) llm_output response.choices[0].message.content except Exception as e: return f调用模型API失败{e} print(f助手思考: {llm_output}) # 2. 解析LLM输出 parsed self._parse_llm_response(llm_output) if parsed[type] final_answer: final_answer parsed[data] self.memory.add_assistant_message(final_answer) print(f最终答案: {final_answer}) return final_answer elif parsed[type] action: action_data parsed[data] action_name action_data[action] action_input action_data[action_input] print(f执行动作: {action_name}, 输入: {action_input}) # 3. 执行工具 observation self._execute_action(action_name, action_input) print(f观察结果: {observation}) # 4. 将观察结果加入记忆开始下一轮循环 self.memory.add_assistant_message(llm_output) # 记录助手的“思考/行动” self.memory.add_tool_message(action_name, observation) # 记录工具结果 # 下一轮循环的“用户输入”实际上是上一步的观察结果但我们的消息构建方式已包含历史。 # 这里只需更新user_input为观察结果以便在下一轮构建消息时历史是最新的。 user_input observation # 简化处理实际ReAct中观察是作为下一轮LLM输入的一部分。 else: return 无法解析LLM的输出。 return f达到最大迭代次数({self.max_iterations})未完成目标。4. 组装与运行你的第一个AI代理现在我们将所有组件组装起来并运行一个完整的示例。4.1 主程序入口创建main.py作为启动文件。# main.py from config import Config from agent.core import SimpleAIAgent from agent.tools import CalculatorTool, WebSearchTool def main(): # 1. 初始化配置 config Config() # 2. 创建代理实例 agent SimpleAIAgent( model_nameconfig.MODEL_NAME, api_keyconfig.OPENAI_API_KEY, base_urlconfig.OPENAI_API_BASE ) # 3. 注册工具 agent.register_tool(CalculatorTool()) # 如果需要真实的搜索在这里传入API Key # agent.register_tool(WebSearchTool(api_keyyour_serpapi_key)) agent.register_tool(WebSearchTool()) # 使用模拟搜索 # 4. 运行代理 print( AI 代理已启动输入 quit 或 exit 退出 ) while True: try: user_input input(\n请输入你的问题或指令: ).strip() if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input: continue final_answer agent.run(user_input) # run方法内部已打印过程这里可以额外处理最终答案 # 例如保存到文件或触发其他操作 except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f运行过程中发生错误{e}) if __name__ __main__: main()4.2 运行与验证在终端中确保虚拟环境已激活并且.env文件中的OPENAI_API_KEY已正确设置。python main.py程序启动后你可以尝试输入一些指令来测试代理的能力纯推理问题“法国的首都是哪里”代理可能直接回答无需调用工具。需要计算的问题“计算一下 15 的平方加上 28 除以 4 等于多少”代理应该识别出需要计算调用calculator工具并返回结果。需要搜索的问题“今天北京的天气怎么样”代理会调用web_search工具模拟并基于“搜索结果”给出回答。混合任务“先搜索一下爱因斯坦的生日然后计算他到今天假设是2024年1月1日活了多少天”代理需要先搜索获取生日信息再进行日期计算这里我们的计算器工具不支持日期可以观察代理如何处理。观察控制台输出你会看到类似以下的流程用户: 计算一下 15 的平方加上 28 除以 4 等于多少 助手思考: 思考用户需要计算一个数学表达式。我需要使用计算器工具。 行动{action: calculator, action_input: {expression: 15**2 28/4}} 执行动作: calculator, 输入: {expression: 15**2 28/4} 观察结果: 计算结果232.0 助手思考: 思考我已经得到了计算结果。 最终答案: 15的平方是22528除以4是7两者相加等于232。这个流程清晰地展示了代理的“思考-行动-观察”循环。5. 从原型到业务关键考量与进阶方向一个能跑通的原型距离一个可靠、可用的AI自动化代理业务还有很长的路。以下是几个必须深入考虑的方向。5.1 稳定性与可靠性提升错误处理与重试当前代码对工具执行和API调用错误处理很基础。生产环境需要更健壮的机制例如网络超时重试、模型API的退避重试、工具执行异常的回退方案等。循环检测与中断代理可能陷入死循环。除了设置max_iterations还可以检测重复的动作、无进展的对话并引入人工中断或策略调整。输出格式校验我们简单使用JSON解析很脆弱。应使用更严格的校验如Pydantic模型并为LLM提供更清晰的输出格式指导甚至采用支持“函数调用”功能的模型让模型直接返回结构化调用对象。5.2 记忆与上下文管理优化短期与长期记忆SimpleConversationMemory是短期记忆。对于需要记住大量历史或知识的场景需要引入向量数据库如Chroma, Pinecone作为长期记忆实现基于语义的检索。上下文窗口限制大模型有token限制。需要实现智能的上下文窗口管理例如总结冗长对话、选择性保留关键信息、使用更高效的token压缩技术。记忆的结构化将记忆分类存储如用户偏好、任务历史、学到的知识等便于代理更精准地利用。5.3 工具生态与安全性工具的动态注册与发现在大型系统中工具可能来自不同模块。需要设计机制让代理能动态发现和调用新工具。工具权限与安全这是重中之重。必须为工具调用设置严格的权限控制。例如execute_python_code工具必须在安全的沙箱环境中运行文件操作工具必须有路径白名单数据库操作工具必须有查询限制。永远不要在生产环境直接eval用户输入。工具描述的质量工具的描述description和参数定义直接影响LLM调用的准确性。需要精心编写并可以通过少量示例few-shot来提升模型的理解。5.4 监控、评估与成本控制全链路日志记录每一轮的用户输入、LLM思考、工具调用、工具结果、最终输出。这是排查问题、分析代理行为和改进提示词的基石。关键指标监控延迟每轮循环的耗时。成本API调用token消耗。工具调用分布哪些工具被频繁使用。成功率任务完成的比例。人工接管率需要人工干预的次数。评估体系如何判断代理做得好不好需要设计评估任务Benchmark可以是基于规则输出是否包含关键词、基于模型用另一个LLM评估回答质量或基于人工的评估。5.5 架构模式扩展多代理协作对于复杂任务可以引入多个各司其职的代理如规划者、执行者、审查者让它们通过对话协同工作。AutoGen框架专门擅长于此。分层规划对于宏大目标代理可以先制定高层计划再为每个子目标创建子代理或进行更细粒度的规划。与外部系统集成代理需要融入现有的业务系统。这意味着它需要能监听消息队列如Kafka、调用内部RPC/HTTP服务、写入业务数据库等。此时代理更像一个智能的“流程自动化中心”。6. 常见问题排查清单在开发和使用AI代理过程中你会遇到各种问题。以下是一个快速排查清单。问题现象可能原因检查步骤解决方案代理不调用工具直接回答1. 系统提示词不清晰。2. 工具描述不够准确。3. 模型温度temperature过高输出随机。1. 检查_build_messages中的system prompt。2. 检查工具的name和description是否清晰。3. 检查模型调用参数temperature建议设为0.1-0.3。1. 优化提示词明确要求使用指定JSON格式。2. 参考OpenAI Function Calling文档优化工具描述。3. 降低temperature使用支持function calling的模型如gpt-3.5-turbo-1106以上版本。JSON解析失败1. LLM输出格式不符合预期。2. 输出中包含多余标记或思考过程。打印出完整的llm_output检查其格式。1. 在提示词中强制要求输出纯JSON并给出更严格的示例。2. 使用模型的“函数调用”功能而非文本生成。3. 在解析前用正则表达式提取JSON部分。工具执行出错1. 参数类型或数量不匹配。2. 工具内部代码有bug。3. 网络或依赖问题。1. 检查_execute_action中传递给工具的参数字典。2. 在工具execute方法内添加详细日志和异常捕获。3. 单独测试工具函数。1. 在工具调用前增加参数验证和转换逻辑。2. 修复工具内部代码。3. 确保工具所需的环境和依赖已正确安装。达到最大迭代次数1. 任务过于复杂。2. 代理陷入循环或无效动作。3. 工具无法满足任务需求。查看完整的历史日志观察代理在循环做什么。1. 增加max_iterations或设计任务分解机制。2. 引入循环检测当动作重复时强制改变策略或终止。3. 检查是否需要增加新的工具能力。API调用超时或失败1. 网络问题。2. API密钥无效或额度不足。3. 服务端限流。1. 检查网络连接。2. 检查API密钥和账单。3. 查看错误信息中是否包含rate_limit。1. 实现重试机制如指数退避。2. 更换有效的API密钥。3. 降低请求频率或升级API套餐。7. 生产环境部署与运维建议当你准备将代理投入实际业务时需要考虑以下方面服务化不要以脚本形式运行。将代理封装成Web服务如使用FastAPI提供标准的HTTP API便于集成和扩展。配置化管理将所有配置模型参数、工具列表、提示词模板外置到配置文件或配置中心支持热更新。异步与并发代理的循环可能耗时较长使用异步框架如asyncio避免阻塞并处理好并发请求。持久化存储将会话记忆、任务状态等持久化到数据库如Redis、PostgreSQL支持服务重启后恢复。可观测性集成日志系统如ELK、指标监控如Prometheus和分布式追踪如Jaeger全面掌握系统运行状态。版本管理与回滚对代理的核心逻辑、提示词、工具集进行版本控制确保出现问题能快速回滚。构建AI自动化代理是一个迭代过程。从本文的最小可行产品出发理解其核心循环和组件然后根据你的具体业务需求逐步强化它的记忆、工具、规划和稳定性。记住最强大的代理不是拥有最多功能的而是在特定领域内最可靠、最懂业务的那一个。 从零构建AI自动化代理：基于ReAct模式的核心原理与工程实践
在实际项目中将AI能力从简单的对话或生成工具转变为能够自主执行任务、管理流程的“自动化代理”是当前技术落地的一个关键跃迁。很多开发者学习了AI模型的基础调用但面对“构建一个能独立工作的AI代理”这个目标时却不知从何入手陷入工具堆砌或设计复杂的困境。本文旨在为初学者提供一条清晰的路径从核心概念理解到环境搭建再到实现一个具备基础感知、决策和执行能力的AI代理原型并最终探讨将其转化为可持续业务所需考虑的生产级要素。本文适合已经了解Python基础、对主流大模型API如OpenAI、智谱、DeepSeek等有初步调用经验并希望将这些能力系统化、自动化的开发者。我们将避开空泛的理论聚焦于可运行的代码、可复现的配置以及从开发到部署的完整闭环。读完本文你将能够搭建一个可以接收目标、分解任务、调用工具并持续运行的AI代理骨架并理解其扩展为复杂业务系统的核心模式。1. 理解AI自动化代理的核心从工具到智能体在开始写代码之前必须厘清几个关键概念这决定了我们构建的系统的设计方向和复杂度。1.1 AI代理与简单API调用的本质区别简单的AI API调用例如让模型生成一段文本或分析一张图片是一个“请求-响应”的瞬时过程。用户提供输入模型返回输出过程结束。而AI自动化代理则是一个具有状态持续性、目标导向性和自主行动能力的系统。状态持续性代理能记住之前的交互历史、执行结果和环境上下文并基于此做出后续决策。这通常通过维护一个“记忆”或“会话历史”模块来实现。目标导向性代理接收一个高级目标如“帮我分析本周销售数据并写一份报告”而非具体的指令。代理需要自己规划如何达成这个目标。自主行动能力代理不仅会“思考”生成文本还会“行动”。它可以自主调用外部工具如执行代码、查询数据库、调用第三方API、操作文件系统等。这是自动化得以实现的关键。一个典型的AI代理工作流遵循“感知-规划-执行-反思”的循环感知接收用户输入或环境信号。规划分析当前状态和目标决定下一步做什么思考可能分解为子任务。执行调用合适的工具或能力来完成规划出的动作。反思评估执行结果更新内部状态决定是继续下一个子任务、调整规划还是向用户请求澄清。1.2 关键组件与架构模式要构建这样一个系统我们需要设计几个核心组件大脑LLM Core负责所有的推理、规划和决策。通常由一个大语言模型担任它处理自然语言理解目标并生成下一步的行动计划通常以结构化格式如JSON。记忆Memory存储代理的历史对话、工具执行结果、学到的知识等。可以是简单的列表、向量数据库用于语义检索长期记忆或更复杂的图结构。工具Tools代理可以调用的函数或API集合。这是代理与外部世界交互的“手”和“脚”。例如search_web,execute_python_code,read_file,send_email。执行器Orchestrator控制整个循环流程的组件。它负责初始化代理、管理记忆、调用大脑进行规划、解析大脑的输出、调用对应的工具、处理工具返回结果并决定下一步。目前社区流行的架构模式主要有两种ReAct模式强调将“推理”和“行动”明确分离。代理的输出格式为Thought: ... Action: ... Action Input: ... 执行器解析后执行对应动作再将结果Observation: ...返回给代理进行下一轮思考。结构清晰易于调试。Plan-and-Execute模式代理先制定一个完整的计划任务列表然后按顺序或并行地执行这些任务。适合目标明确、步骤可预先分解的场景。对于初学者从ReAct模式入手更容易理解代理的运作机制。1.3 技术选型框架与模型手动从零搭建所有轮子是一个很好的学习过程但对于快速构建原型使用成熟的框架是更高效的选择。以下是一些常见选择及其特点框架/库语言特点适用场景LangChainPython/JS生态最丰富组件齐全文档庞大。概念较多学习曲线稍陡。快速构建复杂、生产级的AI应用需要大量集成。LlamaIndexPython专注于数据连接和检索RAG在构建知识库代理方面有优势。代理需要深度处理私有文档、知识库的场景。Semantic KernelC#/Python微软出品强于规划、插件技能管理和流程编排。.NET生态或需要复杂工作流编排的项目。AutoGenPython支持多代理对话和协作擅长构建代理团队。需要多个专业代理协同完成任务的场景。简单自建Python完全可控深度定制依赖少。学习原理、构建轻量级或特定场景的代理。对于初学者指南我们将采用“简单自建”结合LangChain的部分核心思想的方式以确保你能理解每一行代码在做什么同时又不至于陷入底层细节。模型方面我们将使用OpenAI GPT-4/3.5-Turbo的API作为“大脑”因为它提供了稳定且强大的推理能力。你也可以替换为其他兼容OpenAI API格式的模型服务。2. 环境准备与项目初始化我们将在Python环境中构建代理。请确保你的开发环境满足以下要求。2.1 基础环境与依赖安装首先确保你已安装Python推荐3.8以上版本。然后创建一个新的项目目录并初始化虚拟环境这能有效隔离依赖。# 创建项目目录 mkdir ai_agent_project cd ai_agent_project # 创建虚拟环境以venv为例 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate接下来安装核心依赖。我们将使用openai库与模型交互langchain库主要借用其优秀的“工具”抽象和调用模板但核心循环我们自己控制。# 安装核心库 pip install openai langchain # 安装可能用到的工具库示例 pip install requests # 用于调用网页API pip install python-dotenv # 用于管理环境变量2.2 配置模型API密钥为了调用大模型你需要一个API密钥。这里以OpenAI为例。永远不要将密钥硬编码在代码中。在项目根目录创建.env文件。在.env文件中填入你的密钥OPENAI_API_KEYsk-your-actual-api-key-here在代码中通过python-dotenv加载。同时创建一个config.py文件来集中管理配置# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) OPENAI_API_BASE os.getenv(OPENAI_API_BASE, https://api.openai.com/v1) # 支持自定义端点 MODEL_NAME os.getenv(MODEL_NAME, gpt-3.5-turbo) # 默认使用 3.5-turbo2.3 项目结构设计一个清晰的项目结构有助于后续扩展。建议如下ai_agent_project/ ├── .env # 环境变量列入.gitignore ├── config.py # 配置文件 ├── main.py # 主程序入口 ├── agent/ │ ├── __init__.py │ ├── core.py # 代理核心循环、大脑 │ ├── memory.py # 记忆模块 │ └── tools/ # 工具集目录 │ ├── __init__.py │ ├── base_tool.py # 工具基类 │ ├── web_search.py # 网页搜索工具示例 │ └── calculator.py # 计算器工具示例 └── utils/ └── __init__.py3. 构建核心组件工具、记忆与大脑现在我们从下往上构建代理的各个部件。3.1 定义工具Tools工具是代理能力的扩展。我们定义一个简单的工具基类所有具体工具都继承它。# agent/tools/base_tool.py from abc import ABC, abstractmethod from typing import Any, Dict class BaseTool(ABC): 工具基类 name: str # 工具名称代理通过这个名称来调用 description: str # 工具描述用于帮助LLM理解何时使用此工具 parameters: Dict[str, Any] # 工具所需的参数定义 def __init__(self, name: str, description: str): self.name name self.description description self.parameters {} abstractmethod def execute(self, **kwargs) - str: 执行工具的核心逻辑返回字符串格式的结果 pass def to_function_call_schema(self) - Dict: 将工具转换为OpenAI Function Calling所需的格式 return { type: function, function: { name: self.name, description: self.description, parameters: { type: object, properties: self.parameters, required: list(self.parameters.keys()) } } }接着实现两个简单的工具一个计算器和一个模拟的网页搜索。# agent/tools/calculator.py from .base_tool import BaseTool import math class CalculatorTool(BaseTool): def __init__(self): super().__init__( namecalculator, description执行数学计算。支持加()、减(-)、乘(*)、除(/)、乘方(**)、平方根(sqrt)等。 ) self.parameters { expression: {type: string, description: 数学表达式例如3 5 * 2 或 sqrt(16)} } def execute(self, expression: str) - str: try: # 警告在生产环境中直接eval用户输入是极度危险的 # 这里仅为演示实际应使用安全的表达式解析库如 ast.literal_eval 限制操作。 # 此处我们做一个极简的安全过滤和计算。 allowed_chars set(0123456789-*/.() sqrt) if not all(c in allowed_chars for c in expression): return 错误表达式中包含不安全字符。 # 替换sqrt为math.sqrt expression expression.replace(sqrt, math.sqrt) result eval(expression, {__builtins__: {}}, {math: math}) return f计算结果{result} except Exception as e: return f计算出错{e} # agent/tools/web_search.py from .base_tool import BaseTool import requests import json class WebSearchTool(BaseTool): def __init__(self, api_key: str None, search_engine: str google): # 注意这里需要接入真实的搜索API如SerpAPI、Google Custom Search等。 # 此处仅为模拟。 super().__init__( nameweb_search, description在互联网上搜索信息。当需要获取最新、未知或事实性信息时使用。 ) self.api_key api_key self.search_engine search_engine self.parameters { query: {type: string, description: 搜索关键词} } def execute(self, query: str) - str: # 模拟搜索返回 # 真实情况下这里应调用第三方API mock_results [ f关于{query}的搜索结果1..., f关于{query}的搜索结果2..., f关于{query}的搜索结果3... ] return f搜索完成。摘要{; .join(mock_results[:2])}在agent/tools/__init__.py中导出工具方便管理。# agent/tools/__init__.py from .calculator import CalculatorTool from .web_search import WebSearchTool __all__ [CalculatorTool, WebSearchTool]3.2 实现记忆Memory我们实现一个简单的对话历史记忆只保存最近的几轮交互。# agent/memory.py from typing import List, Dict, Any class SimpleConversationMemory: 简单的对话记忆保存固定轮数的历史 def __init__(self, max_turns: int 10): self.max_turns max_turns self.history: List[Dict[str, str]] [] # 每个元素是 {role: user/assistant/tool, content: ...} def add_user_message(self, message: str): self.history.append({role: user, content: message}) self._trim() def add_assistant_message(self, message: str): self.history.append({role: assistant, content: message}) self._trim() def add_tool_message(self, tool_name: str, result: str): # 工具执行结果通常以“Observation”的形式加入历史 self.history.append({role: tool, content: f[{tool_name}] 返回: {result}}) self._trim() def get_conversation_history(self) - List[Dict[str, str]]: 获取用于LLM提示的对话历史格式 return self.history.copy() def _trim(self): 保持历史记录不超过最大轮数 if len(self.history) self.max_turns: # 简单策略移除最早的一对用户/助手消息如果可能 # 更复杂的策略可以基于token数裁剪 self.history self.history[-self.max_turns:] def clear(self): self.history []3.3 构建大脑与执行器Core这是最核心的部分我们将实现一个遵循ReAct模式的代理循环。# agent/core.py import json import openai from typing import List, Dict, Any, Optional from .memory import SimpleConversationMemory from .tools.base_tool import BaseTool class SimpleAIAgent: def __init__(self, model_name: str gpt-3.5-turbo, api_key: str None, base_url: str None): self.client openai.OpenAI(api_keyapi_key, base_urlbase_url) self.model_name model_name self.memory SimpleConversationMemory() self.tools: Dict[str, BaseTool] {} # 工具名 - 工具实例的映射 self.max_iterations 10 # 防止无限循环 def register_tool(self, tool: BaseTool): 注册一个工具到代理 self.tools[tool.name] tool def _build_messages(self, user_input: str) - List[Dict[str, str]]: 构建发送给LLM的消息列表 messages [ { role: system, content: 你是一个有帮助的AI助手可以调用工具来解决问题。请遵循以下格式 思考你需要先思考当前情况和目标。 行动调用工具时请严格按照以下JSON格式输出 {action: 工具名称, action_input: {参数名: 参数值}} 如果你已经得到最终答案或者无需使用工具请直接输出最终答案。 } ] # 加入历史记忆 messages.extend(self.memory.get_conversation_history()) # 加入当前用户输入 messages.append({role: user, content: user_input}) return messages def _parse_llm_response(self, response: str) - Dict[str, Any]: 解析LLM的回复判断是最终答案还是工具调用 response response.strip() # 尝试解析JSON格式的工具调用 if response.startswith({) and response.endswith(}): try: data json.loads(response) if action in data and action_input in data: return {type: action, data: data} except json.JSONDecodeError: pass # 否则视为最终答案 return {type: final_answer, data: response} def _execute_action(self, action_name: str, action_input: Dict) - str: 执行工具调用 if action_name not in self.tools: return f错误未知工具 {action_name}。 tool self.tools[action_name] try: # 这里简化处理实际需要根据工具定义的参数进行校验和适配 result tool.execute(**action_input) return result except Exception as e: return f工具执行出错{e} def run(self, user_input: str) - str: 运行代理的主要循环 self.memory.add_user_message(user_input) print(f用户: {user_input}) for i in range(self.max_iterations): # 1. 构建消息并调用LLM messages self._build_messages(user_input) try: response self.client.chat.completions.create( modelself.model_name, messagesmessages, temperature0.1, # 低温度保证输出稳定易于解析 streamFalse, ) llm_output response.choices[0].message.content except Exception as e: return f调用模型API失败{e} print(f助手思考: {llm_output}) # 2. 解析LLM输出 parsed self._parse_llm_response(llm_output) if parsed[type] final_answer: final_answer parsed[data] self.memory.add_assistant_message(final_answer) print(f最终答案: {final_answer}) return final_answer elif parsed[type] action: action_data parsed[data] action_name action_data[action] action_input action_data[action_input] print(f执行动作: {action_name}, 输入: {action_input}) # 3. 执行工具 observation self._execute_action(action_name, action_input) print(f观察结果: {observation}) # 4. 将观察结果加入记忆开始下一轮循环 self.memory.add_assistant_message(llm_output) # 记录助手的“思考/行动” self.memory.add_tool_message(action_name, observation) # 记录工具结果 # 下一轮循环的“用户输入”实际上是上一步的观察结果但我们的消息构建方式已包含历史。 # 这里只需更新user_input为观察结果以便在下一轮构建消息时历史是最新的。 user_input observation # 简化处理实际ReAct中观察是作为下一轮LLM输入的一部分。 else: return 无法解析LLM的输出。 return f达到最大迭代次数({self.max_iterations})未完成目标。4. 组装与运行你的第一个AI代理现在我们将所有组件组装起来并运行一个完整的示例。4.1 主程序入口创建main.py作为启动文件。# main.py from config import Config from agent.core import SimpleAIAgent from agent.tools import CalculatorTool, WebSearchTool def main(): # 1. 初始化配置 config Config() # 2. 创建代理实例 agent SimpleAIAgent( model_nameconfig.MODEL_NAME, api_keyconfig.OPENAI_API_KEY, base_urlconfig.OPENAI_API_BASE ) # 3. 注册工具 agent.register_tool(CalculatorTool()) # 如果需要真实的搜索在这里传入API Key # agent.register_tool(WebSearchTool(api_keyyour_serpapi_key)) agent.register_tool(WebSearchTool()) # 使用模拟搜索 # 4. 运行代理 print( AI 代理已启动输入 quit 或 exit 退出 ) while True: try: user_input input(\n请输入你的问题或指令: ).strip() if user_input.lower() in [quit, exit, q]: print(再见) break if not user_input: continue final_answer agent.run(user_input) # run方法内部已打印过程这里可以额外处理最终答案 # 例如保存到文件或触发其他操作 except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f运行过程中发生错误{e}) if __name__ __main__: main()4.2 运行与验证在终端中确保虚拟环境已激活并且.env文件中的OPENAI_API_KEY已正确设置。python main.py程序启动后你可以尝试输入一些指令来测试代理的能力纯推理问题“法国的首都是哪里”代理可能直接回答无需调用工具。需要计算的问题“计算一下 15 的平方加上 28 除以 4 等于多少”代理应该识别出需要计算调用calculator工具并返回结果。需要搜索的问题“今天北京的天气怎么样”代理会调用web_search工具模拟并基于“搜索结果”给出回答。混合任务“先搜索一下爱因斯坦的生日然后计算他到今天假设是2024年1月1日活了多少天”代理需要先搜索获取生日信息再进行日期计算这里我们的计算器工具不支持日期可以观察代理如何处理。观察控制台输出你会看到类似以下的流程用户: 计算一下 15 的平方加上 28 除以 4 等于多少 助手思考: 思考用户需要计算一个数学表达式。我需要使用计算器工具。 行动{action: calculator, action_input: {expression: 15**2 28/4}} 执行动作: calculator, 输入: {expression: 15**2 28/4} 观察结果: 计算结果232.0 助手思考: 思考我已经得到了计算结果。 最终答案: 15的平方是22528除以4是7两者相加等于232。这个流程清晰地展示了代理的“思考-行动-观察”循环。5. 从原型到业务关键考量与进阶方向一个能跑通的原型距离一个可靠、可用的AI自动化代理业务还有很长的路。以下是几个必须深入考虑的方向。5.1 稳定性与可靠性提升错误处理与重试当前代码对工具执行和API调用错误处理很基础。生产环境需要更健壮的机制例如网络超时重试、模型API的退避重试、工具执行异常的回退方案等。循环检测与中断代理可能陷入死循环。除了设置max_iterations还可以检测重复的动作、无进展的对话并引入人工中断或策略调整。输出格式校验我们简单使用JSON解析很脆弱。应使用更严格的校验如Pydantic模型并为LLM提供更清晰的输出格式指导甚至采用支持“函数调用”功能的模型让模型直接返回结构化调用对象。5.2 记忆与上下文管理优化短期与长期记忆SimpleConversationMemory是短期记忆。对于需要记住大量历史或知识的场景需要引入向量数据库如Chroma, Pinecone作为长期记忆实现基于语义的检索。上下文窗口限制大模型有token限制。需要实现智能的上下文窗口管理例如总结冗长对话、选择性保留关键信息、使用更高效的token压缩技术。记忆的结构化将记忆分类存储如用户偏好、任务历史、学到的知识等便于代理更精准地利用。5.3 工具生态与安全性工具的动态注册与发现在大型系统中工具可能来自不同模块。需要设计机制让代理能动态发现和调用新工具。工具权限与安全这是重中之重。必须为工具调用设置严格的权限控制。例如execute_python_code工具必须在安全的沙箱环境中运行文件操作工具必须有路径白名单数据库操作工具必须有查询限制。永远不要在生产环境直接eval用户输入。工具描述的质量工具的描述description和参数定义直接影响LLM调用的准确性。需要精心编写并可以通过少量示例few-shot来提升模型的理解。5.4 监控、评估与成本控制全链路日志记录每一轮的用户输入、LLM思考、工具调用、工具结果、最终输出。这是排查问题、分析代理行为和改进提示词的基石。关键指标监控延迟每轮循环的耗时。成本API调用token消耗。工具调用分布哪些工具被频繁使用。成功率任务完成的比例。人工接管率需要人工干预的次数。评估体系如何判断代理做得好不好需要设计评估任务Benchmark可以是基于规则输出是否包含关键词、基于模型用另一个LLM评估回答质量或基于人工的评估。5.5 架构模式扩展多代理协作对于复杂任务可以引入多个各司其职的代理如规划者、执行者、审查者让它们通过对话协同工作。AutoGen框架专门擅长于此。分层规划对于宏大目标代理可以先制定高层计划再为每个子目标创建子代理或进行更细粒度的规划。与外部系统集成代理需要融入现有的业务系统。这意味着它需要能监听消息队列如Kafka、调用内部RPC/HTTP服务、写入业务数据库等。此时代理更像一个智能的“流程自动化中心”。6. 常见问题排查清单在开发和使用AI代理过程中你会遇到各种问题。以下是一个快速排查清单。问题现象可能原因检查步骤解决方案代理不调用工具直接回答1. 系统提示词不清晰。2. 工具描述不够准确。3. 模型温度temperature过高输出随机。1. 检查_build_messages中的system prompt。2. 检查工具的name和description是否清晰。3. 检查模型调用参数temperature建议设为0.1-0.3。1. 优化提示词明确要求使用指定JSON格式。2. 参考OpenAI Function Calling文档优化工具描述。3. 降低temperature使用支持function calling的模型如gpt-3.5-turbo-1106以上版本。JSON解析失败1. LLM输出格式不符合预期。2. 输出中包含多余标记或思考过程。打印出完整的llm_output检查其格式。1. 在提示词中强制要求输出纯JSON并给出更严格的示例。2. 使用模型的“函数调用”功能而非文本生成。3. 在解析前用正则表达式提取JSON部分。工具执行出错1. 参数类型或数量不匹配。2. 工具内部代码有bug。3. 网络或依赖问题。1. 检查_execute_action中传递给工具的参数字典。2. 在工具execute方法内添加详细日志和异常捕获。3. 单独测试工具函数。1. 在工具调用前增加参数验证和转换逻辑。2. 修复工具内部代码。3. 确保工具所需的环境和依赖已正确安装。达到最大迭代次数1. 任务过于复杂。2. 代理陷入循环或无效动作。3. 工具无法满足任务需求。查看完整的历史日志观察代理在循环做什么。1. 增加max_iterations或设计任务分解机制。2. 引入循环检测当动作重复时强制改变策略或终止。3. 检查是否需要增加新的工具能力。API调用超时或失败1. 网络问题。2. API密钥无效或额度不足。3. 服务端限流。1. 检查网络连接。2. 检查API密钥和账单。3. 查看错误信息中是否包含rate_limit。1. 实现重试机制如指数退避。2. 更换有效的API密钥。3. 降低请求频率或升级API套餐。7. 生产环境部署与运维建议当你准备将代理投入实际业务时需要考虑以下方面服务化不要以脚本形式运行。将代理封装成Web服务如使用FastAPI提供标准的HTTP API便于集成和扩展。配置化管理将所有配置模型参数、工具列表、提示词模板外置到配置文件或配置中心支持热更新。异步与并发代理的循环可能耗时较长使用异步框架如asyncio避免阻塞并处理好并发请求。持久化存储将会话记忆、任务状态等持久化到数据库如Redis、PostgreSQL支持服务重启后恢复。可观测性集成日志系统如ELK、指标监控如Prometheus和分布式追踪如Jaeger全面掌握系统运行状态。版本管理与回滚对代理的核心逻辑、提示词、工具集进行版本控制确保出现问题能快速回滚。构建AI自动化代理是一个迭代过程。从本文的最小可行产品出发理解其核心循环和组件然后根据你的具体业务需求逐步强化它的记忆、工具、规划和稳定性。记住最强大的代理不是拥有最多功能的而是在特定领域内最可靠、最懂业务的那一个。