企业官网建设流程全解析

1. 项目概述：从“单次问答”到“自主执行”的跨越

最近和几个做产品的朋友聊天，大家不约而同地提到了一个词：AI Agent。不再是年初那种“我有一个想法”的兴奋，而是实打实的困惑——“这东西到底怎么落地？”、“都说能自动化，我该从哪开始写？” 这种感觉我特别理解，就像几年前大家一窝蜂搞微服务，但真上手时，连服务边界怎么划都头疼。AI Agent 现在也到了这个阶段：概念满天飞，但能跑通一个从想法到代码的完整闭环，才是硬道理。

所以，今天我们不谈那些宏大的架构图，也不复读“LLM是大脑”这种正确的废话。我们就干一件事：手把手，从零开始，写一个能真正干活的 AI Agent。我们的目标很明确：让一个大型语言模型（LLM）不仅能理解我们的指令，还能像程序员一样，自己调用工具（Function Calling），并把多个工具串联起来，完成一个复杂的、多步骤的任务。比如，你告诉它“帮我查一下北京明天天气，如果下雨就提醒我带伞，并把提醒发到我的邮箱”，它应该能自动分解为“查询天气API -> 判断逻辑 -> 调用邮件发送服务”这一系列动作。这就是所谓的“自动化任务链”。

为什么从 Function Calling 开始？因为这是让 LLM 从“聊天机器”蜕变为“执行代理”最关键的一步。没有工具调用能力，LLM 再聪明，也只是一个知识渊博的“顾问”，无法在数字世界里留下任何痕迹。而任务链，则是将单点能力编织成复杂工作流的骨架。这个过程会涉及到提示工程、状态管理、错误处理等一系列工程化问题，这正是 Agent 开发的核心乐趣与挑战所在。

无论你是想为自己的产品增加一个智能自动化模块，还是单纯对 Agent 的实现原理感到好奇，这篇内容都将为你提供一个可运行、可调试、可扩展的起点。我们会用最直白的代码，拆解每一个环节，并分享我在搭建过程中踩过的那些坑。

2. 核心架构设计：像搭积木一样构建你的 Agent

在动手写代码之前，我们必须先想清楚这个 Agent 的“工作流程”和“组成部分”。一个典型的、基于 Function Calling 的任务链 Agent，其核心循环可以概括为“思考-行动-观察”。

2.1 核心循环：ReAct 模式的实践

目前最主流、也最有效的 Agent 推理模式是ReAct (Reasoning + Acting)。它模仿了人类解决问题的方式：

1. 推理：根据当前目标和已有信息，思考下一步该做什么。
2. 行动：执行思考后决定的操作，比如调用一个函数。
3. 观察：获取行动的结果，并将其作为新的信息输入下一轮推理。

这个循环会一直持续，直到任务完成或无法继续。在我们的天气-邮件提醒场景里，Agent 的内心活动可能是这样的：

• 推理1：“用户想要明天的天气和提醒。我需要先获取天气数据。有一个get_weather函数可用。”
• 行动1：调用get_weather(location="北京", date="tomorrow")。
• 观察1：函数返回{“weather”: “rain”, “temperature”: “18°C”}。
• 推理2：“天气是下雨。用户的条件是‘如果下雨就提醒’。所以我需要生成提醒内容并发送邮件。我有send_email函数。”
• 行动2：调用send_email(subject="天气提醒", body="明天北京下雨，记得带伞。", to="user@example.com")。
• 观察2：函数返回{“status”: “success”}。
• 推理3：“邮件发送成功，所有用户要求都已满足。任务完成。”

我们的代码架构就是要支撑起这个循环。

2.2 组件拆解：四大核心模块

为了实现上述循环，我们需要设计几个关键模块：

1. 工具层：这是 Agent 的“手”和“脚”。每个工具对应一个 Python 函数，并附带一个清晰的描述，告诉 LLM 这个工具是干什么的、需要什么参数。例如get_weather工具的描述会说明它用于查询指定地点和日期的天气。
2. 推理引擎：这是 Agent 的“大脑”，通常就是 LLM 本身（如 GPT-4, Claude, 或本地部署的模型）。它的核心职责是：理解用户请求和当前对话历史，决定下一步是“直接回答”还是“调用某个工具”。如果决定调用工具，它还需要根据工具描述，生成符合要求的参数。
3. 函数调用处理器：这是“神经中枢”，负责连接大脑和手脚。它接收 LLM 发出的“调用工具X，参数为Y”的指令，在代码中找到对应的工具函数，安全地执行它，并将执行结果格式化，准备送回给 LLM 进行下一轮推理。
4. 状态管理与任务链调度：这是“工作记忆”和“项目经理”。它需要维护整个对话的历史（包括用户消息、AI回复、工具调用及结果），确保上下文不丢失。更重要的是，它要驱动整个 ReAct 循环，判断任务是否结束，防止陷入无限循环。

