从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 Calling：Agent与外部世界交互的“手和脚”

在构建Agent之前，我们必须先解决一个基本问题：AI如何与我们的代码、数据库、API乃至整个互联网进行交互？答案就是Function Calling（函数调用）。这是目前大模型与外部工具集成最主流、最可靠的方式。你可以把它理解为Agent的“手和脚”——没有它，Agent再聪明，也只能困在文本的牢笼里空想。

2.1 Function Calling的本质：结构化指令的生成与解析

很多人误以为Function Calling是大模型“直接执行”了我们的函数。其实不然。大模型（如GPT-4）本身并不能运行你的Python代码。Function Calling的流程是一个精巧的“请求-响应-执行”循环：

1. 定义工具（函数）：你告诉大模型，你有哪些工具可用。这通过一个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至关重要，它是大模型决定是否、以及如何调用该工具的唯一依据。描述必须清晰、无歧义。

2. 模型决策与结构化输出：你将用户的问题（如“北京天气怎么样？”）和定义好的工具列表一起发送给大模型。模型会理解问题，并判断是否需要、以及需要调用哪个工具。如果需要，它不会直接执行，而是返回一个结构化的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里。这标志着模型进入了“工具调用模式”。

3. 本地执行与结果返回：你的程序接收到这个JSON后，在本地找到对应的get_current_weather函数，用解析出的参数（location=”北京”）执行它，并得到真实结果（如{“temperature”: 22, “condition”: “晴朗”}）。

4. 结果反馈与继续对话：你将工具执行的结果，以特定格式追加到对话历史中，再次发送给大模型。

   { "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需要具备几个要素：

1. 明确角色和工具清单：让模型清楚自己的能力和可用资源。
2. 要求结构化输出：强制模型以JSON等格式输出，便于程序解析，避免自由文本的随意性。
3. 引导深度思考：通过要求输出“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”: “user@example.com”, “subject”: “上月销售分析报告”, “body”: “[上一步的报告内容]”}} ] }

踩坑实录：在早期实现中，我让模型直接输出步骤列表，但经常遇到步骤逻辑混乱、工具选择错误的问题。后来强制加入“thought”推理字段后，计划的合理性大幅提升。因为模型为了写出合理的“思考”，必须更深入地理解任务和工具之间的关系。这相当于让模型自己给自己做了一次“审核”。

3.2 动态规划与ReAct模式：边执行边调整

上述的“先规划，后执行”模式是静态的。但在真实场景中，计划赶不上变化。执行第一步“读取文件”时，可能发现文件不存在；执行“数据分析”时，可能发现数据格式不符合预期。一个鲁棒的Agent必须具备动态重新规划的能力。

这就是ReAct (Reason + Act)模式大放异彩的地方。ReAct的核心思想是将推理（Reason）和行动（Act）交织在一个循环中：

1. 思考（Think）：根据当前目标、已知信息和历史观察，思考下一步应该做什么。
2. 行动（Act）：根据思考，选择执行一个动作（如调用工具）。
3. 观察（Observe）：获取动作执行的结果（成功或失败）。
4. 循环：基于新的观察，再次进入“思考”步骤，直到任务完成或无法继续。

如何用代码实现这个循环？关键在于维护一个不断增长的“上下文”，其中包含：

• 系统指令：定义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的记忆通常分为两类：

1. 短期记忆/对话记忆：保存当前任务会话中的上下文，包括用户消息、Agent的思考、工具调用和结果。这直接服务于当前的规划和执行循环，我们之前维护的“对话历史”就是短期记忆。它的特点是容量小、关联性强，通常以列表或滑动窗口的形式保存在内存中。

2. 长期记忆：存储跨越多个任务会话的、需要持久化的信息。例如：

   • 实体记忆：用户的偏好（“用户喜欢用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(path=persist_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_function=self.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_results=n_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装备一套数据分析相关的“工具”。

1. read_csv_file(file_path): 读取CSV文件，返回DataFrame的JSON字符串或摘要。
2. get_data_summary(data): 获取数据的基本概览（行数、列数、数据类型、缺失值）。
3. clean_data(data, instructions): 根据指令清洗数据（处理缺失值、去重、类型转换）。指令可以是自然语言，由LLM解析。
4. perform_analysis(data, analysis_type, columns): 执行特定分析，如描述性统计、相关性分析、分组聚合等。
5. create_visualization(data, viz_type, x, y, title): 创建图表（折线图、柱状图、散点图等），并保存为图片文件，返回文件路径。
6. write_markdown_report(sections): 将分析结果、图表路径和文本描述整合成Markdown报告。
7. 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 组装与运行：见证自动化链的威力

现在，我们将所有组件组装起来：

1. 初始化：创建ToolRegistry，注册所有数据分析工具。创建TaskChainEngine和LongTermMemory（可选）。
2. 接收任务：用户输入file_path=“sales_data.csv”和analysis_goal=“分析各产品线的月度销售趋势和贡献率”。
3. 生成计划：将任务目标和工具Schema发送给LLM，获得初始执行计划。
4. 创建任务对象：将LLM返回的plan转化为Task和Step对象。
5. 执行引擎：启动TaskChainEngine.run_task(task)。
6. 动态调整（可选）：在引擎循环中，如果某一步骤失败（如数据清洗指令不明确），可以触发一个“重新规划”的子流程，让LLM根据当前错误和上下文，调整后续步骤。
7. 输出结果：任务完成后，从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_json（DataFrame的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应用不可或缺的核心能力。