AI Agent开发实战:从Tool到Workflow的完整构建指南
1. 先搞清楚 Agent、Skill、Workflow 到底在解决什么问题如果你刚接触 AI Agent 开发看到一堆 Multi-Agent、Tool、Workflow、Skill 这些词很容易被绕晕。它们听起来很高级但核心要解决的问题其实很具体如何让一个或多个大模型像人一样通过调用工具、执行步骤去完成一个复杂的任务。举个例子你想让 AI 帮你分析一份财报然后生成一份摘要报告最后发邮件给指定的人。这个任务单靠一个“问答”式的大模型很难一步到位。你需要拆解先调用“文件读取”工具再调用“数据分析”工具然后调用“文本生成”模型最后调用“邮件发送”接口。这个“拆解任务、调度工具、按序执行”的过程就是 Agent 架构要处理的核心。所以这篇文章不是讲空洞的理论而是帮你把 Agent 开发落地。我会围绕几个关键点展开Agent任务的执行者和决策者它知道自己能干什么Skill需要什么Tool。SkillAgent 具备的“技能”比如“数据查询”、“文本总结”。一个 Skill 可能由多个步骤Workflow组成。Tool最底层的“工具”是 Skill 和 Workflow 调用的具体功能比如一个 Python 函数、一个 API 接口。Workflow定义任务执行的“流程图”或“剧本”明确先做什么、后做什么、失败了怎么办。Multi-Agent当任务太复杂一个 Agent 搞不定时就需要多个各司其职的 Agent 协作。最值得你关注的不是这些概念本身而是如何在你自己的开发环境里把这些组件串起来跑通一个从想法到可运行代码的完整流程。很多人卡在第一步环境配不对或者不知道从哪里开始写第一行代码。2. 环境准备避开依赖冲突和路径坑在开始写任何 Agent 代码之前环境是第一个拦路虎。网上教程经常直接pip install一堆包但版本冲突、CUDA 问题、权限错误能卡住你半天。我的建议是先建立一个干净、可复现的环境。2.1 Python 环境与核心依赖不要用系统自带的 Python。用 Conda 或 venv 创建一个独立的虚拟环境。这里以 Conda 为例因为它管理不同 Python 版本和包更方便。# 创建一个名为 ai_agent 的虚拟环境指定 Python 3.10这是一个比较稳定的版本 conda create -n ai_agent python3.10 -y conda activate ai_agent接下来安装核心依赖。Agent 开发框架很多比如 LangChain、AutoGen、CrewAI 等。为了演示最通用的模式我们以LangChain为核心因为它生态丰富概念清晰适合理解底层原理。# 安装 LangChain 及其常用扩展 pip install langchain langchain-community langchain-core # 安装 OpenAI 的 SDK如果你使用 OpenAI 的模型作为 Agent 的“大脑” pip install openai # 安装用于定义工作流的框架例如 LangGraphLangChain 官方的工作流/多Agent框架 pip install langgraph为什么是这些包langchain: 提供 Agent、Chain、Tool 等核心抽象。langchain-community: 包含大量第三方 Tool 和集成。langgraph: 用于构建有状态、多步骤的 Workflow 和 Multi-Agent 系统比简单的 Chain 更强大。openai: 提供一个 LLM大语言模型后端作为 Agent 的推理引擎。2.2 模型访问API 与本地部署的权衡你的 Agent 需要一个“大脑”也就是大模型。有两种主要方式调用云端 API推荐初学者简单无需考虑显存和算力。你需要一个 API Key例如来自 OpenAI、DeepSeek、智谱AI等。将 Key 设置为环境变量# 在 Linux/macOS 的终端或 Windows 的 PowerShell 中设置 export OPENAI_API_KEYyour-api-key-here在代码中这样初始化一个 LLMfrom langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, api_keyos.getenv(OPENAI_API_KEY))本地部署大模型适合深度定制、数据隐私要求高这涉及下载模型文件、配置推理框架如 Ollama、vLLM、Transformers。这对机器资源尤其是 GPU 显存有要求。简易方案使用Ollama。它简化了本地模型的拉取和运行。# 安装 Ollama (请参考官网) # 拉取一个轻量级模型如 Llama 3.2 ollama pull llama3.2:latest在代码中连接from langchain_community.llms import Ollama llm Ollama(modelllama3.2:latest)关键判断如果你的任务不涉及敏感数据且追求开发效率先用 API。等核心逻辑跑通后再考虑为性能或隐私优化而转向本地模型。2.3 项目目录结构一个清晰的目录结构能避免后续的混乱。建议这样组织your_agent_project/ ├── agents/ # 存放不同 Agent 的类定义 ├── skills/ # 存放 Skill 的实现 ├── tools/ # 存放最基础的 Tool函数或类 ├── workflows/ # 存放 Workflow 的配置或定义文件 ├── config/ # 配置文件API Key、模型路径等 ├── data/ # 存放输入输出数据 ├── logs/ # 存放运行日志 └── main.py # 主入口文件现在环境准备好了我们可以开始构建第一个基础组件Tool。3. 从 Tool 到 Skill构建可复用的基础能力Tool 是 Agent 世界里的“螺丝刀”和“万用表”是最原子的操作单元。Skill 则是完成一个具体目标所需的“一套组合工具和流程”。3.1 创建你的第一个 Tool一个 Tool 本质上是一个能被 LLM 理解和调用的函数。在 LangChain 中用tool装饰器可以轻松创建。假设我们要做一个能查询天气的 Agent。首先创建一个tools/weather_tool.py# tools/weather_tool.py from langchain.tools import tool import requests tool def get_weather(city: str) - str: 根据城市名称查询当前天气。输入必须是城市名如‘北京’、‘Shanghai’。 # 这里使用一个模拟的天气API。实际项目中请替换为真实API如和风天气、OpenWeatherMap等。 # 注意处理API密钥等敏感信息时应从配置文件读取不要硬编码。 try: # 模拟API调用 # response requests.get(fhttps://api.weatherapi.com/v1/current.json?keyYOUR_KEYq{city}) # data response.json() # return f{city}的天气是{data[current][condition][text]}温度{data[current][temp_c]}摄氏度。 # 为了演示返回模拟数据 return f{city}的天气是晴朗温度25摄氏度。 except Exception as e: return f查询{city}天气时出错{str(e)} # 你可以创建更多工具 tool def get_current_time() - str: 获取当前的日期和时间。 from datetime import datetime now datetime.now() return now.strftime(%Y-%m-%d %H:%M:%S)关键点函数文档字符串 (docstring)这极其重要LLM 完全依靠这个描述来理解这个工具是干什么的、输入什么、输出什么。描述要清晰、准确。输入参数明确类型如str。LLM 会尝试生成符合类型的参数。错误处理Tool 内部要有try...except返回友好的错误信息避免整个 Agent 因为一个 Tool 崩溃。3.2 将多个 Tool 组合成一个 SkillSkill 是更高层次的抽象。例如“出行建议”这个 Skill可能需要先后调用“查询天气”、“查询交通”、“查询日历”等多个 Tool并有一定的逻辑。在 LangChain 中你可以用一个Chain或者一个更复杂的RunnableSequence来初步实现一个 Skill。但更符合 Skill 概念的是使用LangGraph来构建一个有状态、可分支的工作流。我们先看一个简单的 Chain 式 Skill在skills/travel_advice_skill.py中# skills/travel_advice_skill.py from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from .tools.weather_tool import get_weather, get_current_time # 假设tools在上一级目录 def create_travel_advice_skill(llm): 创建一个旅行建议技能结合天气和时间给出简单建议。 这是一个简单的链式实现。 # 1. 定义提示词模板 prompt_template ChatPromptTemplate.from_messages([ (system, 你是一个旅行助手。请根据提供的天气和时间信息给用户一句简单的出行建议。), (user, 城市{city}\n当前天气{weather}\n当前时间{time}) ]) # 2. 构建链提示词 - LLM - 解析输出 chain prompt_template | llm | StrOutputParser() def skill_function(city: str): Skill的主函数 # 调用底层Tool获取数据 weather_info get_weather.invoke({city: city}) time_info get_current_time.invoke({}) # 将数据传入链 advice chain.invoke({ city: city, weather: weather_info, time: time_info }) return advice return skill_function # 使用示例 # from langchain_openai import ChatOpenAI # llm ChatOpenAI(modelgpt-4o-mini) # skill create_travel_advice_skill(llm) # result skill(北京) # print(result) # 输出当前北京天气晴朗温度25度时间是下午适合户外活动。这个skill_function封装了内部调用多个 Tool 并与 LLM 协作的细节。对外它就像一个更强大的“工具”。4. 构建智能体 (Agent) 与工作流 (Workflow)有了 Tool 和 Skill现在需要一个大模型作为“指挥官”Agent来根据用户目标决定调用哪个 Tool/Skill并处理它们的结果。4.1 创建基础的单 Agent 系统LangChain 提供了多种 Agent 类型。最常用的是create_react_agent它采用 ReAct 框架Reasoning Acting让 Agent 边思考边行动。在agents/basic_agent.py中# agents/basic_agent.py from langchain import hub from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from ..tools.weather_tool import get_weather, get_current_time def create_basic_agent(llm): 创建一个具备天气和时间查询能力的简单Agent # 1. 将我们的函数包装成 LangChain 的 Tool 对象列表 tools [ Tool( nameGetWeather, funcget_weather.invoke, # 注意这里调用 .invoke description查询指定城市的当前天气。输入是一个城市名称。 ), Tool( nameGetCurrentTime, funcget_current_time.invoke, description获取当前的日期和时间。不需要输入参数。 ), ] # 2. 从 LangChain Hub 拉取一个预设的 ReAct 提示词这是一个很好的起点 prompt hub.pull(hwchase17/react) # 3. 创建 Agent agent create_react_agent(llm, tools, prompt) # 4. 创建 Agent 执行器它负责运行 Agent处理工具调用循环 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 设为 True 可以看到 Agent 的思考过程调试时非常有用 handle_parsing_errorsTrue, # 处理解析错误避免崩溃 max_iterations5 # 限制最大迭代次数防止死循环 ) return agent_executor # 使用示例 # llm ChatOpenAI(modelgpt-4o-mini) # agent create_basic_agent(llm) # result agent.invoke({input: 现在北京天气怎么样现在几点了}) # print(result[output])运行这个 Agent当它遇到需要查询天气或时间时会自动调用对应的 Tool。verboseTrue会让你在控制台看到类似下面的思考过程 Entering new AgentExecutor chain... 我需要回答两个问题北京的天气和当前时间。我有两个工具可用GetWeather 和 GetCurrentTime。 我应该先查询天气再查询时间。 Action: GetWeather Action Input: {city: 北京} Observation: 北京的天气是晴朗温度25摄氏度。 Thought: 现在我有了天气信息需要查询当前时间。 Action: GetCurrentTime Action Input: {} Observation: 2024-05-27 14:30:15 Thought: 我现在可以回答用户的问题了。 Final Answer: 北京当前天气晴朗温度25摄氏度。现在时间是2024-05-27 14:30:15。4.2 使用 LangGraph 构建复杂多步骤 Workflow当任务不再是简单的“问答-调用”而是需要严格顺序、条件判断或循环时就需要 Workflow。LangGraph 是构建这种工作流的强大工具。假设我们要构建一个“旅行规划” Workflow先查天气如果天气好就生成户外活动建议如果天气不好就生成室内活动建议。最后汇总报告。在workflows/travel_planner_workflow.py中# workflows/travel_planner_workflow.py from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage import operator # 1. 定义工作流的状态State class PlannerState(TypedDict): 工作流的状态定义所有节点都读写这个状态。 city: str # 用户输入的城市 weather: str # 天气查询结果 activity_type: Literal[outdoor, indoor] # 活动类型决策 activity_suggestion: str # 活动建议 final_report: str # 最终报告 # 2. 定义各个节点Node函数 def fetch_weather(state: PlannerState): 节点1查询天气 print(f[节点1] 正在查询 {state[city]} 的天气...) # 这里应该调用真实的天气 Tool为简化使用模拟 state[weather] f{state[city]}天气晴朗。 return state def decide_activity(state: PlannerState): 节点2根据天气决定活动类型 print(f[节点2] 根据天气「{state[weather]}」做决策...) # 简单的规则如果天气描述里有“晴朗”或“晴”就户外否则室内 if 晴朗 in state[weather] or 晴 in state[weather]: state[activity_type] outdoor else: state[activity_type] indoor return state def suggest_outdoor(state: PlannerState): 节点3生成户外活动建议 print(f[节点3] 生成户外活动建议...) state[activity_suggestion] f建议在{state[city]}进行户外活动公园散步、骑行、参观露天景点。 return state def suggest_indoor(state: PlannerState): 节点4生成室内活动建议 print(f[节点4] 生成室内活动建议...) state[activity_suggestion] f建议在{state[city]}进行室内活动参观博物馆、美术馆、享受当地美食。 return state def generate_report(state: PlannerState): 节点5生成最终报告 print(f[节点5] 生成最终报告...) state[final_report] f 旅行规划报告 目的地{state[city]} 天气情况{state[weather]} 推荐活动类型{state[activity_type]} 具体建议{state[activity_suggestion]} return state # 3. 构建图Graph def create_travel_planner_workflow(): builder StateGraph(PlannerState) # 添加节点 builder.add_node(fetch_weather, fetch_weather) builder.add_node(decide_activity, decide_activity) builder.add_node(suggest_outdoor, suggest_outdoor) builder.add_node(suggest_indoor, suggest_indoor) builder.add_node(generate_report, generate_report) # 设置入口点 builder.set_entry_point(fetch_weather) # 添加边定义执行流程 builder.add_edge(fetch_weather, decide_activity) # 条件边根据 decide_activity 的结果决定下一个节点 builder.add_conditional_edges( decide_activity, # 这个函数根据当前状态返回下一个节点的名称 lambda state: suggest_outdoor if state[activity_type] outdoor else suggest_indoor, { suggest_outdoor: suggest_outdoor, suggest_indoor: suggest_indoor } ) builder.add_edge(suggest_outdoor, generate_report) builder.add_edge(suggest_indoor, generate_report) builder.add_edge(generate_report, END) # 连接到结束 # 编译图 graph builder.compile() return graph # 使用示例 # graph create_travel_planner_workflow() # initial_state {city: 杭州} # result graph.invoke(initial_state) # print(result[final_report])这个 Workflow 清晰地定义了任务流并且包含了条件分支。你可以通过graph.get_graph().draw_mermaid_png()来生成可视化的流程图这对于理解和调试复杂工作流至关重要。4.3 迈向 Multi-Agent 系统当单个 Agent 负担过重或需要专业分工时就需要 Multi-Agent。例如一个“数据分析报告生成系统”可能包含数据收集 Agent负责调用 API 获取数据。数据分析 Agent负责清洗、计算指标。报告撰写 Agent负责根据分析结果生成文本。校对审核 Agent负责检查报告的格式和逻辑。在 LangGraph 中每个 Agent 可以建模为一个节点它们通过共享的状态State或消息传递进行协作。构建 Multi-Agent 系统的关键点在于明确每个 Agent 的职责和工具集。设计清晰的消息路由逻辑哪个 Agent 在什么情况下将任务或结果传递给谁。管理好共享状态避免冲突。一个简单的设计模式是“主管-工作者”Manager-Worker一个主管 Agent 接收用户请求将其分解然后分配给不同的工作者 Agent 执行最后汇总结果。5. 调试、优化与生产化考量把 Demo 跑起来只是第一步。要让 Agent 系统真正可用你需要关注以下方面。5.1 调试与日志开启 Verbose 模式在开发阶段将AgentExecutor或Chain的verboseTrue这是理解 Agent 思考过程的最直接方式。结构化日志不要只用print。使用 Python 的logging模块将不同级别INFO, DEBUG, ERROR的日志输出到文件方便追踪。LangSmith如果你使用 LangChain强烈建议集成LangSmithLangChain 官方平台。它能可视化追踪每一次链、每一次工具调用的输入、输出、耗时和内部步骤是调试复杂 Agent 系统的神器。5.2 性能与稳定性优化超时与重试为 Tool 的 API 调用设置超时timeout和自动重试逻辑如tenacity库防止网络波动导致整个流程卡死。限制迭代次数务必为AgentExecutor设置max_iterations如 10-20防止 Agent 陷入无休止的思考循环。缓存对于耗时的、结果不变的查询如某些数据查询可以使用langchain.cache来缓存结果提升响应速度。流式输出对于生成时间较长的响应使用流式输出Streaming来提升用户体验。5.3 生产部署考虑配置管理将所有配置API Keys、模型名称、超时时间抽离到配置文件如config.yaml或环境变量中不要硬编码在代码里。异常处理在每个 Tool、Skill 和 Workflow 节点都要有完善的异常捕获返回有意义的错误信息而不是让整个系统崩溃。可观测性除了日志考虑集成监控指标如 Prometheus监控 Agent 的调用次数、成功率、耗时等。API 化使用 FastAPI 或 Flask 将你的 Agent 系统包装成 HTTP API 服务方便与其他系统集成。版本管理对 Agent 的提示词Prompt、工具集、工作流定义进行版本控制。5.4 常见问题排查清单当你的 Agent 表现不如预期时按这个顺序排查LLM 调用失败检查 API Key 是否正确、是否有余额。检查网络连接。如果是本地模型检查模型服务如 Ollama是否正在运行。Tool 无法被调用检查 Tool 的description是否清晰。LLM 完全依赖这个描述来选择工具。检查 Tool 的输入参数格式。LLM 生成的参数是否与函数签名匹配在verbose模式下观察 Agent 是否生成了正确的Action和Action Input。Agent 陷入循环或无法完成任务降低max_iterations看看它卡在哪一步。检查提示词Prompt是否清晰定义了任务边界和停止条件。可能是 Tool 返回的结果格式让 LLM 困惑尝试让 Tool 返回更简洁、结构化的结果。Workflow 状态错误在 LangGraph 中仔细检查每个节点对state的读写是否正确。使用graph.get_graph().draw_mermaid_png()可视化你的工作流检查边Edges的连接逻辑。性能低下检查是 LLM 生成慢还是某个 Tool如网络请求慢。考虑对 LLM 调用或 Tool 调用进行缓存。对于可并行的任务在 LangGraph 中可以使用异步节点。从 Tool 这个最小的零件开始到 Skill 这个功能模块再到能自主决策的 Agent最后用 Workflow 把多个 Agent 或步骤编排起来——这就是构建 AI Agent 应用的完整路径。最关键的始终是先让最简单的单 Agent 单 Tool 跑起来然后逐步增加复杂度每加一步都充分测试。不要一开始就试图设计一个庞大的 Multi-Agent 系统那会让你在细节中迷失。先把基础打牢理解数据流和控制流后续的扩展就会水到渠成。