一个常见的架构误区是试图用一个超级复杂的类来搞定一切。我的经验是，初期务必保持模块的轻量和清晰。下面，我们就从最基础的 Function Calling 实现开始。

3. 基础实现：让 LLM 学会“用手”（Function Calling）

Function Calling 的本质，是让 LLM 的输出结构化。普通的聊天输出是自然语言文本，而函数调用要求 LLM 输出一个标准的 JSON 对象，指明要调用的函数名和参数。

3.1 定义你的工具集

首先，我们定义几个简单的工具。这里的关键在于工具描述，它必须清晰、无歧义。

# tools.py import json import requests from datetime import datetime, timedelta def get_weather(location: str, date: str) -> str: """ 获取指定城市和日期的天气信息。 Args: location: 城市名，例如“北京”、“上海”。 date: 日期，支持“today”、“tomorrow”或“YYYY-MM-DD”格式。 Returns: 一个描述天气的字符串。 """ # 注意：这里是一个模拟函数。真实场景应调用如和风天气、OpenWeatherMap等API。 # 为演示，我们返回模拟数据。 weather_map = { “北京”: {“today”: “sunny, 25°C”, “tomorrow”: “rainy, 18°C”}, “上海”: {“today”: “cloudy, 28°C”, “tomorrow”: “sunny, 30°C”}, } # 简单的日期解析 if date == “today”: date_key = “today” elif date == “tomorrow”: date_key = “tomorrow” else: date_key = “today” # 简化处理 weather = weather_map.get(location, {}).get(date_key, “Weather data not available”) return json.dumps({“location”: location, “date”: date, “weather”: weather}) def send_email(subject: str, body: str, to: str) -> str: """ 发送一封电子邮件。 Args: subject: 邮件主题。 body: 邮件正文内容。 to: 收件人邮箱地址。 Returns: 发送状态的字符串。 """ # 模拟发送邮件，真实场景可使用smtplib或第三方邮件服务API。 print(f”[模拟] 发送邮件给 {to}“) print(f”主题：{subject}“) print(f”正文：{body}“) return json.dumps({“status”: “success”, “message”: f”Email to {to} sent successfully.”}) def search_web(query: str) -> str: """ 在互联网上搜索信息。 Args: query: 搜索关键词。 Returns: 搜索结果的摘要字符串。 """ # 模拟搜索，真实场景可集成Serper API、Google Search API等。 print(f”[模拟] 正在搜索：{query}“) # 这里可以模拟返回一些搜索结果 result = f”关于‘{query}’的模拟搜索结果：相关文章1，相关文章2。“ return json.dumps({“query”: query, “result”: result}) # 工具元数据列表，用于提供给LLM TOOLS = [ { “type”: “function”, “function”: { “name”: “get_weather”, “description”: “获取指定城市和日期的天气信息。”, “parameters”: { “type”: “object”, “properties”: { “location”: {“type”: “string”, “description”: “城市名，如‘北京’、‘上海’。”}, “date”: {“type”: “string”, “description”: “日期，如‘today’、‘tomorrow’或‘2024-05-20’。”} }, “required”: [“location”, “date”] } } }, { “type”: “function”, “function”: { “name”: “send_email”, “description”: “发送一封电子邮件。”, “parameters”: { “type”: “object”, “properties”: { “subject”: {“type”: “string”, “description”: “邮件主题。”}, “body”: {“type”: “string”, “description”: “邮件正文内容。”}, “to”: {“type”: “string”, “description”: “收件人邮箱地址。”} }, “required”: [“subject”, “body”, “to”] } } }, # 可以继续添加其他工具... ]

注意：工具描述中的parameters定义至关重要。它必须严格遵循 JSON Schema 格式。模糊的描述会导致 LLM 生成错误的参数。例如，date字段明确说明支持的格式，能极大提高调用准确率。

3.2 实现单轮函数调用

