1. 从“单次问答”到“自主执行”为什么我们需要手写一个AI Agent如果你最近关注AI领域大概率已经被“AI Agent”这个词刷屏了。从OpenAI的GPTs到各种创业公司的产品似乎一夜之间所有AI应用都在朝着“智能体”的方向演进。但说实话很多所谓的Agent本质上还是一个“高级版聊天机器人”——你问一句它答一句顶多能调用几个预设好的工具。这离我们想象中的、能像人类一样理解复杂目标、拆解步骤、调用工具并最终完成任务的“智能代理”还有不小的距离。我自己在尝试将AI集成到工作流中时就遇到了这个瓶颈。比如我想让AI帮我分析一份市场报告然后根据分析结果生成一份PPT大纲最后再自动搜索一些相关的图片素材。这个过程涉及多个步骤、多种工具数据分析、文档生成、图像搜索和复杂的逻辑判断。传统的“Function Calling”模式虽然能让大模型调用外部函数但它本质上还是被动的、单次的。你需要为每一个微小的步骤手动发起请求并处理中间状态这既不智能也谈不上“自动化”。这正是“手写一个AI Agent”的核心驱动力。我们不再满足于让大模型扮演一个“超级API调用器”而是希望它成为一个拥有“大脑”的自主执行者。这个大脑能理解“分析报告并做PPT”这样的高层级目标能自己规划出“先提取关键数据 - 归纳成几个观点 - 为每个观点匹配幻灯片结构 - 根据关键词搜索图片”这样一条任务链并驱动整个流程自动执行直到最终目标达成。所以这篇内容我想和你分享的就是如何从最基础的Function Calling出发一步步构建出一个具备自动化任务链能力的AI Agent。这不是一个简单的框架调用教程而是一次从设计思想到代码实现的深度拆解。我们会探讨Agent的核心心智模型Planning, Memory, Tools并动手实现一个能处理多步骤、带状态、可回溯的任务执行引擎。无论你是想深入理解Agent的运作原理还是希望为自己的项目添加真正的自动化智能我相信接下来的内容都会对你有所启发。2. Function CallingAgent与外部世界交互的“手和脚”在构建Agent之前我们必须先解决一个基本问题AI如何与我们的代码、数据库、API乃至整个互联网进行交互答案就是Function Calling函数调用。这是目前大模型与外部工具集成最主流、最可靠的方式。你可以把它理解为Agent的“手和脚”——没有它Agent再聪明也只能困在文本的牢笼里空想。2.1 Function Calling的本质结构化指令的生成与解析很多人误以为Function Calling是大模型“直接执行”了我们的函数。其实不然。大模型如GPT-4本身并不能运行你的Python代码。Function Calling的流程是一个精巧的“请求-响应-执行”循环定义工具函数你告诉大模型你有哪些工具可用。这通过一个JSON Schema来完成描述函数的名称、功能说明以及参数。tools [ { type: function, function: { name: get_current_weather, description: 获取指定城市的当前天气, parameters: { type: object, properties: { location: { type: string, description: 城市名例如北京上海, }, unit: {type: string, enum: [celsius, fahrenheit]}, }, required: [location], }, }, } ]这里的description至关重要它是大模型决定是否、以及如何调用该工具的唯一依据。描述必须清晰、无歧义。模型决策与结构化输出你将用户的问题如“北京天气怎么样”和定义好的工具列表一起发送给大模型。模型会理解问题并判断是否需要、以及需要调用哪个工具。如果需要它不会直接执行而是返回一个结构化的JSON对象其中包含了它“想调用”的函数名和参数。{ role: assistant, content: null, tool_calls: [ { id: call_abc123, type: function, function: { name: get_current_weather, arguments: {\location\: \北京\, \unit\: \celsius\} } } ] }注意content为null而关键信息在tool_calls里。这标志着模型进入了“工具调用模式”。本地执行与结果返回你的程序接收到这个JSON后在本地找到对应的get_current_weather函数用解析出的参数location”北京”执行它并得到真实结果如{“temperature”: 22, “condition”: “晴朗”}。结果反馈与继续对话你将工具执行的结果以特定格式追加到对话历史中再次发送给大模型。{ role: tool, content: {\temperature\: 22, \condition\: \晴朗\}, tool_call_id: call_abc123 }模型接收到这个结果后会结合之前的对话上下文生成面向用户的最终回答比如“北京现在天气晴朗气温22摄氏度。”关键心得Function Calling的核心价值在于它将非结构化的自然语言指令转化为了结构化的、可编程的操作指令。这相当于给大模型装上了标准化的“插槽”任何能力都可以通过定义函数来接入。在设计和描述函数时一定要站在模型的角度思考怎样的描述能让它最准确地理解这个工具的用途和适用场景参数描述要具体避免使用“数据”、“信息”等泛泛之词。2.2 超越简单调用构建一个可靠的工具执行层单个函数调用很简单但当我们有几十个、上百个工具时管理就成了问题。一个健壮的Agent需要一个统一的工具执行层。这个层需要负责工具注册与管理提供一个中心化的地方来注册所有工具通常用一个字典或类来管理键为工具名值为函数对象和它的Schema。安全沙箱与验证对于执行外部命令、访问数据库或网络请求的工具必须有权限控制和输入验证防止Prompt注入导致恶意操作。规范化结果与错误处理工具执行可能成功也可能失败。执行层需要捕获异常并将结果统一格式化为模型能理解的字符串对于错误可以返回“工具XXX执行失败原因XXX”让模型有机会调整策略或告知用户。在我的实现中我通常会创建一个ToolRegistry类class ToolRegistry: def __init__(self): self._tools {} def register(self, func: Callable, schema: dict): 注册一个工具 self._tools[schema[function][name]] { function: func, schema: schema } def get_tool_schemas(self): 获取所有工具的JSON Schema用于发送给LLM return [tool[schema] for tool in self._tools.values()] def execute(self, tool_name: str, arguments: dict): 根据名称和参数执行工具 if tool_name not in self._tools: raise ValueError(f工具 {tool_name} 未注册。) tool self._tools[tool_name] try: # 这里可以加入参数验证、权限检查等 result tool[function](**arguments) return str(result) # 确保返回字符串 except Exception as e: return f执行工具 {tool_name} 时出错: {str(e)}这个简单的注册中心将工具的声明、管理和执行解耦是构建复杂Agent的基础设施。3. 任务规划与分解为Agent装上“思考”的大脑有了与世界交互的“手脚”Tools我们的Agent还缺一个“大脑”来指挥它们。这个大脑的核心能力就是任务规划与分解。这是区分一个“能调用工具的聊天机器人”和一个“真正Agent”的关键。当用户说“帮我分析上个月的销售数据并总结成一份报告发到我邮箱”时Agent需要自己思考“要完成这个目标我需要先后做哪些事”3.1 规划的本质将模糊目标转化为可执行指令序列规划能力对于当前的大模型来说本质上是一个复杂的推理和文本生成问题。我们通过Prompt工程来引导模型进行思考。一种经典且有效的方法是Chain of Thought (CoT) 及其变体。我们不会简单地问模型“该怎么做”而是要求它以一种结构化的格式输出思考过程。例如我们可以设计这样的Prompt你是一个AI助手需要完成用户给定的任务。你可以使用以下工具 - web_search(query): 使用搜索引擎搜索网络信息。 - read_file(file_path): 读取本地文件内容。 - data_analyzer(data, metrics): 对数据进行指定指标的分析。 - write_report(content, format): 根据内容撰写报告。 - send_email(to, subject, body): 发送电子邮件。 请为以下任务制定一个分步执行计划。你的回答必须严格遵循以下JSON格式 { thought: 你的逐步推理思考过程解释为什么需要这些步骤。, plan: [ {step: 1, action: 要执行的动作描述, tool: 使用的工具名, input: {参数1: 值1}}, {step: 2, action: ..., tool: ..., input: {...}}, ... ] } 任务分析上个月的销售数据并总结成一份报告发到我的邮箱。一个优秀的规划Prompt需要具备几个要素明确角色和工具清单让模型清楚自己的能力和可用资源。要求结构化输出强制模型以JSON等格式输出便于程序解析避免自由文本的随意性。引导深度思考通过要求输出“thought”字段促使模型进行逻辑推理而不仅仅是罗列步骤。这能提高计划的质量和可解释性。模型可能会返回如下计划{ thought: “用户需要分析销售数据并邮件报告。首先我需要获取数据假设数据在‘sales_last_month.csv’文件中。接着我需要分析关键指标如总销售额、环比增长率、最佳销售产品等。然后将分析结果组织成一份简洁的文本报告。最后调用邮件工具发送报告。”, plan: [ {step: 1, action: “读取上个月销售数据文件”, “tool”: “read_file”, “input”: {“file_path”: “sales_last_month.csv”}}, {step”: 2, “action”: “分析销售额和增长情况”, “tool”: “data_analyzer”, “input”: {“data”: “[上一步的结果]”, “metrics”: [“total_sales”, “growth_rate”, “top_product”]}}, {step”: 3, “action”: “撰写分析报告”, “tool”: “write_report”, “input”: {“content”: “[上一步的结果]”, “format”: “markdown”}}, {step”: 4, “action”: “发送报告到用户邮箱”, “tool”: “send_email”, “input”: {“to”: “userexample.com”, “subject”: “上月销售分析报告”, “body”: “[上一步的报告内容]”}} ] }踩坑实录在早期实现中我让模型直接输出步骤列表但经常遇到步骤逻辑混乱、工具选择错误的问题。后来强制加入“thought”推理字段后计划的合理性大幅提升。因为模型为了写出合理的“思考”必须更深入地理解任务和工具之间的关系。这相当于让模型自己给自己做了一次“审核”。3.2 动态规划与ReAct模式边执行边调整上述的“先规划后执行”模式是静态的。但在真实场景中计划赶不上变化。执行第一步“读取文件”时可能发现文件不存在执行“数据分析”时可能发现数据格式不符合预期。一个鲁棒的Agent必须具备动态重新规划的能力。这就是ReAct (Reason Act)模式大放异彩的地方。ReAct的核心思想是将推理Reason和行动Act交织在一个循环中思考Think根据当前目标、已知信息和历史观察思考下一步应该做什么。行动Act根据思考选择执行一个动作如调用工具。观察Observe获取动作执行的结果成功或失败。循环基于新的观察再次进入“思考”步骤直到任务完成或无法继续。如何用代码实现这个循环关键在于维护一个不断增长的“上下文”其中包含系统指令定义Agent的角色和核心目标。工具定义可用的工具列表。对话历史用户的问题、模型之前的“思考”、工具调用和工具返回的结果。每一次循环我们都将整个上下文发送给大模型并提示它根据最新情况尤其是上一步工具的“观察”结果来决定下一步是继续调用工具还是直接给出最终答案。一个简化的ReAct循环Prompt模板如下你是一个任务执行AI。你的目标{用户目标}。 你可以使用的工具{工具列表描述}。 历史记录 {将之前的思考、行动、观察按时间顺序拼接} 当前状态任务尚未完成。 请根据以上信息决定下一步做什么。你必须输出以下JSON格式 { “thought”: “对当前状况的分析和下一步的理由”, “action”: {“type”: “tool_call”, “tool”: “工具名”, “input”: {参数}} 或 {“type”: “final_answer”, “answer”: “给用户的最终回答”} }假设第一步读取文件失败模型在第二次循环中的“思考”可能是“上一步尝试读取‘sales_last_month.csv’文件失败返回‘文件不存在’。我需要先确认正确的文件路径或询问用户。我可以先使用‘list_files’工具查看当前目录有哪些文件。” 然后它可能会选择调用list_files工具。这种模式赋予了Agent强大的容错和适应能力让它能够处理不确定性是构建实用Agent的基石。4. 构建任务链引擎实现状态管理与自动化执行当我们把Function Calling作为“手脚”把规划与ReAct模式作为“大脑”后接下来就需要一个“神经系统”来协调它们让整个系统能够自动、连贯地运行。这就是任务链引擎。它负责管理整个执行流程的生命周期包括状态维护、步骤调度、异常处理和结果汇总。4.1 定义任务与步骤状态机的视角我们可以将一个复杂的Agent任务看作一个状态机。每个“步骤”是一个状态步骤的执行调用工具是状态间的转换而工具执行的结果观察则是转换的输入和触发条件。首先我们需要定义核心的数据结构from dataclasses import dataclass from typing import Any, Dict, List, Optional from enum import Enum class StepStatus(Enum): PENDING “pending” # 等待执行 RUNNING “running” # 执行中 SUCCESS “success” # 成功 FAILED “failed” # 失败 dataclass class Step: 代表任务链中的一个步骤 id: str description: str # 步骤描述 tool_name: Optional[str] None # 要调用的工具名 tool_input: Optional[Dict[str, Any]] None # 工具输入参数 status: StepStatus StepStatus.PENDING result: Optional[str] None # 工具执行结果或错误信息 depends_on: List[str] None # 依赖的步骤ID列表 def __post_init__(self): if self.depends_on is None: self.depends_on [] dataclass class Task: 代表一个完整的Agent任务 task_id: str objective: str # 任务目标 steps: List[Step] # 步骤列表 current_step_index: int 0 context: Dict[str, Any] None # 共享上下文用于存储中间结果 final_result: Optional[str] None def __post_init__(self): if self.context is None: self.context {}在这个设计中Task是顶级容器Step是执行单元。Step可以指定依赖关系depends_on这允许我们构建非线性的、有向无环图DAG结构的任务链而不仅仅是简单的列表。context字典是一个共享内存用于在不同步骤间传递数据例如步骤1读取的数据可以存入context[“raw_data”]供步骤2使用。4.2 实现执行引擎调度、执行与上下文传递引擎的核心是一个循环它不断地从任务中取出“就绪”的步骤即所有依赖步骤都已成功来执行。一个基础引擎的骨架如下class TaskChainEngine: def __init__(self, tool_registry: ToolRegistry, llm_client): self.tool_registry tool_registry self.llm_client llm_client # 用于动态规划/ReAct的LLM客户端 def run_task(self, task: Task) - Task: 执行一个任务 print(f“开始执行任务: {task.objective}”) while not self._is_task_finished(task): # 1. 获取下一个待执行的步骤或由LLM动态决定 next_step self._get_next_step(task) if not next_step: # 可能所有步骤都完成或进入死锁 print(“无法确定下一步任务可能阻塞。”) break # 2. 执行该步骤 print(f“执行步骤: {next_step.description}”) next_step.status StepStatus.RUNNING success, result self._execute_step(next_step, task.context) # 3. 更新步骤状态和上下文 if success: next_step.status StepStatus.SUCCESS next_step.result result # 将结果存入上下文键名可以由步骤ID或约定决定 task.context[f“step_{next_step.id}_result”] result print(f“步骤成功结果: {result[:100]}...”) else: next_step.status StepStatus.FAILED next_step.result result print(f“步骤失败: {result}”) # 这里可以引入错误处理策略如重试、跳过或调用LLM重新规划 # 4. (可选) 动态规划路径。如果步骤失败或需要复杂决策可以在此处调用LLM # 根据当前所有步骤的状态和上下文重新生成后续计划。 # if next_step.status StepStatus.FAILED: # new_plan self._replan_with_llm(task) # task.steps self._merge_plan(task, new_plan) # 任务结束后汇总结果 task.final_result self._summarize_results(task) return task def _execute_step(self, step: Step, context: Dict) - (bool, str): 执行单个步骤 if step.tool_name: # 执行工具调用 try: # 这里可以做一个简单的模板渲染将context中的值注入到tool_input中 # 例如tool_input 是 {“file”: “{data_file}”}context 是 {“data_file”: “sales.csv”} resolved_input self._resolve_input(step.tool_input, context) result self.tool_registry.execute(step.tool_name, resolved_input) return True, result except Exception as e: return False, f“工具执行异常: {str(e)}” else: # 可能是一些不需要工具的内部操作比如条件判断、数据合并等 # 这里可以实现自定义逻辑 return True, “Internal step completed.” def _resolve_input(self, tool_input: Dict, context: Dict) - Dict: 解析工具输入参数将形如‘{var_name}’的占位符替换为context中的实际值 # 这是一个简单的实现示例 import json input_str json.dumps(tool_input) for key, value in context.items(): placeholder f“{{{key}}}” if placeholder in input_str: input_str input_str.replace(placeholder, str(value)) return json.loads(input_str) def _get_next_step(self, task: Task) - Optional[Step]: 根据依赖关系获取下一个可执行的步骤简单FIFO调度 for step in task.steps: if step.status StepStatus.PENDING: # 检查所有依赖是否都已完成 deps_met all( any(s.id dep_id and s.status StepStatus.SUCCESS for s in task.steps) for dep_id in step.depends_on ) if deps_met: return step return None def _is_task_finished(self, task: Task) - bool: 判断任务是否完成所有步骤都成功或失败或有最终结果 if task.final_result is not None: return True # 简单逻辑所有步骤都不是PENDING或RUNNING状态 for step in task.steps: if step.status in [StepStatus.PENDING, StepStatus.RUNNING]: return False return True def _summarize_results(self, task: Task) - str: 汇总任务结果可以简单拼接最后一步的结果也可以用LLM总结 # 简单实现返回最后一个成功步骤的结果 for step in reversed(task.steps): if step.status StepStatus.SUCCESS: return step.result return “任务未产生有效结果。”这个引擎实现了最基本的功能依赖检查、顺序执行、上下文传递和简单的错误处理。它构成了自动化任务链的骨架。实操心得上下文context的管理是任务链引擎中最容易出乱子的地方。我建议采用命名规范比如使用step_[step_id]_result作为键来存储步骤原始结果。对于需要被后续步骤复用的结构化数据可以设计一个更精细的上下文管理器支持数据类型文本、JSON、二进制和版本避免数据被意外覆盖或格式错误导致下游步骤失败。5. 记忆与反思让Agent在长程任务中持续学习一个只会执行预设流程的Agent其智能是有限的。想象一下你让Agent“每天下午三点检查服务器日志如果有错误就发邮件通知我”。如果它每次检查都从零开始完全记不住昨天发现了什么错误、已经通知过谁那它的价值就大打折扣。因此记忆和反思能力是高级Agent的标配。5.1 短期记忆与长期记忆存储什么如何存储Agent的记忆通常分为两类短期记忆/对话记忆保存当前任务会话中的上下文包括用户消息、Agent的思考、工具调用和结果。这直接服务于当前的规划和执行循环我们之前维护的“对话历史”就是短期记忆。它的特点是容量小、关联性强通常以列表或滑动窗口的形式保存在内存中。长期记忆存储跨越多个任务会话的、需要持久化的信息。例如实体记忆用户的偏好“用户喜欢用Markdown格式看报告”、项目信息“项目A的API密钥是XXX”。过程记忆/经验过去执行类似任务的成功步骤、遇到的坑及其解决方案。事实知识从工具调用中提取并沉淀下来的结构化信息如“公司上季度销售额为1000万”。实现长期记忆最简单的方式是使用一个键值数据库如Redis或文档数据库如MongoDB。更复杂的系统会使用向量数据库如ChromaDB, Pinecone来存储记忆的嵌入向量从而实现基于语义相似度的记忆检索。一个结合了向量检索的长期记忆模块示例import chromadb from chromadb.utils import embedding_functions class LongTermMemory: def __init__(self, persist_path“./chroma_db”): self.client chromadb.PersistentClient(pathpersist_path) # 使用一个通用的嵌入模型如all-MiniLM-L6-v2 self.embedding_fn embedding_functions.SentenceTransformerEmbeddingFunction(model_name“all-MiniLM-L6-v2”) self.collection self.client.get_or_create_collection( name“agent_memories”, embedding_functionself.embedding_fn ) def store(self, content: str, metadata: dict None): 存储一段记忆 doc_id f“mem_{int(time.time())}_{hash(content)}” self.collection.add( documents[content], metadatas[metadata] if metadata else [{}], ids[doc_id] ) return doc_id def retrieve(self, query: str, n_results: int 3) - List[dict]: 根据查询检索相关记忆 results self.collection.query( query_texts[query], n_resultsn_results ) memories [] if results[‘documents’]: for doc, meta in zip(results[‘documents’][0], results[‘metadatas’][0]): memories.append({“content”: doc, “metadata”: meta}) return memories def retrieve_relevant_to_task(self, task_objective: str, context: str) - List[str]: 检索与当前任务相关的记忆 # 可以将任务目标和当前上下文拼接起来作为查询提高相关性 query f“Task: {task_objective}. Context: {context}” memories self.retrieve(query) return [m[“content”] for m in memories]在任务开始或执行到某个决策点时我们可以调用retrieve_relevant_to_task方法将相关的历史经验注入到给LLM的Prompt中例如“根据过去经验处理类似‘销售数据分析’任务时先检查数据完整性再进行聚合计算成功率更高。” 这能显著提升Agent的决策质量。5.2 反思从经验中提炼“元知识”记忆的更高阶应用是反思。反思是指Agent在任务结束后或关键节点主动回顾执行过程总结经验教训并将其转化为可复用的知识存入长期记忆。我们可以设计一个“反思”步骤在任务结束时自动触发或者当任务失败时触发。这个步骤本身也是一个LLM调用请你作为AI助手对刚刚完成的任务进行反思和总结。 任务目标{task_objective} 执行步骤与结果 {将整个任务链的步骤状态和结果以文本形式列出} 请从以下角度进行总结 1. 任务是否成功完成如果没有根本原因是什么 2. 哪些步骤是关键的或效率较低的 3. 遇到了哪些意外或错误是如何解决的 4. 从本次任务中可以提炼出哪些对未来执行类似任务有帮助的经验或最佳实践 请将你的反思输出为结构化的JSON格式包含‘success’, ‘root_cause’, ‘key_insights’, ‘best_practices’等字段。将LLM生成的反思结果特别是best_practices存储到长期记忆中。当下次遇到类似任务时这些“元知识”就能被检索出来指导新的规划形成“执行 - 反思 - 学习 - 更好执行”的闭环。深度思考记忆和反思机制是Agent实现“个性化”和“持续进化”的关键。一个没有记忆的Agent每次交互都是独立的无法建立与用户或环境的深度联系。而一个具备反思能力的Agent则能从错误中学习不断优化自己的行为模式。在设计时需要考虑记忆的隐私、安全性和相关性过滤避免注入无关或过时的信息干扰当前任务。6. 实战构建一个自动化数据分析与报告Agent理论说了这么多我们来动手实现一个具体的、有实用价值的Agent一个自动化数据分析与报告生成Agent。它的目标是用户给定一个数据文件如CSV和一个分析需求Agent能自动完成数据读取、清洗、分析、可视化并生成一份图文并茂的分析报告。6.1 定义工具集Agent的能力边界首先我们需要为这个Agent装备一套数据分析相关的“工具”。read_csv_file(file_path): 读取CSV文件返回DataFrame的JSON字符串或摘要。get_data_summary(data): 获取数据的基本概览行数、列数、数据类型、缺失值。clean_data(data, instructions): 根据指令清洗数据处理缺失值、去重、类型转换。指令可以是自然语言由LLM解析。perform_analysis(data, analysis_type, columns): 执行特定分析如描述性统计、相关性分析、分组聚合等。create_visualization(data, viz_type, x, y, title): 创建图表折线图、柱状图、散点图等并保存为图片文件返回文件路径。write_markdown_report(sections): 将分析结果、图表路径和文本描述整合成Markdown报告。save_report(report_content, output_path): 将报告保存到指定路径。每个工具都需要精心设计其description和parameters的Schema确保LLM能准确理解和使用。例如create_visualization的Schema可能如下{ “name”: “create_visualization”, “description”: “根据提供的数据和参数创建图表。数据应是一个列名和值的字典列表。支持的类型有line折线图bar柱状图scatter散点图。图表将保存为PNG文件。”, “parameters”: { “type”: “object”, “properties”: { “data”: {“type”: “array”, “description”: “要绘制的数据格式为字典列表。”}, “viz_type”: {“type”: “string”, “enum”: [“line”, “bar”, “scatter”]}, “x_column”: {“type”: “string”, “description”: “用作X轴的列名。”}, “y_column”: {“type”: “string”, “description”: “用作Y轴的列名。”}, “title”: {“type”: “string”, “description”: “图表的标题。”} }, “required”: [“data”, “viz_type”, “x_column”, “y_column”] } }6.2 设计任务规划Prompt引导Agent思考分析流程对于数据分析任务我们可以设计一个更具引导性的规划Prompt将通用数据分析流程融入其中你是一个数据分析专家AI。用户将提供一个数据文件和分析目标。 你的任务是制定一个详细的数据分析流水线。 请遵循标准数据分析流程1. 数据获取与概览 2. 数据清洗 3. 探索性分析 4. 深入分析/可视化 5. 报告撰写。 可用的工具 {tool_schemas_json} 请为以下任务生成执行计划。你的回答必须是严格的JSON格式 { “thought”: “你的推理过程解释每一步的必要性。”, “plan”: [ {“step”: 1, “action”: “...”, “tool”: “...”, “input”: {...}}, ... ] } 任务文件路径为“{file_path}”分析目标是“{analysis_goal}”。这个Prompt通过明确“标准数据分析流程”极大地约束和引导了LLM的规划方向使其生成的计划更专业、更可靠。6.3 组装与运行见证自动化链的威力现在我们将所有组件组装起来初始化创建ToolRegistry注册所有数据分析工具。创建TaskChainEngine和LongTermMemory可选。接收任务用户输入file_path“sales_data.csv”和analysis_goal“分析各产品线的月度销售趋势和贡献率”。生成计划将任务目标和工具Schema发送给LLM获得初始执行计划。创建任务对象将LLM返回的plan转化为Task和Step对象。执行引擎启动TaskChainEngine.run_task(task)。动态调整可选在引擎循环中如果某一步骤失败如数据清洗指令不明确可以触发一个“重新规划”的子流程让LLM根据当前错误和上下文调整后续步骤。输出结果任务完成后从task.final_result中获取生成的Markdown报告路径。整个过程中用户只需提供文件和目标剩下的数据读取、质量检查、趋势计算、图表生成、报告整合全部由Agent自动完成。你可能会看到控制台输出类似这样的日志开始执行任务分析各产品线的月度销售趋势和贡献率。 执行步骤读取销售数据文件‘sales_data.csv’... 步骤成功结果数据已加载共1000行5列。 执行步骤检查数据概览和缺失值... 步骤成功结果数据完整无缺失值。各列类型正确。 执行步骤按月份和产品线分组计算销售额... 步骤成功结果分组聚合完成。 执行步骤创建月度销售趋势折线图... 步骤成功结果图表已保存至‘./viz/monthly_trend.png’。 执行步骤创建产品线销售额占比饼图... 步骤成功结果图表已保存至‘./viz/product_pie.png’。 执行步骤撰写分析报告... 步骤成功结果报告已生成包含趋势分析、贡献率分析和图表引用。 任务完成。最终报告位于./reports/sales_analysis_20231027.md避坑指南在实现这类数据密集型Agent时最大的挑战是数据在步骤间的流转。LLM和工具处理的数据格式可能不同如DataFrame vs JSON字符串。我强烈建议在context中定义清晰的数据接口契约。例如规定read_csv_file工具的输出是一个包含df_jsonDataFrame的JSON字符串和df_summary文本摘要的字典。后续工具都约定从context[“df_json”]中获取原始数据。同时要善用Python的ast.literal_eval或json.loads进行安全的格式转换避免eval带来的安全风险。7. 进阶思考Agent架构的挑战与未来方向构建一个可用的Agent原型并不难但要让它真正稳定、可靠、智能地运行在生产环境中我们还需要面对诸多挑战。7.1 可靠性挑战错误处理、超时与回滚工具执行的不可靠性网络API可能超时数据库可能连接失败文件可能被占用。引擎必须为每个工具调用设置超时和重试机制。对于非幂等操作如发送邮件、写入数据库重试需要格外小心。LLM响应的不确定性LLM可能不按要求的格式输出可能生成不合法的工具参数。引擎需要解析失败后的备选方案比如尝试用更严格的Prompt重试或者降级到让用户确认。任务链的回滚如果一个多步骤任务在后期失败可能需要回滚之前步骤产生的副作用如删除已创建的文件、撤销数据库操作。实现完善的Saga模式事务回滚对Agent来说非常复杂通常需要根据业务场景设计补偿操作。7.2 效率与成本优化规划与执行的平衡过于详细的规划每一步都让LLM决定会导致延迟高、成本高。过于僵化的流程完全预设又失去了灵活性。一个混合策略是对于通用、稳定的子流程如“数据清洗流程”可以封装成一个更大的“复合工具”让LLM直接调用这个复合工具内部则用固定代码执行减少LLM调用次数。上下文长度管理随着任务进行对话历史包含所有思考、行动、观察会越来越长可能超出模型的上下文窗口。需要设计摘要策略将过长的历史压缩成精炼的要点只保留对当前决策最关键的信息。缓存与记忆复用相同的工具调用如查询某城市天气结果在一段时间内可以缓存。长期记忆中的经验也可以避免重复的试错。7.3 评估与监控如何知道你的Agent做得好不好定义评估指标对于自动化任务最终结果的成功率Task Success Rate是核心指标。此外还可以衡量平均完成步骤数效率、工具调用准确率、用户满意度等。建立评估数据集构建一组涵盖不同难度和场景的测试任务定期运行Agent进行评估跟踪指标变化。可观测性记录Agent完整的执行轨迹Thought, Action, Observation这对于调试和优化至关重要。可以像日志系统一样将这些轨迹收集起来用于分析故障点和性能瓶颈。手写一个AI Agent的过程是一个不断在“赋予智能”和“施加约束”之间寻找平衡的过程。我们从让大模型调用单个函数开始逐步为它添加规划、记忆、执行引擎等组件本质上是在用软件工程的方法为概率模型构建一个确定性的、可靠的“外壳”。这个外壳决定了Agent的能力下限而内部LLM的智能则决定了其上限。未来的方向可能会朝着更模块化、更可解释的Agent框架发展也会出现更多专注于规划、工具使用或记忆的专项模型。但无论如何理解从Function Calling到自动化任务链这一完整的技术栈都将是你构建下一代AI应用不可或缺的核心能力。 从Function Calling到自动化任务链：手把手构建AI Agent核心架构
1. 从“单次问答”到“自主执行”为什么我们需要手写一个AI Agent如果你最近关注AI领域大概率已经被“AI Agent”这个词刷屏了。从OpenAI的GPTs到各种创业公司的产品似乎一夜之间所有AI应用都在朝着“智能体”的方向演进。但说实话很多所谓的Agent本质上还是一个“高级版聊天机器人”——你问一句它答一句顶多能调用几个预设好的工具。这离我们想象中的、能像人类一样理解复杂目标、拆解步骤、调用工具并最终完成任务的“智能代理”还有不小的距离。我自己在尝试将AI集成到工作流中时就遇到了这个瓶颈。比如我想让AI帮我分析一份市场报告然后根据分析结果生成一份PPT大纲最后再自动搜索一些相关的图片素材。这个过程涉及多个步骤、多种工具数据分析、文档生成、图像搜索和复杂的逻辑判断。传统的“Function Calling”模式虽然能让大模型调用外部函数但它本质上还是被动的、单次的。你需要为每一个微小的步骤手动发起请求并处理中间状态这既不智能也谈不上“自动化”。这正是“手写一个AI Agent”的核心驱动力。我们不再满足于让大模型扮演一个“超级API调用器”而是希望它成为一个拥有“大脑”的自主执行者。这个大脑能理解“分析报告并做PPT”这样的高层级目标能自己规划出“先提取关键数据 - 归纳成几个观点 - 为每个观点匹配幻灯片结构 - 根据关键词搜索图片”这样一条任务链并驱动整个流程自动执行直到最终目标达成。所以这篇内容我想和你分享的就是如何从最基础的Function Calling出发一步步构建出一个具备自动化任务链能力的AI Agent。这不是一个简单的框架调用教程而是一次从设计思想到代码实现的深度拆解。我们会探讨Agent的核心心智模型Planning, Memory, Tools并动手实现一个能处理多步骤、带状态、可回溯的任务执行引擎。无论你是想深入理解Agent的运作原理还是希望为自己的项目添加真正的自动化智能我相信接下来的内容都会对你有所启发。2. Function CallingAgent与外部世界交互的“手和脚”在构建Agent之前我们必须先解决一个基本问题AI如何与我们的代码、数据库、API乃至整个互联网进行交互答案就是Function Calling函数调用。这是目前大模型与外部工具集成最主流、最可靠的方式。你可以把它理解为Agent的“手和脚”——没有它Agent再聪明也只能困在文本的牢笼里空想。2.1 Function Calling的本质结构化指令的生成与解析很多人误以为Function Calling是大模型“直接执行”了我们的函数。其实不然。大模型如GPT-4本身并不能运行你的Python代码。Function Calling的流程是一个精巧的“请求-响应-执行”循环定义工具函数你告诉大模型你有哪些工具可用。这通过一个JSON Schema来完成描述函数的名称、功能说明以及参数。tools [ { type: function, function: { name: get_current_weather, description: 获取指定城市的当前天气, parameters: { type: object, properties: { location: { type: string, description: 城市名例如北京上海, }, unit: {type: string, enum: [celsius, fahrenheit]}, }, required: [location], }, }, } ]这里的description至关重要它是大模型决定是否、以及如何调用该工具的唯一依据。描述必须清晰、无歧义。模型决策与结构化输出你将用户的问题如“北京天气怎么样”和定义好的工具列表一起发送给大模型。模型会理解问题并判断是否需要、以及需要调用哪个工具。如果需要它不会直接执行而是返回一个结构化的JSON对象其中包含了它“想调用”的函数名和参数。{ role: assistant, content: null, tool_calls: [ { id: call_abc123, type: function, function: { name: get_current_weather, arguments: {\location\: \北京\, \unit\: \celsius\} } } ] }注意content为null而关键信息在tool_calls里。这标志着模型进入了“工具调用模式”。本地执行与结果返回你的程序接收到这个JSON后在本地找到对应的get_current_weather函数用解析出的参数location”北京”执行它并得到真实结果如{“temperature”: 22, “condition”: “晴朗”}。结果反馈与继续对话你将工具执行的结果以特定格式追加到对话历史中再次发送给大模型。{ role: tool, content: {\temperature\: 22, \condition\: \晴朗\}, tool_call_id: call_abc123 }模型接收到这个结果后会结合之前的对话上下文生成面向用户的最终回答比如“北京现在天气晴朗气温22摄氏度。”关键心得Function Calling的核心价值在于它将非结构化的自然语言指令转化为了结构化的、可编程的操作指令。这相当于给大模型装上了标准化的“插槽”任何能力都可以通过定义函数来接入。在设计和描述函数时一定要站在模型的角度思考怎样的描述能让它最准确地理解这个工具的用途和适用场景参数描述要具体避免使用“数据”、“信息”等泛泛之词。2.2 超越简单调用构建一个可靠的工具执行层单个函数调用很简单但当我们有几十个、上百个工具时管理就成了问题。一个健壮的Agent需要一个统一的工具执行层。这个层需要负责工具注册与管理提供一个中心化的地方来注册所有工具通常用一个字典或类来管理键为工具名值为函数对象和它的Schema。安全沙箱与验证对于执行外部命令、访问数据库或网络请求的工具必须有权限控制和输入验证防止Prompt注入导致恶意操作。规范化结果与错误处理工具执行可能成功也可能失败。执行层需要捕获异常并将结果统一格式化为模型能理解的字符串对于错误可以返回“工具XXX执行失败原因XXX”让模型有机会调整策略或告知用户。在我的实现中我通常会创建一个ToolRegistry类class ToolRegistry: def __init__(self): self._tools {} def register(self, func: Callable, schema: dict): 注册一个工具 self._tools[schema[function][name]] { function: func, schema: schema } def get_tool_schemas(self): 获取所有工具的JSON Schema用于发送给LLM return [tool[schema] for tool in self._tools.values()] def execute(self, tool_name: str, arguments: dict): 根据名称和参数执行工具 if tool_name not in self._tools: raise ValueError(f工具 {tool_name} 未注册。) tool self._tools[tool_name] try: # 这里可以加入参数验证、权限检查等 result tool[function](**arguments) return str(result) # 确保返回字符串 except Exception as e: return f执行工具 {tool_name} 时出错: {str(e)}这个简单的注册中心将工具的声明、管理和执行解耦是构建复杂Agent的基础设施。3. 任务规划与分解为Agent装上“思考”的大脑有了与世界交互的“手脚”Tools我们的Agent还缺一个“大脑”来指挥它们。这个大脑的核心能力就是任务规划与分解。这是区分一个“能调用工具的聊天机器人”和一个“真正Agent”的关键。当用户说“帮我分析上个月的销售数据并总结成一份报告发到我邮箱”时Agent需要自己思考“要完成这个目标我需要先后做哪些事”3.1 规划的本质将模糊目标转化为可执行指令序列规划能力对于当前的大模型来说本质上是一个复杂的推理和文本生成问题。我们通过Prompt工程来引导模型进行思考。一种经典且有效的方法是Chain of Thought (CoT) 及其变体。我们不会简单地问模型“该怎么做”而是要求它以一种结构化的格式输出思考过程。例如我们可以设计这样的Prompt你是一个AI助手需要完成用户给定的任务。你可以使用以下工具 - web_search(query): 使用搜索引擎搜索网络信息。 - read_file(file_path): 读取本地文件内容。 - data_analyzer(data, metrics): 对数据进行指定指标的分析。 - write_report(content, format): 根据内容撰写报告。 - send_email(to, subject, body): 发送电子邮件。 请为以下任务制定一个分步执行计划。你的回答必须严格遵循以下JSON格式 { thought: 你的逐步推理思考过程解释为什么需要这些步骤。, plan: [ {step: 1, action: 要执行的动作描述, tool: 使用的工具名, input: {参数1: 值1}}, {step: 2, action: ..., tool: ..., input: {...}}, ... ] } 任务分析上个月的销售数据并总结成一份报告发到我的邮箱。一个优秀的规划Prompt需要具备几个要素明确角色和工具清单让模型清楚自己的能力和可用资源。要求结构化输出强制模型以JSON等格式输出便于程序解析避免自由文本的随意性。引导深度思考通过要求输出“thought”字段促使模型进行逻辑推理而不仅仅是罗列步骤。这能提高计划的质量和可解释性。模型可能会返回如下计划{ thought: “用户需要分析销售数据并邮件报告。首先我需要获取数据假设数据在‘sales_last_month.csv’文件中。接着我需要分析关键指标如总销售额、环比增长率、最佳销售产品等。然后将分析结果组织成一份简洁的文本报告。最后调用邮件工具发送报告。”, plan: [ {step: 1, action: “读取上个月销售数据文件”, “tool”: “read_file”, “input”: {“file_path”: “sales_last_month.csv”}}, {step”: 2, “action”: “分析销售额和增长情况”, “tool”: “data_analyzer”, “input”: {“data”: “[上一步的结果]”, “metrics”: [“total_sales”, “growth_rate”, “top_product”]}}, {step”: 3, “action”: “撰写分析报告”, “tool”: “write_report”, “input”: {“content”: “[上一步的结果]”, “format”: “markdown”}}, {step”: 4, “action”: “发送报告到用户邮箱”, “tool”: “send_email”, “input”: {“to”: “userexample.com”, “subject”: “上月销售分析报告”, “body”: “[上一步的报告内容]”}} ] }踩坑实录在早期实现中我让模型直接输出步骤列表但经常遇到步骤逻辑混乱、工具选择错误的问题。后来强制加入“thought”推理字段后计划的合理性大幅提升。因为模型为了写出合理的“思考”必须更深入地理解任务和工具之间的关系。这相当于让模型自己给自己做了一次“审核”。3.2 动态规划与ReAct模式边执行边调整上述的“先规划后执行”模式是静态的。但在真实场景中计划赶不上变化。执行第一步“读取文件”时可能发现文件不存在执行“数据分析”时可能发现数据格式不符合预期。一个鲁棒的Agent必须具备动态重新规划的能力。这就是ReAct (Reason Act)模式大放异彩的地方。ReAct的核心思想是将推理Reason和行动Act交织在一个循环中思考Think根据当前目标、已知信息和历史观察思考下一步应该做什么。行动Act根据思考选择执行一个动作如调用工具。观察Observe获取动作执行的结果成功或失败。循环基于新的观察再次进入“思考”步骤直到任务完成或无法继续。如何用代码实现这个循环关键在于维护一个不断增长的“上下文”其中包含系统指令定义Agent的角色和核心目标。工具定义可用的工具列表。对话历史用户的问题、模型之前的“思考”、工具调用和工具返回的结果。每一次循环我们都将整个上下文发送给大模型并提示它根据最新情况尤其是上一步工具的“观察”结果来决定下一步是继续调用工具还是直接给出最终答案。一个简化的ReAct循环Prompt模板如下你是一个任务执行AI。你的目标{用户目标}。 你可以使用的工具{工具列表描述}。 历史记录 {将之前的思考、行动、观察按时间顺序拼接} 当前状态任务尚未完成。 请根据以上信息决定下一步做什么。你必须输出以下JSON格式 { “thought”: “对当前状况的分析和下一步的理由”, “action”: {“type”: “tool_call”, “tool”: “工具名”, “input”: {参数}} 或 {“type”: “final_answer”, “answer”: “给用户的最终回答”} }假设第一步读取文件失败模型在第二次循环中的“思考”可能是“上一步尝试读取‘sales_last_month.csv’文件失败返回‘文件不存在’。我需要先确认正确的文件路径或询问用户。我可以先使用‘list_files’工具查看当前目录有哪些文件。” 然后它可能会选择调用list_files工具。这种模式赋予了Agent强大的容错和适应能力让它能够处理不确定性是构建实用Agent的基石。4. 构建任务链引擎实现状态管理与自动化执行当我们把Function Calling作为“手脚”把规划与ReAct模式作为“大脑”后接下来就需要一个“神经系统”来协调它们让整个系统能够自动、连贯地运行。这就是任务链引擎。它负责管理整个执行流程的生命周期包括状态维护、步骤调度、异常处理和结果汇总。4.1 定义任务与步骤状态机的视角我们可以将一个复杂的Agent任务看作一个状态机。每个“步骤”是一个状态步骤的执行调用工具是状态间的转换而工具执行的结果观察则是转换的输入和触发条件。首先我们需要定义核心的数据结构from dataclasses import dataclass from typing import Any, Dict, List, Optional from enum import Enum class StepStatus(Enum): PENDING “pending” # 等待执行 RUNNING “running” # 执行中 SUCCESS “success” # 成功 FAILED “failed” # 失败 dataclass class Step: 代表任务链中的一个步骤 id: str description: str # 步骤描述 tool_name: Optional[str] None # 要调用的工具名 tool_input: Optional[Dict[str, Any]] None # 工具输入参数 status: StepStatus StepStatus.PENDING result: Optional[str] None # 工具执行结果或错误信息 depends_on: List[str] None # 依赖的步骤ID列表 def __post_init__(self): if self.depends_on is None: self.depends_on [] dataclass class Task: 代表一个完整的Agent任务 task_id: str objective: str # 任务目标 steps: List[Step] # 步骤列表 current_step_index: int 0 context: Dict[str, Any] None # 共享上下文用于存储中间结果 final_result: Optional[str] None def __post_init__(self): if self.context is None: self.context {}在这个设计中Task是顶级容器Step是执行单元。Step可以指定依赖关系depends_on这允许我们构建非线性的、有向无环图DAG结构的任务链而不仅仅是简单的列表。context字典是一个共享内存用于在不同步骤间传递数据例如步骤1读取的数据可以存入context[“raw_data”]供步骤2使用。4.2 实现执行引擎调度、执行与上下文传递引擎的核心是一个循环它不断地从任务中取出“就绪”的步骤即所有依赖步骤都已成功来执行。一个基础引擎的骨架如下class TaskChainEngine: def __init__(self, tool_registry: ToolRegistry, llm_client): self.tool_registry tool_registry self.llm_client llm_client # 用于动态规划/ReAct的LLM客户端 def run_task(self, task: Task) - Task: 执行一个任务 print(f“开始执行任务: {task.objective}”) while not self._is_task_finished(task): # 1. 获取下一个待执行的步骤或由LLM动态决定 next_step self._get_next_step(task) if not next_step: # 可能所有步骤都完成或进入死锁 print(“无法确定下一步任务可能阻塞。”) break # 2. 执行该步骤 print(f“执行步骤: {next_step.description}”) next_step.status StepStatus.RUNNING success, result self._execute_step(next_step, task.context) # 3. 更新步骤状态和上下文 if success: next_step.status StepStatus.SUCCESS next_step.result result # 将结果存入上下文键名可以由步骤ID或约定决定 task.context[f“step_{next_step.id}_result”] result print(f“步骤成功结果: {result[:100]}...”) else: next_step.status StepStatus.FAILED next_step.result result print(f“步骤失败: {result}”) # 这里可以引入错误处理策略如重试、跳过或调用LLM重新规划 # 4. (可选) 动态规划路径。如果步骤失败或需要复杂决策可以在此处调用LLM # 根据当前所有步骤的状态和上下文重新生成后续计划。 # if next_step.status StepStatus.FAILED: # new_plan self._replan_with_llm(task) # task.steps self._merge_plan(task, new_plan) # 任务结束后汇总结果 task.final_result self._summarize_results(task) return task def _execute_step(self, step: Step, context: Dict) - (bool, str): 执行单个步骤 if step.tool_name: # 执行工具调用 try: # 这里可以做一个简单的模板渲染将context中的值注入到tool_input中 # 例如tool_input 是 {“file”: “{data_file}”}context 是 {“data_file”: “sales.csv”} resolved_input self._resolve_input(step.tool_input, context) result self.tool_registry.execute(step.tool_name, resolved_input) return True, result except Exception as e: return False, f“工具执行异常: {str(e)}” else: # 可能是一些不需要工具的内部操作比如条件判断、数据合并等 # 这里可以实现自定义逻辑 return True, “Internal step completed.” def _resolve_input(self, tool_input: Dict, context: Dict) - Dict: 解析工具输入参数将形如‘{var_name}’的占位符替换为context中的实际值 # 这是一个简单的实现示例 import json input_str json.dumps(tool_input) for key, value in context.items(): placeholder f“{{{key}}}” if placeholder in input_str: input_str input_str.replace(placeholder, str(value)) return json.loads(input_str) def _get_next_step(self, task: Task) - Optional[Step]: 根据依赖关系获取下一个可执行的步骤简单FIFO调度 for step in task.steps: if step.status StepStatus.PENDING: # 检查所有依赖是否都已完成 deps_met all( any(s.id dep_id and s.status StepStatus.SUCCESS for s in task.steps) for dep_id in step.depends_on ) if deps_met: return step return None def _is_task_finished(self, task: Task) - bool: 判断任务是否完成所有步骤都成功或失败或有最终结果 if task.final_result is not None: return True # 简单逻辑所有步骤都不是PENDING或RUNNING状态 for step in task.steps: if step.status in [StepStatus.PENDING, StepStatus.RUNNING]: return False return True def _summarize_results(self, task: Task) - str: 汇总任务结果可以简单拼接最后一步的结果也可以用LLM总结 # 简单实现返回最后一个成功步骤的结果 for step in reversed(task.steps): if step.status StepStatus.SUCCESS: return step.result return “任务未产生有效结果。”这个引擎实现了最基本的功能依赖检查、顺序执行、上下文传递和简单的错误处理。它构成了自动化任务链的骨架。实操心得上下文context的管理是任务链引擎中最容易出乱子的地方。我建议采用命名规范比如使用step_[step_id]_result作为键来存储步骤原始结果。对于需要被后续步骤复用的结构化数据可以设计一个更精细的上下文管理器支持数据类型文本、JSON、二进制和版本避免数据被意外覆盖或格式错误导致下游步骤失败。5. 记忆与反思让Agent在长程任务中持续学习一个只会执行预设流程的Agent其智能是有限的。想象一下你让Agent“每天下午三点检查服务器日志如果有错误就发邮件通知我”。如果它每次检查都从零开始完全记不住昨天发现了什么错误、已经通知过谁那它的价值就大打折扣。因此记忆和反思能力是高级Agent的标配。5.1 短期记忆与长期记忆存储什么如何存储Agent的记忆通常分为两类短期记忆/对话记忆保存当前任务会话中的上下文包括用户消息、Agent的思考、工具调用和结果。这直接服务于当前的规划和执行循环我们之前维护的“对话历史”就是短期记忆。它的特点是容量小、关联性强通常以列表或滑动窗口的形式保存在内存中。长期记忆存储跨越多个任务会话的、需要持久化的信息。例如实体记忆用户的偏好“用户喜欢用Markdown格式看报告”、项目信息“项目A的API密钥是XXX”。过程记忆/经验过去执行类似任务的成功步骤、遇到的坑及其解决方案。事实知识从工具调用中提取并沉淀下来的结构化信息如“公司上季度销售额为1000万”。实现长期记忆最简单的方式是使用一个键值数据库如Redis或文档数据库如MongoDB。更复杂的系统会使用向量数据库如ChromaDB, Pinecone来存储记忆的嵌入向量从而实现基于语义相似度的记忆检索。一个结合了向量检索的长期记忆模块示例import chromadb from chromadb.utils import embedding_functions class LongTermMemory: def __init__(self, persist_path“./chroma_db”): self.client chromadb.PersistentClient(pathpersist_path) # 使用一个通用的嵌入模型如all-MiniLM-L6-v2 self.embedding_fn embedding_functions.SentenceTransformerEmbeddingFunction(model_name“all-MiniLM-L6-v2”) self.collection self.client.get_or_create_collection( name“agent_memories”, embedding_functionself.embedding_fn ) def store(self, content: str, metadata: dict None): 存储一段记忆 doc_id f“mem_{int(time.time())}_{hash(content)}” self.collection.add( documents[content], metadatas[metadata] if metadata else [{}], ids[doc_id] ) return doc_id def retrieve(self, query: str, n_results: int 3) - List[dict]: 根据查询检索相关记忆 results self.collection.query( query_texts[query], n_resultsn_results ) memories [] if results[‘documents’]: for doc, meta in zip(results[‘documents’][0], results[‘metadatas’][0]): memories.append({“content”: doc, “metadata”: meta}) return memories def retrieve_relevant_to_task(self, task_objective: str, context: str) - List[str]: 检索与当前任务相关的记忆 # 可以将任务目标和当前上下文拼接起来作为查询提高相关性 query f“Task: {task_objective}. Context: {context}” memories self.retrieve(query) return [m[“content”] for m in memories]在任务开始或执行到某个决策点时我们可以调用retrieve_relevant_to_task方法将相关的历史经验注入到给LLM的Prompt中例如“根据过去经验处理类似‘销售数据分析’任务时先检查数据完整性再进行聚合计算成功率更高。” 这能显著提升Agent的决策质量。5.2 反思从经验中提炼“元知识”记忆的更高阶应用是反思。反思是指Agent在任务结束后或关键节点主动回顾执行过程总结经验教训并将其转化为可复用的知识存入长期记忆。我们可以设计一个“反思”步骤在任务结束时自动触发或者当任务失败时触发。这个步骤本身也是一个LLM调用请你作为AI助手对刚刚完成的任务进行反思和总结。 任务目标{task_objective} 执行步骤与结果 {将整个任务链的步骤状态和结果以文本形式列出} 请从以下角度进行总结 1. 任务是否成功完成如果没有根本原因是什么 2. 哪些步骤是关键的或效率较低的 3. 遇到了哪些意外或错误是如何解决的 4. 从本次任务中可以提炼出哪些对未来执行类似任务有帮助的经验或最佳实践 请将你的反思输出为结构化的JSON格式包含‘success’, ‘root_cause’, ‘key_insights’, ‘best_practices’等字段。将LLM生成的反思结果特别是best_practices存储到长期记忆中。当下次遇到类似任务时这些“元知识”就能被检索出来指导新的规划形成“执行 - 反思 - 学习 - 更好执行”的闭环。深度思考记忆和反思机制是Agent实现“个性化”和“持续进化”的关键。一个没有记忆的Agent每次交互都是独立的无法建立与用户或环境的深度联系。而一个具备反思能力的Agent则能从错误中学习不断优化自己的行为模式。在设计时需要考虑记忆的隐私、安全性和相关性过滤避免注入无关或过时的信息干扰当前任务。6. 实战构建一个自动化数据分析与报告Agent理论说了这么多我们来动手实现一个具体的、有实用价值的Agent一个自动化数据分析与报告生成Agent。它的目标是用户给定一个数据文件如CSV和一个分析需求Agent能自动完成数据读取、清洗、分析、可视化并生成一份图文并茂的分析报告。6.1 定义工具集Agent的能力边界首先我们需要为这个Agent装备一套数据分析相关的“工具”。read_csv_file(file_path): 读取CSV文件返回DataFrame的JSON字符串或摘要。get_data_summary(data): 获取数据的基本概览行数、列数、数据类型、缺失值。clean_data(data, instructions): 根据指令清洗数据处理缺失值、去重、类型转换。指令可以是自然语言由LLM解析。perform_analysis(data, analysis_type, columns): 执行特定分析如描述性统计、相关性分析、分组聚合等。create_visualization(data, viz_type, x, y, title): 创建图表折线图、柱状图、散点图等并保存为图片文件返回文件路径。write_markdown_report(sections): 将分析结果、图表路径和文本描述整合成Markdown报告。save_report(report_content, output_path): 将报告保存到指定路径。每个工具都需要精心设计其description和parameters的Schema确保LLM能准确理解和使用。例如create_visualization的Schema可能如下{ “name”: “create_visualization”, “description”: “根据提供的数据和参数创建图表。数据应是一个列名和值的字典列表。支持的类型有line折线图bar柱状图scatter散点图。图表将保存为PNG文件。”, “parameters”: { “type”: “object”, “properties”: { “data”: {“type”: “array”, “description”: “要绘制的数据格式为字典列表。”}, “viz_type”: {“type”: “string”, “enum”: [“line”, “bar”, “scatter”]}, “x_column”: {“type”: “string”, “description”: “用作X轴的列名。”}, “y_column”: {“type”: “string”, “description”: “用作Y轴的列名。”}, “title”: {“type”: “string”, “description”: “图表的标题。”} }, “required”: [“data”, “viz_type”, “x_column”, “y_column”] } }6.2 设计任务规划Prompt引导Agent思考分析流程对于数据分析任务我们可以设计一个更具引导性的规划Prompt将通用数据分析流程融入其中你是一个数据分析专家AI。用户将提供一个数据文件和分析目标。 你的任务是制定一个详细的数据分析流水线。 请遵循标准数据分析流程1. 数据获取与概览 2. 数据清洗 3. 探索性分析 4. 深入分析/可视化 5. 报告撰写。 可用的工具 {tool_schemas_json} 请为以下任务生成执行计划。你的回答必须是严格的JSON格式 { “thought”: “你的推理过程解释每一步的必要性。”, “plan”: [ {“step”: 1, “action”: “...”, “tool”: “...”, “input”: {...}}, ... ] } 任务文件路径为“{file_path}”分析目标是“{analysis_goal}”。这个Prompt通过明确“标准数据分析流程”极大地约束和引导了LLM的规划方向使其生成的计划更专业、更可靠。6.3 组装与运行见证自动化链的威力现在我们将所有组件组装起来初始化创建ToolRegistry注册所有数据分析工具。创建TaskChainEngine和LongTermMemory可选。接收任务用户输入file_path“sales_data.csv”和analysis_goal“分析各产品线的月度销售趋势和贡献率”。生成计划将任务目标和工具Schema发送给LLM获得初始执行计划。创建任务对象将LLM返回的plan转化为Task和Step对象。执行引擎启动TaskChainEngine.run_task(task)。动态调整可选在引擎循环中如果某一步骤失败如数据清洗指令不明确可以触发一个“重新规划”的子流程让LLM根据当前错误和上下文调整后续步骤。输出结果任务完成后从task.final_result中获取生成的Markdown报告路径。整个过程中用户只需提供文件和目标剩下的数据读取、质量检查、趋势计算、图表生成、报告整合全部由Agent自动完成。你可能会看到控制台输出类似这样的日志开始执行任务分析各产品线的月度销售趋势和贡献率。 执行步骤读取销售数据文件‘sales_data.csv’... 步骤成功结果数据已加载共1000行5列。 执行步骤检查数据概览和缺失值... 步骤成功结果数据完整无缺失值。各列类型正确。 执行步骤按月份和产品线分组计算销售额... 步骤成功结果分组聚合完成。 执行步骤创建月度销售趋势折线图... 步骤成功结果图表已保存至‘./viz/monthly_trend.png’。 执行步骤创建产品线销售额占比饼图... 步骤成功结果图表已保存至‘./viz/product_pie.png’。 执行步骤撰写分析报告... 步骤成功结果报告已生成包含趋势分析、贡献率分析和图表引用。 任务完成。最终报告位于./reports/sales_analysis_20231027.md避坑指南在实现这类数据密集型Agent时最大的挑战是数据在步骤间的流转。LLM和工具处理的数据格式可能不同如DataFrame vs JSON字符串。我强烈建议在context中定义清晰的数据接口契约。例如规定read_csv_file工具的输出是一个包含df_jsonDataFrame的JSON字符串和df_summary文本摘要的字典。后续工具都约定从context[“df_json”]中获取原始数据。同时要善用Python的ast.literal_eval或json.loads进行安全的格式转换避免eval带来的安全风险。7. 进阶思考Agent架构的挑战与未来方向构建一个可用的Agent原型并不难但要让它真正稳定、可靠、智能地运行在生产环境中我们还需要面对诸多挑战。7.1 可靠性挑战错误处理、超时与回滚工具执行的不可靠性网络API可能超时数据库可能连接失败文件可能被占用。引擎必须为每个工具调用设置超时和重试机制。对于非幂等操作如发送邮件、写入数据库重试需要格外小心。LLM响应的不确定性LLM可能不按要求的格式输出可能生成不合法的工具参数。引擎需要解析失败后的备选方案比如尝试用更严格的Prompt重试或者降级到让用户确认。任务链的回滚如果一个多步骤任务在后期失败可能需要回滚之前步骤产生的副作用如删除已创建的文件、撤销数据库操作。实现完善的Saga模式事务回滚对Agent来说非常复杂通常需要根据业务场景设计补偿操作。7.2 效率与成本优化规划与执行的平衡过于详细的规划每一步都让LLM决定会导致延迟高、成本高。过于僵化的流程完全预设又失去了灵活性。一个混合策略是对于通用、稳定的子流程如“数据清洗流程”可以封装成一个更大的“复合工具”让LLM直接调用这个复合工具内部则用固定代码执行减少LLM调用次数。上下文长度管理随着任务进行对话历史包含所有思考、行动、观察会越来越长可能超出模型的上下文窗口。需要设计摘要策略将过长的历史压缩成精炼的要点只保留对当前决策最关键的信息。缓存与记忆复用相同的工具调用如查询某城市天气结果在一段时间内可以缓存。长期记忆中的经验也可以避免重复的试错。7.3 评估与监控如何知道你的Agent做得好不好定义评估指标对于自动化任务最终结果的成功率Task Success Rate是核心指标。此外还可以衡量平均完成步骤数效率、工具调用准确率、用户满意度等。建立评估数据集构建一组涵盖不同难度和场景的测试任务定期运行Agent进行评估跟踪指标变化。可观测性记录Agent完整的执行轨迹Thought, Action, Observation这对于调试和优化至关重要。可以像日志系统一样将这些轨迹收集起来用于分析故障点和性能瓶颈。手写一个AI Agent的过程是一个不断在“赋予智能”和“施加约束”之间寻找平衡的过程。我们从让大模型调用单个函数开始逐步为它添加规划、记忆、执行引擎等组件本质上是在用软件工程的方法为概率模型构建一个确定性的、可靠的“外壳”。这个外壳决定了Agent的能力下限而内部LLM的智能则决定了其上限。未来的方向可能会朝着更模块化、更可解释的Agent框架发展也会出现更多专注于规划、工具使用或记忆的专项模型。但无论如何理解从Function Calling到自动化任务链这一完整的技术栈都将是你构建下一代AI应用不可或缺的核心能力。