1. 项目概述为什么我们需要亲手搭建一个AI Agent框架最近几年AI Agent这个概念火得不行几乎成了每个技术讨论的焦点。但说实话市面上关于Agent的讨论要么是停留在“它很智能、能自主完成任务”这种概念层面要么就是直接甩给你一个OpenAI的API调用示例告诉你用几个函数调用Function Calling就能实现一个简单的Agent。这中间存在一个巨大的断层从一个简单的API调用到一个真正健壮、可扩展、能处理复杂任务的AI Agent系统到底该怎么走这其中的理论支撑是什么工程上的坑又有哪些这就是我动手写这个系列的原因。我不满足于只是“使用”别人封装好的Agent框架我更想搞清楚它的“骨架”和“肌肉”是怎么长出来的。从零开始实现一个AI Agent框架不仅仅是为了炫技其核心价值在于深度理解与掌控。当你亲手设计它的思考循环ReAct, CoT实现工具调用Tool Calling的调度处理长期记忆Memory的存储与检索并搭建一个可靠的任务规划Planning与执行Execution引擎时你才会真正明白那些现成框架比如LangChain、AutoGen里每个设计抉择背后的深意。你会知道为什么有些任务它处理得好有些却会陷入死循环你会懂得如何为你的特定业务场景定制最合适的Agent架构而不是被框架的抽象所限制。这个项目适合谁我认为有三类朋友会特别有收获一是对AI应用开发有浓厚兴趣不满足于仅仅调用API的开发者二是希望将AI能力深度集成到自家产品中需要高度定制化Agent系统的技术负责人或架构师三是任何想要深入理解“智能体”究竟如何运作的研究者或学生。我们将从最基础的理论模型讲起然后用代码一步步把它们构建出来过程中遇到的每一个错误和解决方案我都会毫无保留地分享出来。2. 核心理论基石构建AI Agent的四大支柱在动手写代码之前我们必须把地基打牢。一个功能完整的AI Agent框架其理论核心可以归纳为四个相互关联的支柱规划Planning、记忆Memory、工具使用Tool Use和行动Action。理解这四者的关系是设计任何Agent系统的前提。2.1 规划模块从目标到步骤的拆解艺术规划本质上是一个问题分解与路径搜索的过程。对于AI Agent而言给定一个模糊的用户指令如“帮我分析一下上季度的销售数据并给出下季度的增长建议”它需要自动将其分解为一系列可执行的具体步骤。这里涉及几个关键理论链式思考Chain-of-Thought, CoT与ReAct范式这是当前让大语言模型LLM进行规划最主流的方法。CoT要求模型展示其推理的中间步骤。而ReActReasoning Acting则更进一步它将推理Reason和行动Act交织在一起。Agent不是一次性规划所有步骤而是采用“思考一步执行一步观察结果再思考下一步”的循环。这种方式更贴近人类解决问题的方式也更能应对执行过程中的不确定性。在我们的框架设计中ReAct循环将是核心执行引擎。任务分解Task Decomposition如何让LLM可靠地将大任务拆成小任务常见策略有1按子目标分解让LLM识别出达成总目标所需的中间子目标。2按工具或能力分解根据可用的工具集如“查询数据库”、“生成图表”、“发送邮件”来反向推导步骤。3按时间或流程顺序分解适用于有严格先后顺序的任务。在实现时我们通常通过精心设计的系统提示词System Prompt来引导LLM进行这种分解。规划算法对于复杂任务简单的逐步分解可能不够。我们可以引入更经典的算法思想如基于树的搜索广度优先、深度优先或基于图的规划。例如Agent可以将每个可能的行动步骤及其预期结果作为一个节点构建一个搜索空间然后寻找从初始状态到目标状态的最优路径。虽然这需要更多的计算和与环境的交互但对于需要多步深度规划的场景至关重要。注意规划模块最怕陷入“空转”。即Agent不停地进行推理却无法产生任何有效的行动步骤。为了防止这一点我们需要在提示词中明确约束如“必须从可用工具列表中选择行动”并在代码层面设置最大推理步数或超时机制。2.2 记忆模块赋予Agent持续性的关键一个没有记忆的Agent每次交互都是全新的开始这显然不是“智能体”该有的样子。记忆模块负责存储、组织和检索历史信息是Agent实现持续对话、学习用户偏好、积累领域知识的基础。我们可以从三个时间维度来设计记忆短期记忆/工作记忆这通常指当前会话的上下文。由于LLM有上下文窗口长度限制我们需要一个高效的上下文窗口管理策略。简单的做法是使用“滑动窗口”只保留最近的N轮对话。更高级的策略则涉及对历史对话进行摘要Summarization将冗长的历史压缩成精炼的要点再放入上下文从而在有限的窗口内承载更长时间跨度的信息。长期记忆这是Agent的“知识库”或“经验库”。所有超越当前会话的有用信息都应存储在这里。实现长期记忆的核心技术是向量数据库Vector Database。我们将对话、执行结果、学到的知识等文本转换成向量Embedding存入向量库。当需要回忆时将当前问题也转换成向量进行相似度搜索召回最相关的记忆片段再注入到当前上下文中。这解决了传统数据库基于关键词匹配的局限性实现了基于语义的关联回忆。记忆的层次与结构记忆不是扁平的。我们可以设计不同的“记忆体”例如核心事实记忆用户提供的明确信息、对话历史记忆、自我反思记忆Agent对自己过去行动成功或失败的分析、技能记忆如何成功使用某个工具的经验。在检索时可以根据当前任务类型决定优先从哪种记忆体中搜索。2.3 工具使用模块Agent与世界的接口Agent再聪明如果无法操作外部系统也只能是个“思想家”。工具使用模块是Agent能力的放大器。其核心理论是将外部API、函数、甚至其他软件的能力抽象成统一的、LLM可理解和调用的“工具”描述。工具的描述与发现每个工具都需要一个清晰的“说明书”通常包括工具名称、功能描述、必需的输入参数及其类型、格式、说明、可能的输出。这个说明书必须以结构化数据如JSON Schema的形式存在并能被动态地加载到Agent的上下文中。这就是所谓的工具发现Tool Discovery机制。工具的选择与调用当Agent决定要采取行动时它需要1从当前可用的工具列表中根据任务描述选择最合适的一个或多个工具。2根据工具说明书生成符合要求的调用参数。3以安全、可控的方式执行该工具对应的代码或API请求。这里的关键是参数验证与错误处理。LLM生成的参数可能是错误的、不完整的框架必须在调用前进行严格的校验并在调用失败时提供清晰的错误信息反馈给Agent以便其进行修正ReAct中的“观察”阶段。工具的扩展性框架必须支持轻松地注册新工具。理想情况下开发者只需要定义一个Python函数并用装饰器或配置文件描述其元信息框架就能自动将其纳入Agent的工具库。这涉及到灵活的插件化架构设计。2.4 行动与执行循环将计划付诸实践这是将前三个模块串联起来的“总控制器”通常体现为ReAct循环。其标准流程如下观察ObserveAgent接收当前的用户输入和来自环境的最新反馈如上一步工具执行的结果或错误。思考Think基于观察、历史记忆和可用工具LLM进行推理决定下一步该做什么。输出应被严格格式化例如必须明确指示是“继续思考”还是“调用某个工具”。行动Act如果决定调用工具则框架解析LLM的输出提取工具名和参数执行工具调用。循环将行动的结果作为新的“观察”进入下一轮循环直到LLM输出“最终答案”或达到终止条件如任务完成、步数超限。这个循环的实现质量直接决定了Agent的可靠性和效率。我们需要处理很多细节比如如何解析LLM非结构化的输出如何管理循环状态如何优雅地处理超时和中断3. 框架设计与核心模块实现有了理论指导我们就可以开始设计框架的顶层架构了。我们的目标是构建一个轻量级、模块化、易于扩展的框架我将其命名为“ThinkFlow”。整个框架的核心是一个中央调度器Orchestrator它协调着各个模块的运作。3.1 整体架构与模块划分ThinkFlow 的核心架构围绕一个主循环展开主要包括以下组件Agent Core封装了与大语言模型LLM的交互负责生成推理和决策。Planner规划模块负责将复杂任务分解为子任务序列。初期我们可以实现一个基于提示词的简单规划器。Memory Manager记忆管理器统一管理短期会话缓存和长期向量存储的读写。Tool Registry工具注册表一个全局的单例负责所有工具的注册、描述和查找。Executor执行器负责安全地调用Tool Registry中的工具并处理执行结果和异常。Orchestrator调度器实现上述的ReAct循环串联所有模块。它们之间的关系如下图所示概念图用户请求首先到达OrchestratorOrchestrator调用Planner进行任务分解对于复杂任务然后进入ReAct循环。在循环中Orchestrator将当前状态用户输入、历史、可用工具交给Agent Core生成思考根据思考决定是调用Executor执行工具还是直接给出答案。整个过程中Memory Manager负责提供相关记忆和保存新的经验。3.2 核心模块代码实现拆解接下来我们深入到几个最关键模块的代码层面。我会用Python来演示因为其生态丰富易于理解。首先是工具系统的实现。这是Agent能力的边界。我们设计一个工具装饰器让开发者能轻松地将任何函数变成Agent可用的工具。# tool.py import inspect import json from typing import Callable, Dict, Any, get_type_hints class Tool: 工具类封装一个可调用函数及其元数据 def __init__(self, func: Callable, name: str, description: str): self.func func self.name name self.description description self.parameters self._extract_parameters(func) def _extract_parameters(self, func: Callable) - Dict[str, Any]: 从函数签名中提取参数信息生成JSON Schema格式 sig inspect.signature(func) type_hints get_type_hints(func) parameters {} for param_name, param in sig.parameters.items(): if param_name self: continue param_info {type: string} # 默认类型 if param_name in type_hints: type_str str(type_hints[param_name]) # 简化类型映射实际项目需要更完善的映射 if int in type_str: param_info[type] integer elif float in type_str: param_info[type] number elif bool in type_str: param_info[type] boolean elif list in type_str or List in type_str: param_info[type] array if param.default ! inspect.Parameter.empty: param_info[default] param.default param_info[description] f参数 {param_name} # 可通过docstring获取更佳描述 parameters[param_name] param_info return parameters def to_schema(self) - Dict[str, Any]: 生成OpenAI Function Calling兼容的工具模式 return { type: function, function: { name: self.name, description: self.description, parameters: { type: object, properties: self.parameters, required: [k for k, v in self.parameters.items() if default not in v] } } } def execute(self, **kwargs): 执行工具调用 # 这里可以加入参数验证、权限检查、调用日志等 try: result self.func(**kwargs) return {status: success, data: result} except Exception as e: return {status: error, message: str(e)} class ToolRegistry: 全局工具注册表 _instance None _tools: Dict[str, Tool] {} def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) return cls._instance classmethod def register(cls, name: str None, description: str ): 装饰器用于注册工具函数 def decorator(func: Callable): tool_name name or func.__name__ tool Tool(func, tool_name, description) cls._tools[tool_name] tool return func return decorator classmethod def get_tool(cls, name: str) - Tool: return cls._tools.get(name) classmethod def get_all_schemas(cls): return [tool.to_schema() for tool in cls._tools.values()] # 使用示例 ToolRegistry.register(description计算两个数字的和) def add(a: int, b: int) - int: 返回a与b的和 return a b ToolRegistry.register(namesearch_web, description在互联网上搜索信息) def search(query: str) - str: # 模拟搜索实际应调用搜索引擎API return f关于{query}的搜索结果...这个工具系统实现了几个关键点1利用装饰器实现无侵入式的工具注册。2自动从函数签名和类型注解中提取参数信息生成标准化的模式描述。3提供了统一的执行和错误处理接口。其次是实现一个基于向量数据库的长期记忆模块。这里我们以ChromaDB为例它是一个轻量级的向量数据库。# memory.py from typing import List, Dict, Any import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import uuid class LongTermMemory: 基于向量数据库的长期记忆 def __init__(self, persist_directory: str ./memory_db): # 初始化嵌入模型 self.embedder SentenceTransformer(all-MiniLM-L6-v2) # 轻量级句子嵌入模型 # 初始化Chroma客户端 self.client chromadb.PersistentClient( pathpersist_directory, settingsSettings(anonymized_telemetryFalse) ) # 获取或创建集合类似于表 self.collection self.client.get_or_create_collection(nameagent_memories) def _generate_embedding(self, text: str) - List[float]: 生成文本的向量嵌入 return self.embedder.encode(text).tolist() def store(self, content: str, metadata: Dict[str, Any] None): 存储一段记忆 if metadata is None: metadata {} embedding self._generate_embedding(content) id str(uuid.uuid4()) self.collection.add( embeddings[embedding], documents[content], metadatas[metadata], ids[id] ) return id def retrieve(self, query: str, n_results: int 5) - List[Dict[str, Any]]: 根据查询检索最相关的记忆 query_embedding self._generate_embedding(query) results self.collection.query( query_embeddings[query_embedding], n_resultsn_results ) memories [] # 解析返回结果 if results[documents]: for i in range(len(results[documents][0])): memories.append({ content: results[documents][0][i], metadata: results[metadatas][0][i], distance: results[distances][0][i] }) return memories def clear(self): 清空所有记忆谨慎使用 self.client.delete_collection(nameagent_memories) self.collection self.client.create_collection(nameagent_memories)这个记忆模块封装了向量的生成、存储和检索。在实际使用中我们可以在Agent完成一个重要任务后将任务总结和关键结果store起来。当遇到新任务时先用retrieve查找相关历史经验并将其作为上下文提供给LLM从而实现“经验复用”。4. 核心执行引擎ReAct循环的工程实现有了工具和记忆现在我们需要一个“大脑”来驱动一切。这个大脑就是ReAct循环执行引擎。它的职责是维持Agent的思考-行动周期并处理所有状态流转。4.1 状态管理与循环控制首先我们需要定义Agent在循环中的状态。一个典型的状态对象应包含# orchestrator.py from dataclasses import dataclass, field from typing import List, Dict, Any, Optional dataclass class AgentState: Agent在一次运行中的完整状态 user_input: str # 原始用户输入 task_description: str # 当前要处理的任务描述可能被分解过 full_history: List[Dict[str, str]] # 完整的对话历史格式[{role: user/assistant/system, content: ...}] recent_observations: List[str] # 最近几步的观察工具执行结果等 available_tools_schema: List[Dict] # 当前可用的工具模式列表 # 从长期记忆中检索到的相关上下文 retrieved_memories: List[str] field(default_factorylist) # 当前循环的步数 step_count: int 0 # 最终答案如果已得出 final_answer: Optional[str] None接下来我们实现Orchestrator的核心循环逻辑。这里的关键是设计一个强引导性的系统提示词让LLM严格按照我们期望的格式如JSON输出以便程序化解析。# orchestrator.py (续) import json import re class ReactOrchestrator: def __init__(self, llm_client, memory: LongTermMemory, max_steps20): self.llm llm_client self.memory memory self.max_steps max_steps self.tool_registry ToolRegistry() def _build_system_prompt(self, state: AgentState) - str: 构建引导ReAct循环的系统提示词 prompt f你是一个智能助手可以通过使用工具来解决问题。请遵循以下步骤 1. 分析当前任务和可用工具。 2. 如果需要使用工具请严格按照以下JSON格式回应 {{thought: 你的推理过程, action: tool_name, action_input: {{arg1: value1, ...}}}} 3. 如果任务已完成可以给出最终答案请使用格式 {{thought: 总结推理, action: final_answer, action_input: 你的最终答案}} 当前任务{state.task_description} 可用工具 {json.dumps(state.available_tools_schema, indent2, ensure_asciiFalse)} 历史观察最近几步 {chr(10).join(state.recent_observations[-3:]) if state.recent_observations else 无} 相关历史经验 {chr(10).join(state.retrieved_memories) if state.retrieved_memories else 无} 请开始你的思考。 return prompt def _parse_llm_output(self, text: str) - Dict[str, Any]: 解析LLM的输出提取结构化的动作指令 # 尝试从文本中提取JSON块 json_pattern rjson\s*(.*?)\s*|\s*(.*?)\s*|\{.*\} matches re.findall(json_pattern, text, re.DOTALL) json_str None for match in matches: json_str match[0] or match[1] or match[2] if json_str: break if not json_str: # 如果没有找到代码块尝试直接查找最外层的花括号 try: start text.find({) end text.rfind(}) 1 if start ! -1 and end ! 0: json_str text[start:end] except: json_str None if json_str: try: return json.loads(json_str) except json.JSONDecodeError: # 如果解析失败回退到启发式解析或报错 pass # 如果无法解析为JSON则假设模型直接给出了最终答案 return {thought: text, action: final_answer, action_input: text} def run(self, user_input: str) - str: 主运行循环 # 1. 初始化状态 state AgentState( user_inputuser_input, task_descriptionuser_input, full_history[{role: user, content: user_input}], recent_observations[], available_tools_schemaself.tool_registry.get_all_schemas(), retrieved_memories[] ) # 2. 从长期记忆中检索相关上下文 if self.memory: memories self.memory.retrieve(user_input, n_results3) state.retrieved_memories [m[content] for m in memories] # 3. ReAct 循环 for step in range(self.max_steps): state.step_count step 1 print(f\n 步骤 {step 1} ) # 3.1 构建当前轮次的对话上下文 messages [] # 添加系统提示词 system_prompt self._build_system_prompt(state) messages.append({role: system, content: system_prompt}) # 添加完整对话历史可选可只添加摘要 messages.extend(state.full_history[-6:]) # 保留最近几轮 # 3.2 调用LLM进行思考 llm_response self.llm.chat_completion(messages) assistant_reply llm_response[choices][0][message][content] state.full_history.append({role: assistant, content: assistant_reply}) print(f助理回复{assistant_reply}) # 3.3 解析LLM输出得到动作指令 action_dict self._parse_llm_output(assistant_reply) thought action_dict.get(thought, ) action action_dict.get(action, ) action_input action_dict.get(action_input, {}) print(f解析结果思考 - {thought} 动作 - {action}) # 3.4 执行动作 if action final_answer: state.final_answer str(action_input) # 可选将成功的任务经验存入长期记忆 if self.memory and state.final_answer: summary f任务{state.task_description}\n结果{state.final_answer[:200]}... self.memory.store(summary, metadata{type: successful_task}) break elif action in [tool.name for tool in self.tool_registry._tools.values()]: # 执行工具调用 tool self.tool_registry.get_tool(action) if tool: try: # 确保action_input是字典 if isinstance(action_input, str): try: action_input json.loads(action_input) except: action_input {input: action_input} result tool.execute(**action_input) observation f工具 {action} 执行结果{result} except Exception as e: observation f工具 {action} 执行出错{str(e)} else: observation f错误未知工具 {action} # 将观察结果加入状态 state.recent_observations.append(observation) state.full_history.append({role: user, content: observation}) print(f观察{observation}) else: # 无法识别的动作作为观察反馈给LLM observation f无法理解的动作指令{action}。请使用指定的JSON格式并选择可用的工具或给出最终答案。 state.recent_observations.append(observation) state.full_history.append({role: user, content: observation}) print(f观察{observation}) # 3.5 检查步数限制 if step self.max_steps - 1: state.final_answer f达到最大步数限制{self.max_steps}任务未完成。最后状态{state.recent_observations[-1] if state.recent_observations else 无进展} break return state.final_answer or 任务执行未产生明确结果。这个Orchestrator实现了一个完整的、可运行的ReAct循环。它包含了状态管理、提示词工程、LLM输出解析、工具执行调度以及循环终止判断。这是整个框架的“心脏”。4.2 与LLM的集成策略上面的代码中llm_client是一个抽象接口。在实际项目中我们需要适配不同的LLM提供商如OpenAI、Anthropic、国内大模型等。一个好的设计是定义一个统一的LLM客户端接口# llm_client.py from abc import ABC, abstractmethod import openai # 示例 class BaseLLMClient(ABC): abstractmethod def chat_completion(self, messages: List[Dict], **kwargs) - Dict[str, Any]: 发送聊天补全请求返回统一格式的响应 pass class OpenAIClient(BaseLLMClient): def __init__(self, model: str gpt-4, api_key: str None): self.model model self.client openai.OpenAI(api_keyapi_key) def chat_completion(self, messages: List[Dict], **kwargs) - Dict[str, Any]: response self.client.chat.completions.create( modelself.model, messagesmessages, **kwargs ) # 将响应格式化为统一的字典格式 return { id: response.id, choices: [{ message: { role: choice.message.role, content: choice.message.content } } for choice in response.choices] }这样Orchestrator就与具体的LLM实现解耦了我们可以轻松切换后端模型。5. 实战演练构建一个数据分析助手Agent理论和技术都讲完了现在我们用一个完整的例子把所有的模块串起来构建一个能实际运行的“数据分析助手”Agent。这个Agent的目标是用户用自然语言提出一个数据分析需求比如“帮我分析一下销售数据找出最畅销的产品”Agent能自动调用相应的工具读取文件、执行查询、绘制图表来完成分析并给出结论。5.1 定义领域专用工具首先我们需要为这个数据分析场景创建几个工具。假设我们有一个CSV格式的销售数据文件sales.csv。# data_analysis_tools.py import pandas as pd import matplotlib.pyplot as plt import io import base64 from ToolRegistry import ToolRegistry ToolRegistry.register(description读取指定路径的CSV文件返回数据预览) def read_csv(file_path: str, n_rows: int 5) - str: 读取CSV文件并返回前几行预览 try: df pd.read_csv(file_path) preview df.head(n_rows).to_string() shape df.shape return f文件 {file_path} 读取成功。数据形状{shape}。前{n_rows}行预览\n{preview} except Exception as e: return f读取文件失败{str(e)} ToolRegistry.register(description对数据进行SQL-like的查询例如按列分组求和) def query_data(file_path: str, group_by: str, aggregate: str sum) - str: 对数据进行简单的分组聚合查询 try: df pd.read_csv(file_path) if group_by not in df.columns: return f错误数据中不存在列 {group_by}。 if aggregate sum: result df.groupby(group_by).sum(numeric_onlyTrue) elif aggregate mean: result df.groupby(group_by).mean(numeric_onlyTrue) elif aggregate count: result df.groupby(group_by).size() else: return f不支持的聚合操作{aggregate}请使用 sum, mean 或 count。 return f按 {group_by} 分组执行 {aggregate} 聚合的结果\n{result.to_string()} except Exception as e: return f查询失败{str(e)} ToolRegistry.register(nameplot_chart, description根据数据生成折线图或柱状图返回图片的base64编码) def plot_chart(file_path: str, x_column: str, y_column: str, chart_type: str line) - str: 生成图表并返回base64字符串 try: df pd.read_csv(file_path) if x_column not in df.columns or y_column not in df.columns: return f错误指定的列不存在。 plt.figure(figsize(10, 6)) if chart_type line: plt.plot(df[x_column], df[y_column]) elif chart_type bar: plt.bar(df[x_column], df[y_column]) else: return f不支持的图表类型{chart_type}请使用 line 或 bar。 plt.xlabel(x_column) plt.ylabel(y_column) plt.title(f{y_column} vs {x_column}) plt.tight_layout() # 将图表保存到内存缓冲区并转换为base64 buf io.BytesIO() plt.savefig(buf, formatpng) plt.close() buf.seek(0) img_base64 base64.b64encode(buf.read()).decode(utf-8) return fdata:image/png;base64,{img_base64} except Exception as e: return f生成图表失败{str(e)} finally: plt.close(all)5.2 组装并运行Agent现在我们把所有部件组装起来并运行一个完整的任务。# main.py from llm_client import OpenAIClient from memory import LongTermMemory from orchestrator import ReactOrchestrator import data_analysis_tools # 导入工具模块以完成注册 def main(): # 1. 初始化组件 llm OpenAIClient(modelgpt-4) # 请替换为你的API Key memory LongTermMemory() orchestrator ReactOrchestrator(llm, memory, max_steps15) # 2. 用户输入一个复杂任务 user_query 请分析项目根目录下的 sales.csv 文件。 首先告诉我这个文件里有什么数据。 然后找出哪个产品product列的总销售额sales列最高。 最后请为我生成一个展示每个产品总销售额的柱状图。 print(用户请求, user_query) print(*50) # 3. 运行Agent final_answer orchestrator.run(user_query) # 4. 输出最终结果 print(\n *50) print(任务完成最终答案) print(final_answer) # 处理可能包含图片的答案 if final_answer and final_answer.startswith(data:image/png;base64,): # 如果是图片可以保存到本地 import base64 img_data final_answer.split(,)[1] with open(output_chart.png, wb) as f: f.write(base64.b64decode(img_data)) print(图表已保存为 output_chart.png) else: print(final_answer) if __name__ __main__: main()当你运行这段代码时你会看到控制台打印出Agent的思考过程步骤1LLM分析任务发现需要先读取文件。它输出JSON调用read_csv工具。步骤2Orchestrator执行工具得到数据预览并将其作为“观察”反馈给LLM。步骤3LLM根据预览知道下一步需要按产品分组求和。它调用query_data工具参数为group_byproduct, aggregatesum。步骤4得到查询结果后LLM决定生成图表。它调用plot_chart工具指定x_columnproduct, y_columnsales, chart_typebar。步骤5收到图表数据后LLM综合所有信息给出最终答案并可能附上分析结论。这个过程完美地展示了ReAct循环思考 - 行动 - 观察 - 再思考。我们的框架成功地协调了LLM的推理能力和外部工具的执行能力。6. 避坑指南与性能优化实战在从零搭建和实际使用这个框架的过程中我踩过不少坑也总结出一些让Agent更稳定、更高效的技巧。这部分是你在官方文档里很难看到的“实战经验”。6.1 提示词工程让LLM乖乖听话LLM不按格式输出是ReAct循环失败的主要原因。除了在代码中做健壮的解析提示词的设计至关重要。结构化输出强制在系统提示词中明确要求LLM以指定格式如JSON回复并给出多个正反示例。例如你必须以JSON格式回应。例如调用工具{thought: ..., action: tool_name, action_input: {...}}。给出最终答案{thought: ..., action: final_answer, action_input: ...}。任何其他格式都将导致错误。思维链CoT引导鼓励LLM在thought字段中详细推理。例如“在thought中请逐步解释你为什么选择这个工具以及你期望得到什么结果。” 这不仅能提高动作的准确性也便于我们调试。工具描述精细化给工具的description和参数描述写得越详细、越自然LLM调用得就越准。避免使用技术性过强的语言用LLM能理解的自然语言描述功能。例如description计算给定列表中所有数字的总和比descriptionsum function要好得多。6.2 错误处理与循环稳定性一个生产级的Agent框架必须能优雅地处理各种异常。工具执行异常工具可能因为网络、参数错误、权限等问题失败。我们的Tool.execute()方法已经返回了包含状态的字典。Orchestrator需要将这些错误信息清晰地反馈给LLM让它有机会调整策略。例如将错误信息格式化为“工具X执行失败原因为Y。请检查你的输入参数或尝试其他方法。”LLM输出解析失败即使用户提示词LLM也可能输出非结构化内容。我们的_parse_llm_output函数尝试了多种提取JSON的方法。如果全部失败最后的兜底策略是将整个输出当作final_answer或者将其作为错误观察反馈给LLM要求它重试。可以设置一个重试计数器避免无限循环。循环停滞检测Agent有时会陷入“死循环”比如反复调用同一个工具且得到相同结果。解决方法有1在状态中记录最近N次的动作和观察如果检测到重复模式则中断循环并提示用户。2设置最大步数这是必须的保险丝。6.3 记忆系统的优化技巧记忆系统用不好反而会成为负担。存储策略不要什么都存。只存储成功的任务总结、重要的用户信息如偏好以及从失败中学习的教训。存储时可以生成一个更概括的摘要而不是原始对话的罗列。检索优化简单基于向量的相似度搜索有时会召回不相关的记忆。可以尝试1元数据过滤在存储时为记忆打上标签如task_type: data_analysis检索时结合语义相似度和标签过滤。2重排序Rerank先用向量数据库召回较多结果如10个再用一个小型的、更精确的交叉编码器模型对它们进行重排序选出最相关的3-5个。上下文窗口管理对于长对话将完整的记忆全部放入上下文是不现实的。可以采用“摘要最近记录”的混合模式。定期例如每10轮对话让LLM对之前的对话历史生成一个简短摘要然后用这个摘要替代之前的大段历史腾出空间给新的对话。6.4 性能与成本考量频繁调用LLM和向量数据库成本和延迟是必须考虑的问题。LLM调用优化缓存对相同的提示词或提示词哈希的LLM响应进行缓存。这在开发调试和重复性问题中能节省大量成本。使用更小的模型对于简单的工具选择、参数提取等任务可以尝试使用更小、更快的模型如GPT-3.5-Turbo而只在需要复杂推理时使用大模型如GPT-4。并行工具调用如果多个工具调用之间没有依赖关系可以探索让LLM一次性输出多个工具调用请求然后并行执行减少循环次数。向量检索优化分块Chunking存储长文本记忆时将其分成有重叠的小块如256个token一块。这样检索时能更精准地定位到相关片段而不是返回一整篇不相关的文档。选择合适的索引ChromaDB、Pinecone等向量数据库都支持多种索引算法如HNSW。根据数据规模和查询延迟要求进行选择和调参。从零开始实现一个AI Agent框架就像亲手组装一台精密的钟表。你不仅知道了指针如何走动更理解了每一个齿轮的咬合与发条的张力。这个过程让我深刻体会到现成的框架固然方便但隐藏了太多关键的细节和设计权衡。当你自己实现一遍你会对“智能体”的脆弱性、对提示词的敏感性、对错误处理的必要性有全新的认识。这套代码只是一个起点你可以在此基础上继续探索多Agent协作、更复杂的规划算法如基于LLM的Tree of Thoughts、动态工具学习等高级特性。最重要的是你拥有了根据具体业务需求定制和优化每一个环节的能力这才是核心竞争力所在。 从零构建AI Agent框架：深入解析ReAct循环、工具调用与长期记忆实现
1. 项目概述为什么我们需要亲手搭建一个AI Agent框架最近几年AI Agent这个概念火得不行几乎成了每个技术讨论的焦点。但说实话市面上关于Agent的讨论要么是停留在“它很智能、能自主完成任务”这种概念层面要么就是直接甩给你一个OpenAI的API调用示例告诉你用几个函数调用Function Calling就能实现一个简单的Agent。这中间存在一个巨大的断层从一个简单的API调用到一个真正健壮、可扩展、能处理复杂任务的AI Agent系统到底该怎么走这其中的理论支撑是什么工程上的坑又有哪些这就是我动手写这个系列的原因。我不满足于只是“使用”别人封装好的Agent框架我更想搞清楚它的“骨架”和“肌肉”是怎么长出来的。从零开始实现一个AI Agent框架不仅仅是为了炫技其核心价值在于深度理解与掌控。当你亲手设计它的思考循环ReAct, CoT实现工具调用Tool Calling的调度处理长期记忆Memory的存储与检索并搭建一个可靠的任务规划Planning与执行Execution引擎时你才会真正明白那些现成框架比如LangChain、AutoGen里每个设计抉择背后的深意。你会知道为什么有些任务它处理得好有些却会陷入死循环你会懂得如何为你的特定业务场景定制最合适的Agent架构而不是被框架的抽象所限制。这个项目适合谁我认为有三类朋友会特别有收获一是对AI应用开发有浓厚兴趣不满足于仅仅调用API的开发者二是希望将AI能力深度集成到自家产品中需要高度定制化Agent系统的技术负责人或架构师三是任何想要深入理解“智能体”究竟如何运作的研究者或学生。我们将从最基础的理论模型讲起然后用代码一步步把它们构建出来过程中遇到的每一个错误和解决方案我都会毫无保留地分享出来。2. 核心理论基石构建AI Agent的四大支柱在动手写代码之前我们必须把地基打牢。一个功能完整的AI Agent框架其理论核心可以归纳为四个相互关联的支柱规划Planning、记忆Memory、工具使用Tool Use和行动Action。理解这四者的关系是设计任何Agent系统的前提。2.1 规划模块从目标到步骤的拆解艺术规划本质上是一个问题分解与路径搜索的过程。对于AI Agent而言给定一个模糊的用户指令如“帮我分析一下上季度的销售数据并给出下季度的增长建议”它需要自动将其分解为一系列可执行的具体步骤。这里涉及几个关键理论链式思考Chain-of-Thought, CoT与ReAct范式这是当前让大语言模型LLM进行规划最主流的方法。CoT要求模型展示其推理的中间步骤。而ReActReasoning Acting则更进一步它将推理Reason和行动Act交织在一起。Agent不是一次性规划所有步骤而是采用“思考一步执行一步观察结果再思考下一步”的循环。这种方式更贴近人类解决问题的方式也更能应对执行过程中的不确定性。在我们的框架设计中ReAct循环将是核心执行引擎。任务分解Task Decomposition如何让LLM可靠地将大任务拆成小任务常见策略有1按子目标分解让LLM识别出达成总目标所需的中间子目标。2按工具或能力分解根据可用的工具集如“查询数据库”、“生成图表”、“发送邮件”来反向推导步骤。3按时间或流程顺序分解适用于有严格先后顺序的任务。在实现时我们通常通过精心设计的系统提示词System Prompt来引导LLM进行这种分解。规划算法对于复杂任务简单的逐步分解可能不够。我们可以引入更经典的算法思想如基于树的搜索广度优先、深度优先或基于图的规划。例如Agent可以将每个可能的行动步骤及其预期结果作为一个节点构建一个搜索空间然后寻找从初始状态到目标状态的最优路径。虽然这需要更多的计算和与环境的交互但对于需要多步深度规划的场景至关重要。注意规划模块最怕陷入“空转”。即Agent不停地进行推理却无法产生任何有效的行动步骤。为了防止这一点我们需要在提示词中明确约束如“必须从可用工具列表中选择行动”并在代码层面设置最大推理步数或超时机制。2.2 记忆模块赋予Agent持续性的关键一个没有记忆的Agent每次交互都是全新的开始这显然不是“智能体”该有的样子。记忆模块负责存储、组织和检索历史信息是Agent实现持续对话、学习用户偏好、积累领域知识的基础。我们可以从三个时间维度来设计记忆短期记忆/工作记忆这通常指当前会话的上下文。由于LLM有上下文窗口长度限制我们需要一个高效的上下文窗口管理策略。简单的做法是使用“滑动窗口”只保留最近的N轮对话。更高级的策略则涉及对历史对话进行摘要Summarization将冗长的历史压缩成精炼的要点再放入上下文从而在有限的窗口内承载更长时间跨度的信息。长期记忆这是Agent的“知识库”或“经验库”。所有超越当前会话的有用信息都应存储在这里。实现长期记忆的核心技术是向量数据库Vector Database。我们将对话、执行结果、学到的知识等文本转换成向量Embedding存入向量库。当需要回忆时将当前问题也转换成向量进行相似度搜索召回最相关的记忆片段再注入到当前上下文中。这解决了传统数据库基于关键词匹配的局限性实现了基于语义的关联回忆。记忆的层次与结构记忆不是扁平的。我们可以设计不同的“记忆体”例如核心事实记忆用户提供的明确信息、对话历史记忆、自我反思记忆Agent对自己过去行动成功或失败的分析、技能记忆如何成功使用某个工具的经验。在检索时可以根据当前任务类型决定优先从哪种记忆体中搜索。2.3 工具使用模块Agent与世界的接口Agent再聪明如果无法操作外部系统也只能是个“思想家”。工具使用模块是Agent能力的放大器。其核心理论是将外部API、函数、甚至其他软件的能力抽象成统一的、LLM可理解和调用的“工具”描述。工具的描述与发现每个工具都需要一个清晰的“说明书”通常包括工具名称、功能描述、必需的输入参数及其类型、格式、说明、可能的输出。这个说明书必须以结构化数据如JSON Schema的形式存在并能被动态地加载到Agent的上下文中。这就是所谓的工具发现Tool Discovery机制。工具的选择与调用当Agent决定要采取行动时它需要1从当前可用的工具列表中根据任务描述选择最合适的一个或多个工具。2根据工具说明书生成符合要求的调用参数。3以安全、可控的方式执行该工具对应的代码或API请求。这里的关键是参数验证与错误处理。LLM生成的参数可能是错误的、不完整的框架必须在调用前进行严格的校验并在调用失败时提供清晰的错误信息反馈给Agent以便其进行修正ReAct中的“观察”阶段。工具的扩展性框架必须支持轻松地注册新工具。理想情况下开发者只需要定义一个Python函数并用装饰器或配置文件描述其元信息框架就能自动将其纳入Agent的工具库。这涉及到灵活的插件化架构设计。2.4 行动与执行循环将计划付诸实践这是将前三个模块串联起来的“总控制器”通常体现为ReAct循环。其标准流程如下观察ObserveAgent接收当前的用户输入和来自环境的最新反馈如上一步工具执行的结果或错误。思考Think基于观察、历史记忆和可用工具LLM进行推理决定下一步该做什么。输出应被严格格式化例如必须明确指示是“继续思考”还是“调用某个工具”。行动Act如果决定调用工具则框架解析LLM的输出提取工具名和参数执行工具调用。循环将行动的结果作为新的“观察”进入下一轮循环直到LLM输出“最终答案”或达到终止条件如任务完成、步数超限。这个循环的实现质量直接决定了Agent的可靠性和效率。我们需要处理很多细节比如如何解析LLM非结构化的输出如何管理循环状态如何优雅地处理超时和中断3. 框架设计与核心模块实现有了理论指导我们就可以开始设计框架的顶层架构了。我们的目标是构建一个轻量级、模块化、易于扩展的框架我将其命名为“ThinkFlow”。整个框架的核心是一个中央调度器Orchestrator它协调着各个模块的运作。3.1 整体架构与模块划分ThinkFlow 的核心架构围绕一个主循环展开主要包括以下组件Agent Core封装了与大语言模型LLM的交互负责生成推理和决策。Planner规划模块负责将复杂任务分解为子任务序列。初期我们可以实现一个基于提示词的简单规划器。Memory Manager记忆管理器统一管理短期会话缓存和长期向量存储的读写。Tool Registry工具注册表一个全局的单例负责所有工具的注册、描述和查找。Executor执行器负责安全地调用Tool Registry中的工具并处理执行结果和异常。Orchestrator调度器实现上述的ReAct循环串联所有模块。它们之间的关系如下图所示概念图用户请求首先到达OrchestratorOrchestrator调用Planner进行任务分解对于复杂任务然后进入ReAct循环。在循环中Orchestrator将当前状态用户输入、历史、可用工具交给Agent Core生成思考根据思考决定是调用Executor执行工具还是直接给出答案。整个过程中Memory Manager负责提供相关记忆和保存新的经验。3.2 核心模块代码实现拆解接下来我们深入到几个最关键模块的代码层面。我会用Python来演示因为其生态丰富易于理解。首先是工具系统的实现。这是Agent能力的边界。我们设计一个工具装饰器让开发者能轻松地将任何函数变成Agent可用的工具。# tool.py import inspect import json from typing import Callable, Dict, Any, get_type_hints class Tool: 工具类封装一个可调用函数及其元数据 def __init__(self, func: Callable, name: str, description: str): self.func func self.name name self.description description self.parameters self._extract_parameters(func) def _extract_parameters(self, func: Callable) - Dict[str, Any]: 从函数签名中提取参数信息生成JSON Schema格式 sig inspect.signature(func) type_hints get_type_hints(func) parameters {} for param_name, param in sig.parameters.items(): if param_name self: continue param_info {type: string} # 默认类型 if param_name in type_hints: type_str str(type_hints[param_name]) # 简化类型映射实际项目需要更完善的映射 if int in type_str: param_info[type] integer elif float in type_str: param_info[type] number elif bool in type_str: param_info[type] boolean elif list in type_str or List in type_str: param_info[type] array if param.default ! inspect.Parameter.empty: param_info[default] param.default param_info[description] f参数 {param_name} # 可通过docstring获取更佳描述 parameters[param_name] param_info return parameters def to_schema(self) - Dict[str, Any]: 生成OpenAI Function Calling兼容的工具模式 return { type: function, function: { name: self.name, description: self.description, parameters: { type: object, properties: self.parameters, required: [k for k, v in self.parameters.items() if default not in v] } } } def execute(self, **kwargs): 执行工具调用 # 这里可以加入参数验证、权限检查、调用日志等 try: result self.func(**kwargs) return {status: success, data: result} except Exception as e: return {status: error, message: str(e)} class ToolRegistry: 全局工具注册表 _instance None _tools: Dict[str, Tool] {} def __new__(cls): if cls._instance is None: cls._instance super().__new__(cls) return cls._instance classmethod def register(cls, name: str None, description: str ): 装饰器用于注册工具函数 def decorator(func: Callable): tool_name name or func.__name__ tool Tool(func, tool_name, description) cls._tools[tool_name] tool return func return decorator classmethod def get_tool(cls, name: str) - Tool: return cls._tools.get(name) classmethod def get_all_schemas(cls): return [tool.to_schema() for tool in cls._tools.values()] # 使用示例 ToolRegistry.register(description计算两个数字的和) def add(a: int, b: int) - int: 返回a与b的和 return a b ToolRegistry.register(namesearch_web, description在互联网上搜索信息) def search(query: str) - str: # 模拟搜索实际应调用搜索引擎API return f关于{query}的搜索结果...这个工具系统实现了几个关键点1利用装饰器实现无侵入式的工具注册。2自动从函数签名和类型注解中提取参数信息生成标准化的模式描述。3提供了统一的执行和错误处理接口。其次是实现一个基于向量数据库的长期记忆模块。这里我们以ChromaDB为例它是一个轻量级的向量数据库。# memory.py from typing import List, Dict, Any import chromadb from chromadb.config import Settings from sentence_transformers import SentenceTransformer import uuid class LongTermMemory: 基于向量数据库的长期记忆 def __init__(self, persist_directory: str ./memory_db): # 初始化嵌入模型 self.embedder SentenceTransformer(all-MiniLM-L6-v2) # 轻量级句子嵌入模型 # 初始化Chroma客户端 self.client chromadb.PersistentClient( pathpersist_directory, settingsSettings(anonymized_telemetryFalse) ) # 获取或创建集合类似于表 self.collection self.client.get_or_create_collection(nameagent_memories) def _generate_embedding(self, text: str) - List[float]: 生成文本的向量嵌入 return self.embedder.encode(text).tolist() def store(self, content: str, metadata: Dict[str, Any] None): 存储一段记忆 if metadata is None: metadata {} embedding self._generate_embedding(content) id str(uuid.uuid4()) self.collection.add( embeddings[embedding], documents[content], metadatas[metadata], ids[id] ) return id def retrieve(self, query: str, n_results: int 5) - List[Dict[str, Any]]: 根据查询检索最相关的记忆 query_embedding self._generate_embedding(query) results self.collection.query( query_embeddings[query_embedding], n_resultsn_results ) memories [] # 解析返回结果 if results[documents]: for i in range(len(results[documents][0])): memories.append({ content: results[documents][0][i], metadata: results[metadatas][0][i], distance: results[distances][0][i] }) return memories def clear(self): 清空所有记忆谨慎使用 self.client.delete_collection(nameagent_memories) self.collection self.client.create_collection(nameagent_memories)这个记忆模块封装了向量的生成、存储和检索。在实际使用中我们可以在Agent完成一个重要任务后将任务总结和关键结果store起来。当遇到新任务时先用retrieve查找相关历史经验并将其作为上下文提供给LLM从而实现“经验复用”。4. 核心执行引擎ReAct循环的工程实现有了工具和记忆现在我们需要一个“大脑”来驱动一切。这个大脑就是ReAct循环执行引擎。它的职责是维持Agent的思考-行动周期并处理所有状态流转。4.1 状态管理与循环控制首先我们需要定义Agent在循环中的状态。一个典型的状态对象应包含# orchestrator.py from dataclasses import dataclass, field from typing import List, Dict, Any, Optional dataclass class AgentState: Agent在一次运行中的完整状态 user_input: str # 原始用户输入 task_description: str # 当前要处理的任务描述可能被分解过 full_history: List[Dict[str, str]] # 完整的对话历史格式[{role: user/assistant/system, content: ...}] recent_observations: List[str] # 最近几步的观察工具执行结果等 available_tools_schema: List[Dict] # 当前可用的工具模式列表 # 从长期记忆中检索到的相关上下文 retrieved_memories: List[str] field(default_factorylist) # 当前循环的步数 step_count: int 0 # 最终答案如果已得出 final_answer: Optional[str] None接下来我们实现Orchestrator的核心循环逻辑。这里的关键是设计一个强引导性的系统提示词让LLM严格按照我们期望的格式如JSON输出以便程序化解析。# orchestrator.py (续) import json import re class ReactOrchestrator: def __init__(self, llm_client, memory: LongTermMemory, max_steps20): self.llm llm_client self.memory memory self.max_steps max_steps self.tool_registry ToolRegistry() def _build_system_prompt(self, state: AgentState) - str: 构建引导ReAct循环的系统提示词 prompt f你是一个智能助手可以通过使用工具来解决问题。请遵循以下步骤 1. 分析当前任务和可用工具。 2. 如果需要使用工具请严格按照以下JSON格式回应 {{thought: 你的推理过程, action: tool_name, action_input: {{arg1: value1, ...}}}} 3. 如果任务已完成可以给出最终答案请使用格式 {{thought: 总结推理, action: final_answer, action_input: 你的最终答案}} 当前任务{state.task_description} 可用工具 {json.dumps(state.available_tools_schema, indent2, ensure_asciiFalse)} 历史观察最近几步 {chr(10).join(state.recent_observations[-3:]) if state.recent_observations else 无} 相关历史经验 {chr(10).join(state.retrieved_memories) if state.retrieved_memories else 无} 请开始你的思考。 return prompt def _parse_llm_output(self, text: str) - Dict[str, Any]: 解析LLM的输出提取结构化的动作指令 # 尝试从文本中提取JSON块 json_pattern rjson\s*(.*?)\s*|\s*(.*?)\s*|\{.*\} matches re.findall(json_pattern, text, re.DOTALL) json_str None for match in matches: json_str match[0] or match[1] or match[2] if json_str: break if not json_str: # 如果没有找到代码块尝试直接查找最外层的花括号 try: start text.find({) end text.rfind(}) 1 if start ! -1 and end ! 0: json_str text[start:end] except: json_str None if json_str: try: return json.loads(json_str) except json.JSONDecodeError: # 如果解析失败回退到启发式解析或报错 pass # 如果无法解析为JSON则假设模型直接给出了最终答案 return {thought: text, action: final_answer, action_input: text} def run(self, user_input: str) - str: 主运行循环 # 1. 初始化状态 state AgentState( user_inputuser_input, task_descriptionuser_input, full_history[{role: user, content: user_input}], recent_observations[], available_tools_schemaself.tool_registry.get_all_schemas(), retrieved_memories[] ) # 2. 从长期记忆中检索相关上下文 if self.memory: memories self.memory.retrieve(user_input, n_results3) state.retrieved_memories [m[content] for m in memories] # 3. ReAct 循环 for step in range(self.max_steps): state.step_count step 1 print(f\n 步骤 {step 1} ) # 3.1 构建当前轮次的对话上下文 messages [] # 添加系统提示词 system_prompt self._build_system_prompt(state) messages.append({role: system, content: system_prompt}) # 添加完整对话历史可选可只添加摘要 messages.extend(state.full_history[-6:]) # 保留最近几轮 # 3.2 调用LLM进行思考 llm_response self.llm.chat_completion(messages) assistant_reply llm_response[choices][0][message][content] state.full_history.append({role: assistant, content: assistant_reply}) print(f助理回复{assistant_reply}) # 3.3 解析LLM输出得到动作指令 action_dict self._parse_llm_output(assistant_reply) thought action_dict.get(thought, ) action action_dict.get(action, ) action_input action_dict.get(action_input, {}) print(f解析结果思考 - {thought} 动作 - {action}) # 3.4 执行动作 if action final_answer: state.final_answer str(action_input) # 可选将成功的任务经验存入长期记忆 if self.memory and state.final_answer: summary f任务{state.task_description}\n结果{state.final_answer[:200]}... self.memory.store(summary, metadata{type: successful_task}) break elif action in [tool.name for tool in self.tool_registry._tools.values()]: # 执行工具调用 tool self.tool_registry.get_tool(action) if tool: try: # 确保action_input是字典 if isinstance(action_input, str): try: action_input json.loads(action_input) except: action_input {input: action_input} result tool.execute(**action_input) observation f工具 {action} 执行结果{result} except Exception as e: observation f工具 {action} 执行出错{str(e)} else: observation f错误未知工具 {action} # 将观察结果加入状态 state.recent_observations.append(observation) state.full_history.append({role: user, content: observation}) print(f观察{observation}) else: # 无法识别的动作作为观察反馈给LLM observation f无法理解的动作指令{action}。请使用指定的JSON格式并选择可用的工具或给出最终答案。 state.recent_observations.append(observation) state.full_history.append({role: user, content: observation}) print(f观察{observation}) # 3.5 检查步数限制 if step self.max_steps - 1: state.final_answer f达到最大步数限制{self.max_steps}任务未完成。最后状态{state.recent_observations[-1] if state.recent_observations else 无进展} break return state.final_answer or 任务执行未产生明确结果。这个Orchestrator实现了一个完整的、可运行的ReAct循环。它包含了状态管理、提示词工程、LLM输出解析、工具执行调度以及循环终止判断。这是整个框架的“心脏”。4.2 与LLM的集成策略上面的代码中llm_client是一个抽象接口。在实际项目中我们需要适配不同的LLM提供商如OpenAI、Anthropic、国内大模型等。一个好的设计是定义一个统一的LLM客户端接口# llm_client.py from abc import ABC, abstractmethod import openai # 示例 class BaseLLMClient(ABC): abstractmethod def chat_completion(self, messages: List[Dict], **kwargs) - Dict[str, Any]: 发送聊天补全请求返回统一格式的响应 pass class OpenAIClient(BaseLLMClient): def __init__(self, model: str gpt-4, api_key: str None): self.model model self.client openai.OpenAI(api_keyapi_key) def chat_completion(self, messages: List[Dict], **kwargs) - Dict[str, Any]: response self.client.chat.completions.create( modelself.model, messagesmessages, **kwargs ) # 将响应格式化为统一的字典格式 return { id: response.id, choices: [{ message: { role: choice.message.role, content: choice.message.content } } for choice in response.choices] }这样Orchestrator就与具体的LLM实现解耦了我们可以轻松切换后端模型。5. 实战演练构建一个数据分析助手Agent理论和技术都讲完了现在我们用一个完整的例子把所有的模块串起来构建一个能实际运行的“数据分析助手”Agent。这个Agent的目标是用户用自然语言提出一个数据分析需求比如“帮我分析一下销售数据找出最畅销的产品”Agent能自动调用相应的工具读取文件、执行查询、绘制图表来完成分析并给出结论。5.1 定义领域专用工具首先我们需要为这个数据分析场景创建几个工具。假设我们有一个CSV格式的销售数据文件sales.csv。# data_analysis_tools.py import pandas as pd import matplotlib.pyplot as plt import io import base64 from ToolRegistry import ToolRegistry ToolRegistry.register(description读取指定路径的CSV文件返回数据预览) def read_csv(file_path: str, n_rows: int 5) - str: 读取CSV文件并返回前几行预览 try: df pd.read_csv(file_path) preview df.head(n_rows).to_string() shape df.shape return f文件 {file_path} 读取成功。数据形状{shape}。前{n_rows}行预览\n{preview} except Exception as e: return f读取文件失败{str(e)} ToolRegistry.register(description对数据进行SQL-like的查询例如按列分组求和) def query_data(file_path: str, group_by: str, aggregate: str sum) - str: 对数据进行简单的分组聚合查询 try: df pd.read_csv(file_path) if group_by not in df.columns: return f错误数据中不存在列 {group_by}。 if aggregate sum: result df.groupby(group_by).sum(numeric_onlyTrue) elif aggregate mean: result df.groupby(group_by).mean(numeric_onlyTrue) elif aggregate count: result df.groupby(group_by).size() else: return f不支持的聚合操作{aggregate}请使用 sum, mean 或 count。 return f按 {group_by} 分组执行 {aggregate} 聚合的结果\n{result.to_string()} except Exception as e: return f查询失败{str(e)} ToolRegistry.register(nameplot_chart, description根据数据生成折线图或柱状图返回图片的base64编码) def plot_chart(file_path: str, x_column: str, y_column: str, chart_type: str line) - str: 生成图表并返回base64字符串 try: df pd.read_csv(file_path) if x_column not in df.columns or y_column not in df.columns: return f错误指定的列不存在。 plt.figure(figsize(10, 6)) if chart_type line: plt.plot(df[x_column], df[y_column]) elif chart_type bar: plt.bar(df[x_column], df[y_column]) else: return f不支持的图表类型{chart_type}请使用 line 或 bar。 plt.xlabel(x_column) plt.ylabel(y_column) plt.title(f{y_column} vs {x_column}) plt.tight_layout() # 将图表保存到内存缓冲区并转换为base64 buf io.BytesIO() plt.savefig(buf, formatpng) plt.close() buf.seek(0) img_base64 base64.b64encode(buf.read()).decode(utf-8) return fdata:image/png;base64,{img_base64} except Exception as e: return f生成图表失败{str(e)} finally: plt.close(all)5.2 组装并运行Agent现在我们把所有部件组装起来并运行一个完整的任务。# main.py from llm_client import OpenAIClient from memory import LongTermMemory from orchestrator import ReactOrchestrator import data_analysis_tools # 导入工具模块以完成注册 def main(): # 1. 初始化组件 llm OpenAIClient(modelgpt-4) # 请替换为你的API Key memory LongTermMemory() orchestrator ReactOrchestrator(llm, memory, max_steps15) # 2. 用户输入一个复杂任务 user_query 请分析项目根目录下的 sales.csv 文件。 首先告诉我这个文件里有什么数据。 然后找出哪个产品product列的总销售额sales列最高。 最后请为我生成一个展示每个产品总销售额的柱状图。 print(用户请求, user_query) print(*50) # 3. 运行Agent final_answer orchestrator.run(user_query) # 4. 输出最终结果 print(\n *50) print(任务完成最终答案) print(final_answer) # 处理可能包含图片的答案 if final_answer and final_answer.startswith(data:image/png;base64,): # 如果是图片可以保存到本地 import base64 img_data final_answer.split(,)[1] with open(output_chart.png, wb) as f: f.write(base64.b64decode(img_data)) print(图表已保存为 output_chart.png) else: print(final_answer) if __name__ __main__: main()当你运行这段代码时你会看到控制台打印出Agent的思考过程步骤1LLM分析任务发现需要先读取文件。它输出JSON调用read_csv工具。步骤2Orchestrator执行工具得到数据预览并将其作为“观察”反馈给LLM。步骤3LLM根据预览知道下一步需要按产品分组求和。它调用query_data工具参数为group_byproduct, aggregatesum。步骤4得到查询结果后LLM决定生成图表。它调用plot_chart工具指定x_columnproduct, y_columnsales, chart_typebar。步骤5收到图表数据后LLM综合所有信息给出最终答案并可能附上分析结论。这个过程完美地展示了ReAct循环思考 - 行动 - 观察 - 再思考。我们的框架成功地协调了LLM的推理能力和外部工具的执行能力。6. 避坑指南与性能优化实战在从零搭建和实际使用这个框架的过程中我踩过不少坑也总结出一些让Agent更稳定、更高效的技巧。这部分是你在官方文档里很难看到的“实战经验”。6.1 提示词工程让LLM乖乖听话LLM不按格式输出是ReAct循环失败的主要原因。除了在代码中做健壮的解析提示词的设计至关重要。结构化输出强制在系统提示词中明确要求LLM以指定格式如JSON回复并给出多个正反示例。例如你必须以JSON格式回应。例如调用工具{thought: ..., action: tool_name, action_input: {...}}。给出最终答案{thought: ..., action: final_answer, action_input: ...}。任何其他格式都将导致错误。思维链CoT引导鼓励LLM在thought字段中详细推理。例如“在thought中请逐步解释你为什么选择这个工具以及你期望得到什么结果。” 这不仅能提高动作的准确性也便于我们调试。工具描述精细化给工具的description和参数描述写得越详细、越自然LLM调用得就越准。避免使用技术性过强的语言用LLM能理解的自然语言描述功能。例如description计算给定列表中所有数字的总和比descriptionsum function要好得多。6.2 错误处理与循环稳定性一个生产级的Agent框架必须能优雅地处理各种异常。工具执行异常工具可能因为网络、参数错误、权限等问题失败。我们的Tool.execute()方法已经返回了包含状态的字典。Orchestrator需要将这些错误信息清晰地反馈给LLM让它有机会调整策略。例如将错误信息格式化为“工具X执行失败原因为Y。请检查你的输入参数或尝试其他方法。”LLM输出解析失败即使用户提示词LLM也可能输出非结构化内容。我们的_parse_llm_output函数尝试了多种提取JSON的方法。如果全部失败最后的兜底策略是将整个输出当作final_answer或者将其作为错误观察反馈给LLM要求它重试。可以设置一个重试计数器避免无限循环。循环停滞检测Agent有时会陷入“死循环”比如反复调用同一个工具且得到相同结果。解决方法有1在状态中记录最近N次的动作和观察如果检测到重复模式则中断循环并提示用户。2设置最大步数这是必须的保险丝。6.3 记忆系统的优化技巧记忆系统用不好反而会成为负担。存储策略不要什么都存。只存储成功的任务总结、重要的用户信息如偏好以及从失败中学习的教训。存储时可以生成一个更概括的摘要而不是原始对话的罗列。检索优化简单基于向量的相似度搜索有时会召回不相关的记忆。可以尝试1元数据过滤在存储时为记忆打上标签如task_type: data_analysis检索时结合语义相似度和标签过滤。2重排序Rerank先用向量数据库召回较多结果如10个再用一个小型的、更精确的交叉编码器模型对它们进行重排序选出最相关的3-5个。上下文窗口管理对于长对话将完整的记忆全部放入上下文是不现实的。可以采用“摘要最近记录”的混合模式。定期例如每10轮对话让LLM对之前的对话历史生成一个简短摘要然后用这个摘要替代之前的大段历史腾出空间给新的对话。6.4 性能与成本考量频繁调用LLM和向量数据库成本和延迟是必须考虑的问题。LLM调用优化缓存对相同的提示词或提示词哈希的LLM响应进行缓存。这在开发调试和重复性问题中能节省大量成本。使用更小的模型对于简单的工具选择、参数提取等任务可以尝试使用更小、更快的模型如GPT-3.5-Turbo而只在需要复杂推理时使用大模型如GPT-4。并行工具调用如果多个工具调用之间没有依赖关系可以探索让LLM一次性输出多个工具调用请求然后并行执行减少循环次数。向量检索优化分块Chunking存储长文本记忆时将其分成有重叠的小块如256个token一块。这样检索时能更精准地定位到相关片段而不是返回一整篇不相关的文档。选择合适的索引ChromaDB、Pinecone等向量数据库都支持多种索引算法如HNSW。根据数据规模和查询延迟要求进行选择和调参。从零开始实现一个AI Agent框架就像亲手组装一台精密的钟表。你不仅知道了指针如何走动更理解了每一个齿轮的咬合与发条的张力。这个过程让我深刻体会到现成的框架固然方便但隐藏了太多关键的细节和设计权衡。当你自己实现一遍你会对“智能体”的脆弱性、对提示词的敏感性、对错误处理的必要性有全新的认识。这套代码只是一个起点你可以在此基础上继续探索多Agent协作、更复杂的规划算法如基于LLM的Tree of Thoughts、动态工具学习等高级特性。最重要的是你拥有了根据具体业务需求定制和优化每一个环节的能力这才是核心竞争力所在。