有了工具定义，我们接下来实现一个简单的 Agent 核心类，处理单轮的“用户提问 -> LLM思考 -> 执行工具 -> 返回结果”流程。这里我们以 OpenAI API 为例。

# simple_agent.py import openai import json class SimpleAgent: def __init__(self, api_key, model=“gpt-3.5-turbo”): self.client = openai.OpenAI(api_key=api_key) self.model = model self.conversation_history = [] # 维护对话历史 def run(self, user_input: str): """运行一轮Agent循环""" # 1. 将用户输入加入历史 self.conversation_history.append({“role”: “user”, “content”: user_input}) # 2. 调用LLM，并告知它可用的工具 try: response = self.client.chat.completions.create( model=self.model, messages=self.conversation_history, tools=TOOLS, # 传入我们定义的工具列表 tool_choice=“auto”, # 让模型自行决定是否调用工具 ) except Exception as e: return f”调用LLM API失败：{e}“ response_message = response.choices[0].message # 3. 检查LLM是否决定调用工具 tool_calls = response_message.tool_calls if tool_calls: # 4. 处理工具调用（可能多个） for tool_call in tool_calls: function_name = tool_call.function.name function_args = json.loads(tool_call.function.arguments) print(f”[Agent] 决定调用工具：{function_name}“) print(f” 参数：{function_args}“) # 5. 执行对应的工具函数 function_to_call = globals().get(function_name) if function_to_call: try: function_response = function_to_call(**function_args) # 6. 将工具执行结果作为新的消息追加到历史中，角色为“tool” self.conversation_history.append(response_message) # 先保存AI的请求 self.conversation_history.append({ “role”: “tool”, “content”: function_response, “tool_call_id”: tool_call.id }) print(f”[Tool] {function_name} 返回：{function_response}“) except Exception as e: error_msg = json.dumps({“error”: str(e)}) self.conversation_history.append({ “role”: “tool”, “content”: error_msg, “tool_call_id”: tool_call.id }) print(f”[Tool] {function_name} 执行出错：{e}“) else: print(f”[Error] 未找到工具函数：{function_name}“) # 7. 工具调用后，需要再次调用LLM，让它根据工具结果生成最终回答 final_response = self.client.chat.completions.create( model=self.model, messages=self.conversation_history, ) final_message = final_response.choices[0].message.content self.conversation_history.append({“role”: “assistant”, “content”: final_message}) return final_message else: # 没有工具调用，直接返回LLM的回复 ai_response = response_message.content self.conversation_history.append({“role”: “assistant”, “content”: ai_response}) return ai_response # 使用示例 if __name__ == “__main__”: agent = SimpleAgent(api_key=“your-openai-api-key”) result = agent.run(“北京明天天气怎么样？”) print(“最终回答：”, result)

运行这段代码，你会看到类似以下的输出：

[Agent] 决定调用工具：get_weather 参数：{‘location’: ‘北京’， ‘date’: ‘tomorrow’} [Tool] get_weather 返回：{“location”: “北京”， “date”: “tomorrow”， “weather”: “rainy, 18°C”} 最终回答： 北京明天（2024-05-21）的天气是雨天，气温大约18°C。记得带伞哦。

实操心得1：tool_choice参数的选择tool_choice参数控制LLM调用工具的倾向性。设为“auto”（默认）时，由模型决定是否调用。如果你明确要求它必须调用某个工具，可以设为{“type”: “function”， “function”: {“name”: “get_weather”}}。这在构建确定性的工作流时非常有用。但大部分情况下，“auto”配合清晰的工具描述和用户指令，效果最好。

实操心得2：对话历史的管理注意代码中conversation_history的维护。我们必须将每一次的user、assistant消息以及tool角色的执行结果都按顺序保存。这是实现多轮对话和任务链的基础。缺少tool消息，LLM 就不知道上次工具调用的结果，任务链就会断裂。

4. 进阶实现：构建自动化任务链

单轮调用只是开始。真正的价值在于让 Agent 自动串联多个步骤。这就需要我们升级 Agent，使其具备循环执行和状态判断的能力。

4.1 实现 ReAct 循环引擎

我们在SimpleAgent的基础上，构建一个更强大的TaskChainAgent。它的核心是一个run循环，在达到停止条件前不断运行“思考-行动-观察”。

