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)。它模仿了人类解决问题的方式推理根据当前目标和已有信息思考下一步该做什么。行动执行思考后决定的操作比如调用一个函数。观察获取行动的结果并将其作为新的信息输入下一轮推理。这个循环会一直持续直到任务完成或无法继续。在我们的天气-邮件提醒场景里Agent 的内心活动可能是这样的推理1“用户想要明天的天气和提醒。我需要先获取天气数据。有一个get_weather函数可用。”行动1调用get_weather(location北京, datetomorrow)。观察1函数返回{“weather”: “rain”, “temperature”: “18°C”}。推理2“天气是下雨。用户的条件是‘如果下雨就提醒’。所以我需要生成提醒内容并发送邮件。我有send_email函数。”行动2调用send_email(subject天气提醒, body明天北京下雨记得带伞。, touserexample.com)。观察2函数返回{“status”: “success”}。推理3“邮件发送成功所有用户要求都已满足。任务完成。”我们的代码架构就是要支撑起这个循环。2.2 组件拆解四大核心模块为了实现上述循环我们需要设计几个关键模块工具层这是 Agent 的“手”和“脚”。每个工具对应一个 Python 函数并附带一个清晰的描述告诉 LLM 这个工具是干什么的、需要什么参数。例如get_weather工具的描述会说明它用于查询指定地点和日期的天气。推理引擎这是 Agent 的“大脑”通常就是 LLM 本身如 GPT-4, Claude, 或本地部署的模型。它的核心职责是理解用户请求和当前对话历史决定下一步是“直接回答”还是“调用某个工具”。如果决定调用工具它还需要根据工具描述生成符合要求的参数。函数调用处理器这是“神经中枢”负责连接大脑和手脚。它接收 LLM 发出的“调用工具X参数为Y”的指令在代码中找到对应的工具函数安全地执行它并将执行结果格式化准备送回给 LLM 进行下一轮推理。状态管理与任务链调度这是“工作记忆”和“项目经理”。它需要维护整个对话的历史包括用户消息、AI回复、工具调用及结果确保上下文不丢失。更重要的是它要驱动整个 ReAct 循环判断任务是否结束防止陷入无限循环。一个常见的架构误区是试图用一个超级复杂的类来搞定一切。我的经验是初期务必保持模块的轻量和清晰。下面我们就从最基础的 Function Calling 实现开始。3. 基础实现让 LLM 学会“用手”Function CallingFunction 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_keyapi_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( modelself.model, messagesself.conversation_history, toolsTOOLS, # 传入我们定义的工具列表 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( modelself.model, messagesself.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。记得带伞哦。实操心得1tool_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_steps10): self.client openai.OpenAI(api_keyapi_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( modelself.model, messagesself.messages, toolsTOOLS, 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(“先查一下北京明天的天气如果是雨天就搜索‘雨天出行注意事项’然后把注意事项总结成一句话发到我的邮箱 testexample.com。”) print(“\n 最终输出 ) print(result)这个TaskChainAgent已经具备了自动化任务链的核心能力。它会自动执行“查询天气 - 判断 - 搜索 - 发送邮件”这一系列操作。4.2 任务链的停止条件与稳定性上面的循环有一个简单的停止条件LLM 不再输出工具调用tool_calls为空而是直接给出自然语言回答。但这并不总是可靠的。常见问题1LLM 陷入循环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_tokens8000): 当历史消息预估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) 工具API3) LLM的“不可预测”的输出。6.1 建立可观测性给你的 Agent 加上详细的日志记录每一步的输入输出。import logging logging.basicConfig(levellogging.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_valuemock_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是字符串它却生成一个数字或复杂对象。调用不存在的工具自己编造一个工具名。在应该给出最终答案时又调用工具。应对策略前置参数校验在_execute_tool中在执行前先用 JSON Schema 验证参数。不通过则直接返回错误给LLM。后置输出解析对LLM的直接回复非工具调用也可以进行结构化解析确保其格式符合预期。优化系统提示这是最重要的。在系统提示中明确规则“你必须严格按照提供的工具描述来调用。如果用户请求无法用现有工具完成请直接告知用户不要编造工具。”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“userexample.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 变得越来越聪明和可靠。