最近在尝试将大模型能力集成到实际业务中时发现单纯调用API生成文本远不能满足复杂需求。无论是处理多步骤任务、调用外部工具还是协调多个“智能体”分工合作都涉及到一个核心概念Agent智能体及其背后的Skill技能、Tool工具和Workflow工作流。网上资料要么过于学术化要么零散不成体系让开发者难以落地。本文旨在为你提供一份从零到一的Agent实战指南。我们将从核心概念讲起逐步深入到如何设计Skill、集成Tool、编排Workflow并最终构建一个可运行的Multi-Agent系统。无论你是想了解AI应用开发的前沿趋势还是急需一个能解决实际问题的智能体方案这篇文章都将提供清晰的路径和可直接复用的代码。1. 智能体Agent与技能Skill核心概念解析在深入代码之前我们必须厘清几个关键概念。这些术语在社区中有时混用但理解其差异对架构设计至关重要。1.1 什么是智能体Agent你可以将Agent理解为一个具备一定自主性的“虚拟员工”。它不仅仅是一个问答模型而是一个系统其核心能力包括感知Perception接收来自用户、环境或其他Agent的输入文本、数据、事件。规划Planning根据目标拆解任务步骤形成执行计划。行动Action执行具体操作例如调用一个函数、访问数据库、生成代码。反思Reflection评估行动结果并根据反馈调整后续计划。一个最简单的Agent可以是一个封装了大模型调用并预设了提示词Prompt的程序。而一个复杂的Agent则可能拥有长期记忆、工具使用能力和多轮对话策略。1.2 技能Skill、工具Tool与工作流Workflow的关系这是最容易混淆的一组概念。我们可以用一个比喻来理解工具Tool就像螺丝刀、扳手。它是一个具体的、原子性的功能函数例如search_web(keywords)calculate(expression)send_email(to, content)。它只负责“做”不负责“何时做”和“为何做”。技能Skill就像“修理自行车”这个能力。它代表Agent完成某一类特定任务Task的能力。一个Skill内部封装了使用一个或多个Tool的逻辑以及引导大模型正确调用这些Tool的提示词Prompt和解析逻辑。例如“天气查询Skill”封装了get_location_from_query和call_weather_api两个Tool的使用方法。工作流Workflow就像“早上上班”的流程起床→洗漱→早餐→通勤。它定义了多个Skill或Agent执行的顺序和条件是一个更高层次的编排Orchestration。Workflow负责协调确保任务按正确流程进行可能涉及条件分支、循环和并行执行。核心关系Agent拥有has一个或多个 Skill。Skill利用uses一个或多个 Tool 来完成任务。复杂的任务可以通过 Workflow 来编排orchestrates多个 Skill 或 Agent 协同工作。1.3 多智能体Multi-Agent系统简介当单个Agent无法处理过于复杂的任务时就需要Multi-Agent系统。在这种架构下不同Agent被赋予特定角色如“产品经理”、“程序员”、“测试员”它们通过通信机制如共享消息总线、直接调用进行协作共同完成一个宏观目标。Multi-Agent系统的优势在于角色分工、知识隔离和解决复杂问题但同时也带来了通信开销和协调复杂度的挑战。2. 环境准备与开发框架选择工欲善其事必先利其器。构建Agent应用选择一个合适的开发框架可以事半功倍。这里我们以Python生态为主进行介绍。2.1 基础环境配置首先确保你的开发环境已就绪Python: 推荐使用 Python 3.9 或以上版本。包管理工具: 使用pip或更推荐的poetry/conda来管理依赖。大模型访问权限: 你需要一个大型语言模型LLM的API密钥。本文示例将使用OpenAI GPT系列模型但你也可以替换为其他兼容OpenAI API的模型如国内大模型或本地部署模型。通过以下命令检查Python环境并安装基础包# 检查Python版本 python --version # 创建项目目录并进入 mkdir ai-agent-tutorial cd ai-agent-tutorial # 创建虚拟环境可选但推荐 python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 安装核心依赖 pip install openai2.2 主流Agent开发框架概览目前社区有多个优秀的Agent框架它们在不同抽象层次上提供了支持LangChain / LangGraph: 这是目前最流行的生态系统。LangChain提供了构建链Chain和Agent的基础模块Tool, Memory, Prompt等。LangGraph是其上用于构建复杂、有状态的多步骤工作流即Workflow和Multi-Agent系统的库。它非常全面但学习曲线稍陡。LlamaIndex: 最初专注于检索增强生成RAG现在也提供了强大的Agent和Workflow构建能力尤其在数据查询和知识处理方面有优势。AutoGen (by Microsoft): 专注于多智能体对话。它允许你定义多个可对话的Agent并通过自动或人工反馈来协调它们完成任务非常适合研究型场景和复杂对话任务。Semantic Kernel (by Microsoft): 一个将传统代码技能与AI模型“插件”结合的开源SDK强调将AI能力作为“插件”集成到现有应用中。简易自研: 对于理解原理或简单需求完全可以基于openaiSDK 和 Python 函数自己封装这有助于深入理解Agent的运行机制。本文选择为了兼顾实用性、流行度和教学清晰度我们将主要使用LangChain来构建核心的Agent和Skill并在需要时提及其他框架作为对比或扩展选择。它的设计哲学与我们的核心概念Tool, Agent, Workflow高度吻合。安装LangChainpip install langchain langchain-openai langchain-community # langchain-openai 是官方维护的OpenAI集成 # langchain-community 包含大量社区贡献的Tool和组件3. 从零构建第一个智能体工具Tool的定义与使用让我们从最基础的单元——Tool开始。一个Tool本质上是一个可以被大模型调用的函数。3.1 如何定义一个Tool在LangChain中有多种方式定义Tool。最直接的方式是使用tool装饰器。这个装饰器会自动生成描述帮助LLM理解何时以及如何调用这个工具。假设我们要构建一个“计算器”Agent首先给它一个计算乘法的工具# 文件tools/calculator_tools.py from langchain.tools import tool import math tool def multiply(a: float, b: float) - float: Multiply two numbers together. Args: a: First number. b: Second number. Returns: The product of a and b. return a * b tool def get_square_root(number: float) - float: Calculate the square root of a non-negative number. Args: number: A non-negative number. Returns: The square root. if number 0: return Error: Cannot calculate square root of a negative number. return math.sqrt(number) # 你可以继续定义 add, subtract, divide 等工具关键点函数文档字符串Docstring至关重要LLM主要依靠这个描述来决定是否以及如何调用该工具。描述应清晰说明功能、参数和返回值。类型提示Type Hints如a: float这有助于框架进行验证和LLM理解。tool装饰器它封装了函数使其成为LangChain可识别的Tool对象。3.2 创建Agent并赋予其使用Tool的能力有了Tool我们需要创建一个Agent来使用它。在LangChain中一个常见的模式是使用create_react_agent。ReActReasoning Acting是一个让LLM在思考生成推理轨迹和行动调用工具之间交替的经典框架。# 文件agent/basic_calculator_agent.py from langchain import hub from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI from tools.calculator_tools import multiply, get_square_root # 1. 初始化大语言模型LLM # 请将 YOUR_OPENAI_API_KEY 替换为你的实际密钥或通过环境变量设置 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyYOUR_OPENAI_API_KEY) # 2. 准备工具列表 tools [multiply, get_square_root] # 3. 获取一个预设的ReAct提示词模板 # LangChain Hub 是一个提示词模板库我们拉取一个标准的ReAct模板 prompt hub.pull(hwchase17/react) # 4. 创建ReAct Agent agent create_react_agent(llm, tools, prompt) # 5. 创建Agent执行器负责运行Agent处理工具调用循环 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 6. 运行Agent if __name__ __main__: # 示例问题1简单计算 result1 agent_executor.invoke({input: 请问123乘以456等于多少}) print(结果1:, result1[output]) # 示例问题2需要多步推理和工具调用 result2 agent_executor.invoke({input: 先计算16的平方根再将结果乘以5。}) print(结果2:, result2[output]) # 示例问题3模型无法直接回答但工具可以 result3 agent_executor.invoke({input: 今天的天气怎么样}) print(结果3:, result3[output]) # 预期输出它应该承认自己没有相关工具。运行与观察 设置verboseTrue后控制台会输出详细的思考过程你会看到类似下面的日志 Entering new AgentExecutor chain... 我需要计算123乘以456。我有一个乘法工具可以使用。 Action: multiply Action Input: {a: 123, b: 456} Observation: 56088 Thought: 我得到了结果56088。 Final Answer: 123乘以456等于56088。 Finished chain. 结果1: 123乘以456等于56088。这个过程清晰地展示了ReAct框架的“思考-行动-观察”循环。4. 设计高级技能Skill封装复杂任务逻辑Skill是比单一Tool更高级的抽象。例如一个“数据可视化Skill”可能包含获取数据、清洗数据、选择图表类型、调用绘图库等多个步骤和Tool。在LangChain中我们可以通过构建复杂的链Chain或使用智能体Agent本身来实现一个Skill。4.1 实现一个“天气查询”Skill这个Skill需要两个核心能力1) 从用户自然语言中提取地理位置2) 调用天气API。我们将创建两个Tool然后用一个Agent将它们组织成一个Skill。首先安装可能需要的额外包pip install requests然后创建工具和Skill# 文件skills/weather_skill.py from langchain.tools import tool from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI from langchain import hub import requests import os # --- 工具1地理位置提取模拟实际可使用NER API--- tool def extract_location(query: str) - str: Extract the location name from a users query about weather. Args: query: Users input, e.g., Whats the weather in Beijing? Returns: The extracted location name, e.g., Beijing. # 这是一个简化版。实际应用中你可以用另一个LLM调用或专门的NER服务来更精准地提取。 # 这里我们简单假设查询格式是“... in [Location]”或“[Location]的天气” import re # 简单匹配模式 patterns [ rweather in (\w), r(\w)的天气, r(\w) weather ] for pattern in patterns: match re.search(pattern, query, re.IGNORECASE) if match: return match.group(1) # 如果没匹配到返回一个默认值或原查询在实际应用中需要更健壮的处理 return query.split()[-1] # 非常简单的回退策略 # --- 工具2调用真实天气API --- tool def get_weather_by_city(city: str) - str: Get current weather information for a given city. Args: city: Name of the city. Returns: A string describing the weather. # 示例使用 Open-Meteo 免费API (无需密钥) # 注意这个API需要经纬度这里做了极大简化。生产环境应使用地理编码服务将城市名转为坐标。 try: # 这里仅为演示直接使用一个静态URL。真实情况需要先获取城市坐标。 # 假设我们已知道北京的大致坐标 if beijing in city.lower() or 北京 in city: url https://api.open-meteo.com/v1/forecast?latitude39.9042longitude116.4074current_weathertrue elif shanghai in city.lower() or 上海 in city: url https://api.open-meteo.com/v1/forecast?latitude31.2304longitude121.4737current_weathertrue else: return f抱歉目前仅支持预设城市的天气查询。 response requests.get(url) data response.json() current data.get(current_weather, {}) temp current.get(temperature) wind_speed current.get(windspeed) weather_code current.get(weathercode) # 简化天气代码转换 weather_map {0: 晴, 1: 晴, 2: 多云, 3: 阴天, 45: 雾, 61: 小雨} condition weather_map.get(weather_code, 未知) return f{city}的当前天气{condition}温度 {temp}°C风速 {wind_speed} km/h。 except Exception as e: return f获取天气信息时出错{e} # --- 创建天气查询Skill本质上是一个专用的Agent--- def create_weather_skill(llm): Factory function to create a weather inquiry skill. weather_tools [extract_location, get_weather_by_city] prompt hub.pull(hwchase17/react) agent create_react_agent(llm, weather_tools, prompt) executor AgentExecutor(agentagent, toolsweather_tools, verboseFalse, handle_parsing_errorsTrue) return executor # --- 主程序使用这个Skill --- if __name__ __main__: llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) weather_skill create_weather_skill(llm) # 测试Skill queries [ 北京天气怎么样, Whats the weather in Shanghai?, 告诉我旧金山的天气 # 这个会触发我们的预设城市限制 ] for query in queries: print(f\n用户问{query}) result weather_skill.invoke({input: query}) print(fSkill答{result[output]})这个weather_skill就是一个封装好的Skill。它对外提供一个简单的invoke接口内部则协调extract_location和get_weather_by_city两个Tool来完成任务。你可以将这个Skill作为一个模块提供给更上层的Agent或Workflow使用。5. 编排工作流Workflow串联多个技能与智能体当任务涉及多个步骤且步骤之间有明确的顺序、条件或循环关系时就需要Workflow。LangChain提供了LangGraph来构建有状态、可循环的图工作流。5.1 使用LangGraph构建一个审批工作流假设我们有一个“内容生成与审核”工作流1) 生成一篇草稿2) 检查草稿是否包含敏感词3) 如果通过则格式化输出如果不通过则重新生成。首先安装LangGraphpip install langgraph然后实现这个工作流# 文件workflows/content_approval_workflow.py from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage import os # 1. 定义工作流的状态State # State是一个字典包含工作流执行过程中所有节点共享的数据 class WorkflowState(TypedDict): # 用户原始输入 topic: str # 生成的草稿 draft: str # 敏感词检查结果 has_sensitive_word: bool # 最终输出 final_output: str # 循环计数器防止无限循环 loop_count: Annotated[int, operator.add] # 这是一个“加法注释”用于在图中自动累加 # 2. 定义各个节点Node函数 # 节点即工作流中的步骤每个节点接收State修改后返回更新后的State def generate_draft(state: WorkflowState) - WorkflowState: 节点1根据主题生成草稿。 print(f[节点-生成草稿] 正在为主题‘{state[topic]}’生成草稿...) llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7, openai_api_keyos.getenv(OPENAI_API_KEY)) prompt f请围绕以下主题写一段简短的介绍100字以内{state[topic]} response llm.invoke([HumanMessage(contentprompt)]) draft response.content state[draft] draft state[loop_count] 1 # 初始化计数器 print(f[节点-生成草稿] 草稿生成完毕。) return state def check_sensitive_words(state: WorkflowState) - WorkflowState: 节点2检查草稿中是否包含敏感词。 print(f[节点-敏感词检查] 正在检查草稿...) # 这里使用一个简单的敏感词列表进行演示 sensitive_words [暴力, 仇恨, 机密] draft state[draft] found any(word in draft for word in sensitive_words) state[has_sensitive_word] found if found: print(f[节点-敏感词检查] 发现敏感词) else: print(f[节点-敏感词检查] 通过。) return state def format_output(state: WorkflowState) - WorkflowState: 节点3格式化最终输出审核通过时执行。 print(f[节点-格式化输出] 正在格式化...) draft state[draft] state[final_output] f【审核通过】\n{draft} return state def regenerate_draft(state: WorkflowState) - WorkflowState: 节点4重新生成草稿审核不通过时执行。 loop_count state[loop_count] print(f[节点-重新生成] 第{loop_count}次尝试重新生成草稿...) # 简单修改主题加入“健康积极”的引导 new_topic state[topic] 请以健康积极的方式描述 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7, openai_api_keyos.getenv(OPENAI_API_KEY)) response llm.invoke([HumanMessage(contentf请重新写一段关于{new_topic}的短文避免敏感内容。)]) state[draft] response.content state[loop_count] loop_count 1 # 计数器加1 state[has_sensitive_word] False # 重置检查标志 print(f[节点-重新生成] 重新生成完成。) return state # 3. 定义条件边Conditional Edge的逻辑 def should_continue(state: WorkflowState) - Literal[approved, rejected, too_many_retries]: 根据检查结果决定下一步走向。 if state[has_sensitive_word]: if state[loop_count] 3: # 最多重试3次 return too_many_retries else: return rejected else: return approved # 4. 构建工作流图Graph def create_content_approval_workflow(): # 初始化图 workflow StateGraph(WorkflowState) # 添加节点 workflow.add_node(generate, generate_draft) workflow.add_node(check, check_sensitive_words) workflow.add_node(format, format_output) workflow.add_node(regenerate, regenerate_draft) # 设置入口点 workflow.set_entry_point(generate) # 添加边连接节点 workflow.add_edge(generate, check) # 生成后必然检查 workflow.add_edge(format, END) # 格式化后结束 # 添加条件边根据检查结果分流 workflow.add_conditional_edges( check, should_continue, # 条件判断函数 { approved: format, # 通过 - 格式化 rejected: regenerate, # 不通过 - 重新生成 too_many_retries: END # 重试过多 - 直接结束失败 } ) workflow.add_edge(regenerate, check) # 重新生成后回到检查节点 # 编译图得到可执行的工作流 app workflow.compile() return app # 5. 运行工作流 if __name__ __main__: # 创建并编译工作流 app create_content_approval_workflow() # 定义初始状态 initial_state: WorkflowState { topic: 城市公园的日常, draft: , has_sensitive_word: False, final_output: , loop_count: 0 } print( 开始执行内容审批工作流 ) # 运行工作流 final_state app.invoke(initial_state) print(\n 工作流执行结束 ) print(f最终输出{final_state.get(final_output, 无输出可能因重试次数过多而终止)}) print(f总循环次数{final_state[loop_count]})这个示例展示了LangGraph的核心能力定义状态、创建节点、通过条件边控制流程。你可以清晰地看到工作流如何根据check_sensitive_words节点的结果决定是走向format_output还是regenerate_draft并可能形成循环。6. 构建多智能体Multi-Agent协作系统Multi-Agent系统由多个具备不同角色和能力的Agent组成它们通过通信协作解决问题。我们来实现一个简单的“软件开发团队”模拟一个ProductManagerAgent负责分析需求一个DeveloperAgent负责写代码一个TesterAgent负责检查代码。6.1 定义角色与通信我们将使用langgraph来协调多个Agent。每个Agent都是一个独立的AgentExecutor拥有自己的工具和提示词。它们通过共享的“消息列表”进行通信。# 文件multi_agent/software_team.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI from langchain import hub from langchain_core.messages import BaseMessage, HumanMessage, AIMessage import os # 定义共享状态主要是一个消息列表 class TeamState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] # “加法注释”用于自动追加消息 final_code: str # --- 定义三个角色的Agent --- def create_product_manager_agent(llm): 产品经理Agent分析需求输出用户故事或功能点。 from langchain.tools import tool tool def analyze_requirement(requirement: str) - str: Analyze a software requirement and break it down into user stories or features. # 在实际应用中这里可以调用LLM进行分析。这里简化为固定响应。 return f需求‘{requirement}’已分析。主要功能点1. 用户输入验证 2. 核心计算逻辑 3. 结果展示。 tools [analyze_requirement] prompt hub.pull(hwchase17/react) agent create_react_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools, verboseFalse, handle_parsing_errorsTrue, max_iterations2) return executor def create_developer_agent(llm): 开发人员Agent根据功能点编写Python代码。 from langchain.tools import tool tool def write_python_code(spec: str) - str: Write Python code based on the given specification. # 简化为调用LLM生成代码 code_prompt f请根据以下功能描述编写一个简单的Python函数。只返回代码块不要解释。\n功能描述{spec} response llm.invoke([HumanMessage(contentcode_prompt)]) return response.content tools [write_python_code] prompt hub.pull(hwchase17/react) agent create_react_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools, verboseFalse, handle_parsing_errorsTrue, max_iterations2) return executor def create_tester_agent(llm): 测试人员Agent检查代码是否存在明显问题。 from langchain.tools import tool tool def review_code(code: str) - str: Review the provided Python code for obvious syntax errors or issues. # 简化为固定检查逻辑 issues [] if print( not in code: issues.append(代码似乎没有输出语句不利于调试。) if def not in code: issues.append(未发现函数定义可能不是完整的可运行单元。) if not issues: return 代码初步检查通过未发现明显语法问题。 else: return 检查发现以下问题 ; .join(issues) tools [review_code] prompt hub.pull(hwchase17/react) agent create_react_agent(llm, tools, prompt) executor AgentExecutor(agentagent, toolstools, verboseFalse, handle_parsing_errorsTrue, max_iterations2) return executor # --- 定义各个节点每个节点代表一个Agent的工作 --- def pm_node(state: TeamState): 产品经理节点分析初始需求。 print(\n[产品经理] 开始工作...) llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) pm_agent create_product_manager_agent(llm) # 从消息历史中提取最新的用户需求假设第一条HumanMessage是需求 last_human_msg next((msg for msg in state[messages] if isinstance(msg, HumanMessage)), None) requirement last_human_msg.content if last_human_msg else 开发一个计算器程序 result pm_agent.invoke({input: f请分析这个需求{requirement}}) analysis result[output] new_msg AIMessage(contentf产品经理分析结果{analysis}) state[messages].append(new_msg) print(f[产品经理] 分析完成。) return state def developer_node(state: TeamState): 开发人员节点根据产品经理的分析写代码。 print(\n[开发人员] 开始工作...) llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) dev_agent create_developer_agent(llm) # 获取产品经理的分析结果最后一条AIMessage last_ai_msg next((msg for msg in reversed(state[messages]) if isinstance(msg, AIMessage)), None) spec last_ai_msg.content if last_ai_msg else 无详细需求 result dev_agent.invoke({input: f请根据以下分析编写Python代码{spec}}) code result[output] new_msg AIMessage(contentf开发人员编写代码\npython\n{code}\n) state[messages].append(new_msg) state[final_code] code # 保存到最终状态 print(f[开发人员] 代码编写完成。) return state def tester_node(state: TeamState): 测试人员节点审查代码。 print(\n[测试人员] 开始工作...) llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, openai_api_keyos.getenv(OPENAI_API_KEY)) tester_agent create_tester_agent(llm) code state.get(final_code, ) if not code: code 未找到代码。 result tester_agent.invoke({input: f请审查以下代码{code}}) review result[output] new_msg AIMessage(contentf测试人员审查意见{review}) state[messages].append(new_msg) print(f[测试人员] 审查完成。) return state def should_continue(state: TeamState) - str: 决定工作流是否继续。这里我们设计为线性执行PM - Dev - Tester - 结束。 # 一个简单的实现根据消息数量判断阶段 # 更复杂的实现可以基于消息内容决定是否循环例如测试不通过则打回给开发 msg_count len(state[messages]) # 初始消息(用户需求) PM消息 Dev消息 Tester消息 if msg_count 4: # 0:初始, 1:PM, 2:Dev, 3:Tester # 决定下一个节点是谁 if msg_count 1: return to_developer elif msg_count 2: return to_tester else: return to_end else: return to_end # --- 构建并运行Multi-Agent工作流 --- def create_software_team_workflow(): workflow StateGraph(TeamState) workflow.add_node(product_manager, pm_node) workflow.add_node(developer, developer_node) workflow.add_node(tester, tester_node) workflow.set_entry_point(product_manager) # 使用条件边来控制流程 workflow.add_conditional_edges( product_manager, lambda s: to_developer, # PM完成后总是去Developer ) workflow.add_conditional_edges( developer, lambda s: to_tester, # Developer完成后总是去Tester ) workflow.add_conditional_edges( tester, should_continue, # Tester完成后根据逻辑决定是否结束 { to_end: END, # 如果需要打回重做可以添加 to_developer 路径 } ) app workflow.compile() return app if __name__ __main__: app create_software_team_workflow() # 初始化状态用户提出需求 initial_state: TeamState { messages: [HumanMessage(content请开发一个能计算斐波那契数列的函数。)], final_code: } print( 启动软件开发团队Multi-Agent) final_state app.invoke(initial_state) print(\n 协作过程记录 ) for i, msg in enumerate(final_state[messages]): role type(msg).__name__.replace(Message, ) print(f{i}. [{role}] {msg.content[:100]}...) # 打印前100字符 print(f\n 最终生成的代码 ) print(final_state.get(final_code, 无代码产出))这个例子展示了一个简单的、线性的Multi-Agent协作流程。在实际系统中你可以设计更复杂的交互模式例如让Tester和Developer之间进行多轮对话或者引入一个CoordinatorAgent来动态分配任务。7. 常见问题、调试与最佳实践构建Agent系统时你会遇到一些典型问题。以下是一些排查思路和工程建议。7.1 常见问题与解决方案问题现象可能原因排查思路与解决方案Agent不调用工具1. Tool描述不清晰。2. LLM温度temperature过高导致输出随机。3. 提示词Prompt未优化。1. 检查Tool的docstring确保清晰描述了功能、参数和用途。2. 将temperature设为0或较低值如0.1使输出更确定。3. 使用verboseTrue查看Agent的思考链观察它是否在正确解析任务。4. 尝试更强大的模型如gpt-4。工具调用参数错误1. LLM未能正确解析用户输入以匹配工具参数。2. 参数类型不匹配。1. 在Tool函数内部添加更严格的参数验证和错误处理。2. 考虑在Agent前添加一个“参数提取”步骤或用Pydantic工具强制类型。3. 使用handle_parsing_errorsTrue让Agent在解析失败时有机会重试。工作流陷入无限循环1. 条件边Conditional Edge逻辑有误。2. 状态更新不正确。1. 在状态中添加循环计数器如loop_count并在条件边中检查上限。2. 使用verboseTrue运行工作流打印每个节点的状态变化。3. 简化工作流逐步添加复杂性进行测试。多智能体通信混乱1. 消息格式不统一。2. 没有清晰的会话回合控制。1. 使用langchain_core.messages中的标准消息类HumanMessage,AIMessage,SystemMessage。2. 为每个Agent设计明确的角色提示词System Prompt例如“你是一个严谨的测试工程师...”。3. 在状态中维护清晰的消息历史并让每个Agent只处理最新的或相关的消息。性能与成本问题1. 每次调用都初始化新的LLM实例。2. 工具调用或工作流步骤过多。3. 使用GPT-4等昂贵模型。1. 复用LLM实例在全局或应用级别初始化一次。2. 优化工作流合并不必要的步骤或使用更轻量的模型处理简单任务。3. 对于简单任务优先使用gpt-3.5-turbo。考虑对工具调用结果进行缓存。7.2 工程化与最佳实践模块化设计将Tool、Skill、Agent、Workflow分别放在不同的模块或文件中。例如tools/、skills/、agents/、workflows/。这有利于代码复用和测试。配置化管理将模型类型、API密钥、温度等参数提取到配置文件如config.yaml或环境变量中。避免在代码中硬编码。完善的日志记录除了使用LangChain的verbose模式还应集成应用级别的日志库如logging记录关键决策点、工具调用输入输出和错误信息便于线上排查。超时与重试机制为LLM调用和外部API调用Tool添加超时和重试逻辑提高系统鲁棒性。验证与测试单元测试为每个Tool函数编写单元测试。集成测试测试完整的Skill或Agent使用Mock对象替代真实的LLM和外部API确保逻辑正确。端到端测试对关键Workflow进行少量真实的端到端测试。提示词工程精心设计System Prompt和Few-Shot示例这是引导Agent行为的关键。将提示词模板化存放在单独的文件或数据库中。安全性工具权限对Tool进行权限分级避免Agent调用危险操作如删除文件、执行任意命令。输入过滤对用户输入和Tool的输入输出进行必要的清洗和过滤防止注入攻击。敏感信息切勿将API密钥、数据库密码等硬编码在代码或提示词中。评估与监控在生产环境中需要建立评估体系监控Agent的响应质量、工具调用成功率、延迟和成本。这有助于持续迭代优化。从理解Agent、Skill、Tool、Workflow的核心概念开始我们一步步实现了基础工具调用、复杂技能封装、有条件的工作流编排最终构建了一个多智能体协作系统。这个过程中LangChain和LangGraph提供了强大的抽象但背后的设计思想——将复杂任务分解为可规划、可执行、可反思的步骤——是通用的。真正的挑战在于将这套技术应用于具体的业务场景。建议从一个明确的、边界清晰的小问题开始如“自动周报生成器”、“智能客服路由”优先实现核心链路再逐步增加复杂性。同时密切关注LangChain等框架的更新其社区正在快速发展不断有新的模式和最佳实践涌现。