# task_chain_agent.py import openai import json from typing import List, Dict, Any, Optional class TaskChainAgent: def __init__(self, api_key, model=“gpt-3.5-turbo”, max_steps=10): self.client = openai.OpenAI(api_key=api_key) self.model = model self.max_steps = max_steps # 防止无限循环 self.messages: List[Dict[str, Any]] = [] def _call_llm(self) -> Dict[str, Any]: """调用LLM，并处理可能的工具调用请求。""" try: response = self.client.chat.completions.create( model=self.model, messages=self.messages, tools=TOOLS, tool_choice=“auto”, ) return response.choices[0].message except Exception as e: # 更健壮的错误处理 raise Exception(f”LLM API调用失败：{e}“) def _execute_tool(self, function_name: str, arguments: Dict) -> str: """查找并执行工具函数。""" function_to_call = globals().get(function_name) if not function_to_call: return json.dumps({“error”: f”Function ‘{function_name}’ not found.”}) try: # 这里可以加入参数验证、权限检查等 result = function_to_call(**arguments) return result if isinstance(result, str) else json.dumps(result) except Exception as e: return json.dumps({“error”: str(e)}) def run(self, initial_input: str) -> str: """运行任务链，直到完成或达到最大步数。""" self.messages = [{“role”: “user”, “content”: initial_input}] final_answer = None for step in range(self.max_steps): print(f”\n=== 步骤 {step + 1} ===") # 1. 思考：LLM生成回复或工具调用 assistant_message = self._call_llm() self.messages.append(assistant_message) # 2. 检查是否需要行动（调用工具） tool_calls = assistant_message.tool_calls if tool_calls: print(f”[Agent] 决定调用 {len(tool_calls)} 个工具。“) for tool_call in tool_calls: func_name = tool_call.function.name func_args = json.loads(tool_call.function.arguments) print(f” -> 执行 {func_name}， 参数：{func_args}“) # 3. 行动：执行工具 tool_result = self._execute_tool(func_name, func_args) print(f” <- 结果：{tool_result[:100]}...“) # 打印部分结果 # 4. 观察：将结果加入对话历史 self.messages.append({ “role”: “tool”, “content”: tool_result, “tool_call_id”: tool_call.id }) # 本轮有工具调用，继续下一轮循环（让LLM基于工具结果继续思考） continue else: # 没有工具调用，说明LLM认为任务已完成，给出了最终答案 final_answer = assistant_message.content print(f”[Agent] 任务完成，最终答案：{final_answer}“) break else: # 如果for循环正常结束（非break），说明达到最大步数 final_answer = f”任务未在 {self.max_steps} 步内完成，可能陷入循环或任务过于复杂。当前历史：{self.messages[-1].get(‘content’， ‘No content’)}“ return final_answer # 使用示例：一个简单的两步骤任务链 if __name__ == “__main__”: agent = TaskChainAgent(api_key=“your-openai-api-key”) # 一个需要多步推理的任务 result = agent.run(“先查一下北京明天的天气，如果是雨天，就搜索‘雨天出行注意事项’，然后把注意事项总结成一句话发到我的邮箱 test@example.com。”) print(“\n=== 最终输出 ===") print(result)

这个TaskChainAgent已经具备了自动化任务链的核心能力。它会自动执行“查询天气 -> 判断 -> 搜索 -> 发送邮件”这一系列操作。

4.2 任务链的停止条件与稳定性

上面的循环有一个简单的停止条件：LLM 不再输出工具调用（tool_calls为空），而是直接给出自然语言回答。但这并不总是可靠的。

常见问题1：LLM 陷入循环LLM 可能会反复调用同一个工具，或者在不同的工具间无效切换。例如，查询天气后，又去查询时间，然后又查天气。

解决方案：

• 设置最大步数：我们已经做了，这是最后防线。
• 在系统提示中明确指令：在self.messages的开头插入一条system消息，明确告诉 LLM：“你是一个任务执行助手。请逐步思考，在获得足够信息后给出最终答案，不要重复执行相同或无关的操作。”
• 实现状态检查：可以设计一个is_task_complete函数，基于当前对话历史和用户初始目标，用规则或另一个简单的LLM调用来判断任务是否实质上已完成。

# 在run循环中，可以在每一步后加入状态检查 def _check_completion(self, initial_goal: str) -> bool: """一个简单的基于规则的任务完成检查（示例）""" last_msg = self.messages[-1][“content”].lower() goal_keywords = [“天气”， “邮件”， “发送”] # 根据你的任务目标定义 # 如果最后一条消息是AI的最终回答，且包含了目标关键词的确认，可以认为完成 if self.messages[-1][“role”] == “assistant” and not self.messages[-1].get(“tool_calls”): for keyword in goal_keywords: if keyword in initial_goal and keyword in last_msg: return True return False

常见问题2：工具执行失败网络错误、API限制、参数错误都可能导致工具调用失败。失败信息需要清晰地反馈给 LLM，让它有机会调整策略（例如重试或选择备用方案）。

解决方案： 我们的_execute_tool已经做了基本的 try-catch，并将错误信息以结构化格式（JSON）返回。LLM 能够理解这种错误格式，并可能做出如下反应：“邮件发送失败（网络错误），我将先保存提醒内容，稍后重试。” 这需要你在工具描述中提前告知 LLM 可能的错误和应对策略。

5. 工程化考量与性能优化

当一个原型能跑通后，我们要考虑如何让它变得健壮、可用，甚至能上生产环境。

5.1 工具管理的设计模式

上面的例子用globals()查找函数，这在小型项目中可行，但不便于管理和扩展。更好的做法是使用工具注册表模式。

# tool_registry.py class ToolRegistry: def __init__(self): self._tools = {} # name -> function self._descriptions = [] # for LLM def register(self, func, description_schema: dict): """注册一个工具函数及其描述""" self._tools[func.__name__] = func self._descriptions.append(description_schema) def get_tool(self, name: str): return self._tools.get(name) def get_descriptions(self): return self._descriptions # 使用装饰器注册工具 registry = ToolRegistry() def tool(description_schema: dict): def decorator(func): registry.register(func, description_schema) return func return decorator # 定义工具时，同时提供描述 @tool({ “type”: “function”, “function”: { “name”: “get_weather”, “description”: “获取天气...”, “parameters”: {...} } }) def get_weather(location: str, date: str) -> str: # ... 实现 ... pass # 在Agent中，使用 registry.get_tool(name) 和 registry.get_descriptions()

这样做的好处是工具集中管理，支持动态加载和卸载，也方便做权限控制（比如某些Agent只能使用部分工具）。

5.2 上下文长度与历史管理

复杂的任务链会产生很长的对话历史，可能超过模型的上下文窗口。你需要一个历史总结或窗口化的策略。

• 滑动窗口：只保留最近 N 轮对话。
• 总结压缩：当历史过长时，调用 LLM 对之前的对话进行总结，用总结文本替代旧历史。
• 向量存储检索：将历史对话存入向量数据库，每次只检索与当前问题最相关的片段。这是构建“长期记忆”的高级方式，但对于大多数任务链场景，滑动窗口或简单总结通常足够。

def summarize_history_if_needed(self, max_tokens=8000): """当历史消息预估token数超限时，进行总结压缩（简化示例）""" # 此处需要估算token数，可用 tiktoken 库 estimated_tokens = self._estimate_tokens(self.messages) if estimated_tokens > max_tokens: # 构造一个总结请求 summary_prompt = [ {“role”: “system”， “content”: “请将以下对话历史简洁地总结成一段话，保留所有关键事实、决策和结果。”}, {“role”: “user”， “content”: str(self.messages[:-10])} # 总结除最近10条外的历史 ] # 调用LLM生成总结... summary = self._call_llm_for_summary(summary_prompt) # 用总结替换旧历史 self.messages = [{“role”: “system”， “content”: f”先前对话的总结：{summary}“}] + self.messages[-10:]

5.3 异步执行与超时控制

如果工具调用涉及网络请求（如调用外部API），同步执行会阻塞整个Agent。使用异步可以大幅提升效率，尤其是当多个工具可以并行执行时。

import asyncio class AsyncTaskChainAgent(TaskChainAgent): async def _execute_tool_async(self, function_name: str, arguments: Dict) -> str: # 将同步工具函数包装为异步，或直接实现异步工具 loop = asyncio.get_event_loop() # 注意：如果工具本身是CPU密集型，需要在线程池中运行，避免阻塞事件循环 result = await loop.run_in_executor(None, self._execute_tool, function_name, arguments) return result async def run_async(self, initial_input: str) -> str: self.messages = [{“role”: “user”, “content”: initial_input}] # ... 异步版本的run循环，使用 await 调用 _call_llm 和 _execute_tool_async ...

同时，务必为每个工具调用和LLM请求设置超时，避免一个环节卡死整个Agent。

6. 调试、测试与监控实战

开发 Agent 最耗时的部分往往是调试。因为错误可能来自：1) 你的代码逻辑，2) 工具API，3) LLM的“不可预测”的输出。

6.1 建立可观测性

给你的 Agent 加上详细的日志，记录每一步的输入输出。

import logging logging.basicConfig(level=logging.INFO, format=‘%(asctime)s - %(name)s - %(levelname)s - %(message)s’) logger = logging.getLogger(__name__) class LoggingAgent(TaskChainAgent): def run(self, initial_input: str) -> str: logger.info(f”开始处理任务：{initial_input}“) self.messages = [{“role”: “user”, “content”: initial_input}] for step in range(self.max_steps): logger.info(f”步骤{step}，当前历史长度：{len(self.messages)}“) assistant_message = self._call_llm() logger.debug(f”LLM回复：{assistant_message}“) # ... 后续处理 ... if tool_calls: logger.info(f”调用工具：{[tc.function.name for tc in tool_calls]}“) # ... logger.info(f”任务结束，结果：{final_answer}“) return final_answer

更高级的做法是，将每一步的(input, output, tool_call, tool_result)记录到数据库或文件中，便于事后分析和复现问题。

6.2 编写确定性测试

Agent 的非确定性是测试的难点。但我们可以通过一些手段提高测试的可靠性。

• Mock 外部依赖：使用unittest.mock模拟所有外部 API 调用（天气API、邮件服务、搜索API），让测试在完全可控的环境下运行。
• 测试固定场景：对于给定的输入，虽然LLM的回复可能略有不同，但其“调用get_weather工具”这个决策应该是稳定的。我们可以断言在特定输入下，Agent 是否发起了预期的工具调用。

from unittest.mock import patch, MagicMock def test_agent_weather_inquiry(): agent = TaskChainAgent(api_key=“fake-key”) # Mock掉OpenAI客户端，让它返回我们预设的、包含工具调用的响应 mock_message = MagicMock() mock_tool_call = MagicMock() mock_tool_call.function.name = “get_weather” mock_tool_call.function.arguments = json.dumps({“location”: “北京”， “date”: “tomorrow”}) mock_message.tool_calls = [mock_tool_call] mock_message.content = None mock_response = MagicMock() mock_response.choices[0].message = mock_message with patch.object(agent.client.chat.completions, ‘create’， return_value=mock_response): with patch(‘tools.get_weather’， return_value=‘{“weather”: “sunny”}’) as mock_weather: result = agent.run(“北京明天天气？”) # 验证是否调用了get_weather mock_weather.assert_called_once_with(location=“北京”， date=“tomorrow”) # 可以进一步验证最终result中是否包含“sunny”等关键词 assert “sunny” in result.lower()

6.3 处理LLM的“偏航”行为

有时LLM会不按常理出牌，比如：

• 参数格式错误：要求date是字符串，它却生成一个数字或复杂对象。
• 调用不存在的工具：自己编造一个工具名。
• 在应该给出最终答案时又调用工具。

应对策略：

1. 前置参数校验：在_execute_tool中，在执行前先用 JSON Schema 验证参数。不通过则直接返回错误给LLM。
2. 后置输出解析：对LLM的直接回复（非工具调用）也可以进行结构化解析，确保其格式符合预期。
3. 优化系统提示：这是最重要的。在系统提示中明确规则：“你必须严格按照提供的工具描述来调用。如果用户请求无法用现有工具完成，请直接告知用户，不要编造工具。”

7. 从原型到应用：扩展思路与框架选择

当你掌握了手写 Agent 的核心逻辑后，可能会发现需要重复处理很多样板代码（历史管理、错误处理、异步、流式输出等）。这时，可以考虑使用成熟的 Agent 框架来提升开发效率。

7.1 主流框架浅析

• LangChain / LangGraph：生态最丰富，提供了大量现成的工具集成、记忆模块和链式编排能力。LangGraph 特别适合构建有复杂状态流转的 Agent。缺点是抽象层次高，有时感觉“黑盒”，调试稍复杂。
• AutoGen：由微软推出，擅长多智能体协作场景。可以轻松定义多个不同角色的Agent，让它们通过对话共同完成任务。如果你需要“客服Agent”和“技术专家Agent”协作，AutoGen 很合适。
• Semantic Kernel：微软另一个框架，强调将传统编程逻辑（“原生函数”）与AI语义技能（“语义函数”）深度融合，更适合将AI能力嵌入到现有.NET或Python应用中。
• LlamaIndex：更侧重于数据的索引和检索，用于构建RAG（检索增强生成）应用非常强大。如果你的Agent核心能力是查询私有知识库，可以优先考虑它。

框架选择建议：如果你的需求是快速验证一个复杂的、多步骤的自动化任务，且对控制粒度要求不是极致，LangGraph是目前最平衡的选择。它用“图”的概念来定义工作流，非常直观。

7.2 基于 LangGraph 重构我们的任务链

下面是一个用 LangGraph 实现同样天气-邮件提醒任务的简化示例，感受一下框架带来的抽象：

from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List import operator # 1. 定义状态 class AgentState(TypedDict): messages: Annotated[List, operator.add] # 对话历史 user_request: str # 用户原始请求 weather_info: str # 存放天气信息 needs_umbrella: bool # 是否需要伞 # 2. 定义节点（函数） def decide_action(state: AgentState): """根据当前状态，决定下一步做什么""" # 这里可以集成LLM调用，判断下一步是查天气、发邮件还是结束 # 为简化，我们用规则判断 last_msg = state[“messages”][-1].content if state[“messages”] else “” if “天气” in state[“user_request”] and not state.get(“weather_info”): return {“next”: “get_weather”} elif state.get(“weather_info”) and “雨” in state[“weather_info”] and “邮件” in state[“user_request”]: return {“next”: “send_email”, “needs_umbrella”: True} else: return {“next”: END} def get_weather_node(state: AgentState): # 调用天气工具 weather = get_weather(location=“北京”， date=“tomorrow”) return {“weather_info”: weather, “messages”: [{“role”: “tool”， “content”: weather}]} def send_email_node(state: AgentState): # 调用邮件工具 email_result = send_email(subject=“天气提醒”， body=“明天有雨，请带伞。”， to=“user@example.com”) return {“messages”: [{“role”: “tool”， “content”: email_result}]} # 3. 构建图 workflow = StateGraph(AgentState) workflow.add_node(“decide”， decide_action) workflow.add_node(“get_weather”， get_weather_node) workflow.add_node(“send_email”， send_email_node) workflow.set_entry_point(“decide”) # 设置条件边：根据 decide 节点的返回值，路由到不同节点 workflow.add_conditional_edges( “decide”, lambda x: x[“next”], { “get_weather”: “get_weather”, “send_email”: “send_email”, END: END } ) workflow.add_edge(“get_weather”， “decide”) # 查完天气后，回到决策点 workflow.add_edge(“send_email”， END) # 发完邮件后结束 app = workflow.compile() # 运行这个图 result = app.invoke({“user_request”: “查北京明天天气，下雨就发邮件提醒”， “messages”: []})

可以看到，LangGraph 将工作流可视化为了一个“图”，节点是操作，边是流转逻辑。这对于复杂业务流程的编排和调试非常有帮助。

7.3 最终决策：手写还是用框架？

• 手写：
  • 优点：完全可控，深度理解底层原理，依赖少，轻量级。
  • 缺点：需要自己处理所有细节，扩展复杂功能时容易代码臃肿。
  • 适用：学习阶段、概念验证、功能简单且固定的场景。
• 使用框架：
  • 优点：开箱即用的组件（记忆、检索、工具集成），社区支持好，通常内置了最佳实践和性能优化。
  • 缺点：学习成本，框架抽象可能带来额外的复杂度，有时调试更困难。
  • 适用：构建生产级应用、需要快速集成多种能力、涉及复杂状态和协作的场景。

我的建议是：先从手写开始，彻底弄懂 ReAct 循环、工具调用、历史管理这几个核心概念。当你觉得手写代码开始重复造轮子，或者业务逻辑复杂到用if-else难以维护时，就是引入框架的好时机。此时你已经有足够的知识去评估和驾驭框架，而不是被框架牵着鼻子走。

手写一个 AI Agent 的过程，就像教一个聪明的孩子如何使用一套复杂的工具箱。你需要清晰地定义每件工具（函数）的用途和用法，设计一套有效的沟通机制（提示词和消息历史），并建立一套行动规则（循环与停止条件）。这个过程充满挑战，但当你看到它自动完成一连串任务时，成就感也是巨大的。希望这篇从 Function Calling 到任务链的实践指南，能为你点亮第一盏灯。剩下的路，就是在不断的调试、迭代和扩展中，让你的 Agent 变得越来越聪明和可靠。