LangChain 这两年的迭代速度很快从早期的链式调用到 LangGraph 的图编排再到 1.x 时代把 Agent 构建推向工程化社区讨论的话题已经从“能不能做”变成“怎么做得稳”。很多人都在聊 Agent Skills但多数教程只讲概念不教落地。这篇文章直接给出 12 个可运行案例从单个技能封装一直写到多 Agent 协作、LangServe 部署目标是把 LangChain 里的 Agent Skills 架构彻底讲透。先说结论Agent Skills 不是一个新的魔法组件而是一种架构模式。它的核心思想是把 Agent 的能力拆成小颗粒技能每个技能只做一件事Agent 负责意图理解和技能路由。LangChain LangGraph 已经提供了完整的组件不需要额外引入重量级框架。你需要的是一套清晰的封装方法和编排思路。本文会从环境准备开始逐步覆盖技能封装、工具调用、RAG 检索、会话记忆、多 Agent 协作、流式输出和 LangServe API 部署。前 6 个案例打底后 6 个案例偏工程化。所有示例都基于 LangGraph 的 prebuilt API代码改动量小适合快速验证效果。如果你准备在自己的项目里引入 Agent或者正在纠结怎么组织工具调用逻辑这篇可以直接收藏。1. 核心能力速览能力项说明项目类型Agent 编排框架与架构模式核心组件LangChain LangGraph LangServe LangSmithAgent Skills 是什么将能力封装为可独立复用、可路由调用的技能单元与 LangGraph 关系LangGraph 提供状态图、记忆、流式等底层编排能力LangChain 负责模型与工具封装所需环境Python 3.9能访问 LLM API 即可是否有 GUI有通过 LangServe Playground 可视化测试是否支持 API支持LangServe 自动暴露 invoke / stream / batch 端点是否支持批量任务支持batch 端点或代码循环调用是否支持本地模型支持通过 Ollama、vLLM 等接入适合场景知识库问答、数据分析、自动化流程、多 Agent 协作12 个案例的分布逻辑是这样的案例 1-3技能封装基础搞懂一个 Agent 怎么调用一个技能。案例 4-6上下文增强把 RAG、记忆、路由接入 Agent。案例 7-9组合编排并行调用、复合技能、多 Agent。案例 10-12工程化落地流式输出、异常处理、API 部署。2. 架构关系与适用场景2.1 LangChain、LangGraph 与 Agent Skills 的关系很多读者问过一个问题LangGraph 和 LangChain 到底有什么区别简单理解LangChain 是组件库负责模型调用、提示词、工具封装、输出解析LangGraph 是编排框架负责状态管理、节点流转、记忆、并行和分支。两者的关系不是替代而是分层配合。Agent Skills 架构在这套体系里属于应用层设计模式。它强调把“某个领域的完整能力”封装成独立模块而不是把几十个函数散落在代码里。在 LangChain 中一个 Skill 最常见的载体是tool装饰器封装出的工具或者是一个独立的 LangGraph 子图。Agent 本身负责分解任务Skill 负责执行具体操作。两者的边界可以这样划分Agent 是决策主体理解目标、拆解步骤、选择技能、检验结果。Skill 是能力单元封装工具、提示词、上下文处理逻辑。一个 Agent 可以挂载多个 Skill一个 Skill 可以被多个 Agent 复用。2.2 适合什么场景Agent Skills 架构最适合以下三类场景第一类是知识库问答。把文档检索封装成一个 SkillAgent 先判断问题是否需要查库需要就调用检索技能不需要就直接回答避免每次都把上下文塞给模型。第二类是数据分析。把 Pandas 处理、报表生成封装成 SkillAgent 根据用户描述匹配合适的数据处理函数返回结构化结果。第三类是自动化流程。比如工单分类、内容审核、周报生成把每一条规则都独立成一个 SkillAgent 根据输入内容做路由。2.3 使用边界与合规提醒这套架构并不适合所有问题。如果任务的步骤完全固定用普通的函数调用比 Agent 更靠谱如果技能的描述写不清楚Agent 会频繁选错工具反而增加延迟和 token 消耗。需要特别注意合规边界。Agent 接入企业数据时要做好脱敏和权限控制确保 Skill 只能访问当前用户有权限读取的内容调用外部 API 时要确认服务条款允许自动化调用如果 Agent 用于对外服务必须在输出环节增加审核机制防止模型生成不实信息。涉及人脸、声音、个人敏感信息的数据处理必须先获得明确授权再进入 Agent 流程。3. 环境准备与前置条件3.1 安装依赖建议使用独立的 Python 虚拟环境避免依赖冲突。python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install -U langchain langchain-openai langchain-text-splitters langchain-community langgraph langserve # 如果使用本地模型额外安装 pip install -U langchain-ollama3.2 配置模型本文示例默认使用 OpenAI 兼容接口。先设置环境变量export OPENAI_API_KEYsk-xxx如果你是本地模型可以使用 Ollama 加载模型from langchain_ollama import ChatOllama model ChatOllama(modelqwen2.5:7b, temperature0)更稳妥的做法是把模型初始化单独放到一个model.py里后续所有案例统一引用。这样可以随时切换远端 API 和本地模型不影响业务代码。4. 案例 1-3技能封装基础4.1 案例一最小 Agent封装一个天气技能这个案例的目标是跑通最小闭环定义技能 - 创建 Agent - 调用成功。from langchain_openai import ChatOpenAI from langchain.tools import tool from langgraph.prebuilt import create_react_agent tool def get_weather(city: str) - str: 查询指定城市的天气返回天气描述。 return f{city} 今天天气晴朗气温 25℃。 model ChatOpenAI(modelgpt-4o-mini, temperature0) agent create_react_agent(model, tools[get_weather]) result agent.invoke({ messages: [{role: user, content: 北京今天天气怎么样}] }) for message in result[messages]: message.pretty_print()运行后可以看到模型的思考过程先判断用户需要天气信息然后调用get_weather最后把结果整理成自然语言回复。判断成功标准是输出里出现 Tool Call 记录并且最后一条消息是带着天气内容的 Assistant 回复。如果模型没有调用工具优先检查技能描述是否清晰以及模型是否支持工具调用。4.2 案例二技能描述优化决定 Agent 什么时候选它同样是工具描述不同Agent 的错误率差异很大。看下面这个例子tool def calculate_fibonacci(n: int) - int: 计算斐波那契数列第 n 项的值。n 从 0 开始0 对应第 0 项。 if n 0: raise ValueError(n 必须是大于等于 0 的整数) if n 1: return n a, b 0, 1 for _ in range(2, n 1): a, b b, a b return b好的技能描述应该包含三层信息这个技能能做什么、参数含义是什么、什么情况下不要使用。比如calculate_fibonacci的描述中明确写了参数范围和起始项模型就不会在用户问“第 1 项”和“第 0 项”时产生歧义。实战中如果发现 Agent 反复调用同一个错误工具最直接的手段不是改 Prompt而是改工具描述。把“何时不用”写进描述里比在系统提示词中强调一百遍更有效。4.3 案例三数据处理技能把 Pandas 能力交给 Agent实际项目里Agent 经常需要处理表格数据。下面这个技能封装了销售报表分析import pandas as pd from langchain.tools import tool tool def analyze_sales_report(csv_path: str) - str: 读取销售报表 CSV 文件返回总销售额、订单数和 TOP3 商品。 文件必须包含 amount 和 product 两列。 df pd.read_csv(csv_path) total_sales float(df[amount].sum()) order_count len(df) top3 ( df.groupby(product)[amount] .sum() .nlargest(3) .to_dict() ) return f总销售额 {total_sales}订单数 {order_count}TOP3 商品{top3}这里的关键点是将数据校验放在技能内部。CSV 缺列、类型不对、路径错误都应该在技能里抛出明确异常而不是让模型去猜。Agent 拿到异常信息后会自动调整调用方式比如重新传参数。这类技能的价值在于隔离复杂度。业务侧不需要知道 Pandas 怎么用Agent 也不需要理解 CSV 结构所有细节都收敛在技能内部。5. 案例 4-6上下文增强5.1 案例四RAG 技能给 Agent 加知识库让 Agent 回答内部制度问题最合理的方式不是把文档全量塞进上下文而是封装一个检索技能。先准备索引from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.vectorstores import InMemoryVectorStore from langchain_community.document_loaders import TextLoader loader TextLoader(./docs/company_policy.txt) documents loader.load() splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50 ) chunks splitter.split_documents(documents) vectorstore InMemoryVectorStore.from_documents( chunks, OpenAIEmbeddings() ) retriever vectorstore.as_retriever()然后封装成技能tool def query_company_policy(question: str) - str: 查询公司制度文档回答请假、报销、考勤等政策问题。 docs retriever.invoke(question) return \n\n.join(doc.page_content for doc in docs)RAG 技能和普通工具最大的区别在于它返回的是检索结果而不是最终答案。最终答案仍由 Agent 根据检索内容生成。这样设计的好处是文档更新不需要改代码只需要重新构建索引。5.2 案例五记忆技能让 Agent 记住上下文对话场景下Agent 必须知道用户之前问过什么。LangGraph 的 checkpointer 提供了开箱即用的记忆能力from langgraph.checkpoint.memory import InMemorySaver from langgraph.prebuilt import create_react_agent memory InMemorySaver() agent_with_memory create_react_agent( model, tools[get_weather, query_company_policy], checkpointermemory, ) config {configurable: {thread_id: user-session-001}} result agent_with_memory.invoke( { messages: [ {role: user, content: 深圳明天适合出行吗} ] }, config, ) result agent_with_memory.invoke( { messages: [ {role: user, content: 我刚刚问的是哪个城市} ] }, config, )注意这里的thread_id。它相当于会话主键同一个线程内的消息会被自动保留。生产环境建议改用SqliteSaver或PostgresSaver把内存状态持久化到数据库避免重启后丢失。5.3 案例六多技能路由意图分发当 Agent 挂载多个技能后核心挑战是路由。下面这个案例用 LangGraph 的条件边实现意图分发from typing import TypedDict from langchain_core.messages import HumanMessage from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: list intent: str def decide_intent(state: AgentState): user_text state[messages][-1].content decision model.invoke( 判断用户意图只输出 weather、policy、data 三者之一。\n f用户消息{user_text} ) return {intent: decision.content.strip().lower()} def weather_node(state: AgentState): return {messages: [已进入天气技能分支]} def policy_node(state: AgentState): return {messages: [已进入制度问答分支]} def data_node(state: AgentState): return {messages: [已进入数据分析分支]} graph StateGraph(AgentState) graph.add_node(router, decide_intent) graph.add_node(weather, weather_node) graph.add_node(policy, policy_node) graph.add_node(data, data_node) graph.add_edge(START, router) graph.add_conditional_edges( router, lambda state: state[intent], { weather: weather, policy: policy, data: data, } ) graph.add_edge(weather, END) graph.add_edge(policy, END) graph.add_edge(data, END) app graph.compile()条件路由的价值在于不需要让模型在一个大 Prompt 里完成所有决策。每个分支节点只做自己职责范围内的事逻辑更清晰也更容易加权限控制和日志。6. 案例 7-9组合编排6.1 案例七并行技能调用如果一次任务需要查询多个城市串行调用会明显增加耗时。LangGraph 的SendAPI 支持动态并行分发from langgraph.types import Send def dispatch_parallel(state): cities state[cities] return [ Send(query_weather, {city: city}) for city in cities ]配合图结构每个城市都会独立进入query_weather节点并行执行完后汇总结果。这个模式适合批量数据获取、多文件处理、独立子任务并发等场景。6.2 案例八复合技能技能调用技能单个 Agent 里的技能也可以互相调用。LangChain 的 tool 对象本身支持.invoke()因此可以在一个技能里嵌套另一个技能tool def business_report(city: str) - str: 生成某城市的综合经营日报先查天气再查门店数据。 weather get_weather.invoke({city: city}) return f【{city} 日报】\n天气{weather}\n门店状态营业中复合技能的核心优势是复用。底层技能保持单一职责上层技能负责流程编排。如果天气查询逻辑升级只需要改get_weather所有依赖它的技能自动生效。6.3 案例九多 Agent 协作当任务跨度大、单一 Agent 难以完成时可以拆成多个 Agent。下面是一个“研究员 写作者”的最小协作图from langgraph.prebuilt import create_react_agent researcher create_react_agent( model, tools[query_company_policy], ) writer create_react_agent(model, tools[]) def research_node(state): result researcher.invoke( {messages: state[task]} ) return {research: result[messages][-1].content} def write_node(state): prompt ( 根据以下材料撰写一份周报要求语言简洁\n f{state[research]} ) result writer.invoke( {messages: [{role: user, content: prompt}]} ) return {report: result[messages][-1].content}这个案例里研究员只负责检索素材写作者负责组织语言。职责分离后每个 Agent 的 Prompt 都可以精简错误率也会下降。多 Agent 不是越多越好两个角色能解决的问题不要拆成三个。7. 案例 10-12工程化落地7.1 案例十流式输出用户等待 Agent 执行时流式输出能显著提升体验。LangGraph 的事件流接口天然支持events agent.stream( {messages: [{role: user, content: 北京今天天气怎么样}]}, stream_modeupdates, ) for event in events: for node_name, node_output in event.items(): if messages in node_output: msg node_output[messages][-1] if getattr(msg, content, None): print(node_name, :, msg.content)流式输出的关键不是炫技而是让调用方可以边生成边渲染。在服务端集成时可以换成 FastAPI 的 StreamingResponse把事件流推送前端。7.2 案例十一异常处理与重试Agent 在长时间运行中会遇到工具报错、模型返回格式异常、超过最大步数等问题。一定要显式捕获from langgraph.errors import GraphRecursionError from langchain_core.messages import HumanMessage try: result agent.invoke( {messages: [HumanMessage(content北京天气怎么样)]}, {recursion_limit: 20}, ) except GraphRecursionError: print(Agent 执行超过限制步数请检查技能描述是否清晰)另外建议给每个技能增加超时和异常返回。工具内部不要抛出不可读的异常堆栈而是返回固定格式的错误信息让模型知道下一步该怎么调整。7.3 案例十二LangServe 部署暴露 API 服务最后一步是把 Agent 部署成标准 API。LangServe 提供了完整的 FastAPI 集成from fastapi import FastAPI from langserve import add_routes from langgraph.checkpoint.memory import InMemorySaver from langgraph.prebuilt import create_react_agent app FastAPI( titleAgent Skills API, version1.0.0, ) agent create_react_agent( model, tools[get_weather, query_company_policy], checkpointerInMemorySaver(), ) add_routes(app, agent, path/agent) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动python server.py启动后可以直接访问http://127.0.0.1:8000/agent/playgroundLangServe 会生成一个可视化的调试界面支持手动输入消息、流式查看工具调用过程。8. 批量任务与接口 API8.1 使用 LangServe 的 Invoke 端点服务启动后客户端可以发送 POST 请求curl -X POST http://127.0.0.1:8000/agent/invoke \ -H Content-Type: application/json \ -d { input: { messages: [ {role: user, content: 北京今天天气怎么样} ] }, config: { configurable: {thread_id: demo-001} } }返回结果里面包含完整的消息列表可以提取最后一条 content 作为 Agent 回复。8.2 Python 批量调用与结果汇总批量任务最简单的实现方式是循环调用。以下示例串行处理多个城市并输出结构化结果import time import requests cities [北京, 上海, 广州, 深圳, 杭州] results [] for city in cities: payload { input: { messages: [ {role: user, content: f{city}今天天气怎么样} ] } } resp requests.post( http://127.0.0.1:8000/agent/invoke, jsonpayload, timeout120, ) data resp.json() reply data[output][messages][-1][content] results.append({city: city, reply: reply}) for row in results: print(row[city], -, row[reply])如果 LangServe 版本支持可以直接调用/agent/batch一次传入多条输入。批量任务必须加日志和失败重试。建议把每次调用的thread_id、入参、返回状态、耗时都记录下来后续排查问题时才有依据。8.3 接口安全建议LangServe 默认没有鉴权部署到服务器后接口完全公开。生产环境必须在前面加一层认证或者通过内网访问。至少要做三件事限制访问 IP、增加 API Key 校验、为模型服务单独设置额度上限。9. 性能观察与调试建议9.1 用 LangSmith 追踪每一步Agent 一旦进入多技能组合调试难度会明显上升。建议从开发初期就接入 LangSmith观察每个节点的输入输出、token 消耗和耗时。重点看三类数据路由是否正确Agent 是否把问题分配给了正确的技能。工具调用成功率技能内部是否有报错、是否反复重试。延迟分布耗时花在模型推理上还是工具执行上。9.2 资源占用观察方法如果使用本地模型可以观察显存和内存变化。启动推理服务后使用nvidia-smi查看显存占用使用top或htop查看 CPU 与内存。分辨率、上下文长度、并发数量会明显影响占用实际数值以本机测试为准。9.3 降低开销的常用手段控制上下文长度是最有效的优化方式。长文档先总结再交给 Agent而不是让 Agent 读全文。另一个方式是限制工具数量Agent 每多一个可选项路由判断的开销和出错概率都会增加。10. 常见问题与排查方法问题现象可能原因排查方式解决方案Agent 不调用任何工具技能描述不清晰或模型不支持工具调用检查模型参数与工具描述重写描述明确参数类型和触发条件工具调用报错技能内部抛出了异常查看服务日志中的 Traceback在技能内部捕获异常并返回错误信息多轮对话丢失记忆thread_id 未传或 checkpointer 未配置检查请求中的 config固定 thread_id 并配置持久化存储路由结果不稳定意图分类依赖模型自由输出增加输出格式约束改用结构化输出或强制枚举值请求超时模型推理慢或工具执行阻塞查看耗时统计增加超时时间拆分长任务端口被占用上一次服务未关闭检查端口监听进程换端口或结束旧进程本地模型显存不足模型参数过大或并发过高使用 nvidia-smi 查看降低并发数或换用更小模型11. 最佳实践与使用建议第一次搭建 Agent Skills 架构时不要一上来就设计复杂的图。先写一个最简单的工具跑通create_react_agent再逐步加入 RAG、记忆、路由。每加一个功能都重新验证旧功能是否正常。建议把技能、模型、图结构拆到不同文件管理。技能统一放在skills/目录每个技能一个模块模型初始化放在model.py图编排放在agents/目录。这样团队协作时新增技能不影响主流程。批量任务务必加日志和失败重试。对每次调用记录入参、出参、耗时、错误信息。任务量超过几百条时使用消息队列而不是简单循环。接口服务必须限制访问范围。LangServe 部署在内网即可满足大部分场景。如果必须暴露公网前面必须加 API 网关和鉴权。最后任何涉及企业内部数据、用户隐私或版权素材的 Agent 流程都需要在技能层加入权限校验和审计日志。框架本身不做权限控制这个责任在业务侧。12. 总结Agent Skills 架构的核心并不复杂用 LangChain 封装能力用 LangGraph 编排流程用 LangServe 暴露接口。12 个案例从单个技能到多 Agent 协作覆盖了日常开发中最常见的落地路径。最值得先跑通的是案例一。用几分钟验证模型能正确调用工具后面的路由、记忆、部署才谈得上稳定性。最容易踩的坑是技能描述写得太模糊导致 Agent 反复选错工具实际开发中大部分问题出在这里而不是出在框架本身。后续可以继续扩展的方向包括把 LangServe 替换成自定义 FastAPI 服务精细控制鉴权和限流用 Postgres 替代内存 checkpointer让记忆持久化把复合技能升级为独立 LangGraph 子图支持更复杂的流程控制。先把一个技能跑通再考虑架构扩展。这是最稳妥的路线。 
长河君 · 2026/8/31 2:18:38 · 阅读 1.2k LangChain Agent Skills 实战：12个案例从封装到部署
LangChain 这两年的迭代速度很快从早期的链式调用到 LangGraph 的图编排再到 1.x 时代把 Agent 构建推向工程化社区讨论的话题已经从“能不能做”变成“怎么做得稳”。很多人都在聊 Agent Skills但多数教程只讲概念不教落地。这篇文章直接给出 12 个可运行案例从单个技能封装一直写到多 Agent 协作、LangServe 部署目标是把 LangChain 里的 Agent Skills 架构彻底讲透。先说结论Agent Skills 不是一个新的魔法组件而是一种架构模式。它的核心思想是把 Agent 的能力拆成小颗粒技能每个技能只做一件事Agent 负责意图理解和技能路由。LangChain LangGraph 已经提供了完整的组件不需要额外引入重量级框架。你需要的是一套清晰的封装方法和编排思路。本文会从环境准备开始逐步覆盖技能封装、工具调用、RAG 检索、会话记忆、多 Agent 协作、流式输出和 LangServe API 部署。前 6 个案例打底后 6 个案例偏工程化。所有示例都基于 LangGraph 的 prebuilt API代码改动量小适合快速验证效果。如果你准备在自己的项目里引入 Agent或者正在纠结怎么组织工具调用逻辑这篇可以直接收藏。1. 核心能力速览能力项说明项目类型Agent 编排框架与架构模式核心组件LangChain LangGraph LangServe LangSmithAgent Skills 是什么将能力封装为可独立复用、可路由调用的技能单元与 LangGraph 关系LangGraph 提供状态图、记忆、流式等底层编排能力LangChain 负责模型与工具封装所需环境Python 3.9能访问 LLM API 即可是否有 GUI有通过 LangServe Playground 可视化测试是否支持 API支持LangServe 自动暴露 invoke / stream / batch 端点是否支持批量任务支持batch 端点或代码循环调用是否支持本地模型支持通过 Ollama、vLLM 等接入适合场景知识库问答、数据分析、自动化流程、多 Agent 协作12 个案例的分布逻辑是这样的案例 1-3技能封装基础搞懂一个 Agent 怎么调用一个技能。案例 4-6上下文增强把 RAG、记忆、路由接入 Agent。案例 7-9组合编排并行调用、复合技能、多 Agent。案例 10-12工程化落地流式输出、异常处理、API 部署。2. 架构关系与适用场景2.1 LangChain、LangGraph 与 Agent Skills 的关系很多读者问过一个问题LangGraph 和 LangChain 到底有什么区别简单理解LangChain 是组件库负责模型调用、提示词、工具封装、输出解析LangGraph 是编排框架负责状态管理、节点流转、记忆、并行和分支。两者的关系不是替代而是分层配合。Agent Skills 架构在这套体系里属于应用层设计模式。它强调把“某个领域的完整能力”封装成独立模块而不是把几十个函数散落在代码里。在 LangChain 中一个 Skill 最常见的载体是tool装饰器封装出的工具或者是一个独立的 LangGraph 子图。Agent 本身负责分解任务Skill 负责执行具体操作。两者的边界可以这样划分Agent 是决策主体理解目标、拆解步骤、选择技能、检验结果。Skill 是能力单元封装工具、提示词、上下文处理逻辑。一个 Agent 可以挂载多个 Skill一个 Skill 可以被多个 Agent 复用。2.2 适合什么场景Agent Skills 架构最适合以下三类场景第一类是知识库问答。把文档检索封装成一个 SkillAgent 先判断问题是否需要查库需要就调用检索技能不需要就直接回答避免每次都把上下文塞给模型。第二类是数据分析。把 Pandas 处理、报表生成封装成 SkillAgent 根据用户描述匹配合适的数据处理函数返回结构化结果。第三类是自动化流程。比如工单分类、内容审核、周报生成把每一条规则都独立成一个 SkillAgent 根据输入内容做路由。2.3 使用边界与合规提醒这套架构并不适合所有问题。如果任务的步骤完全固定用普通的函数调用比 Agent 更靠谱如果技能的描述写不清楚Agent 会频繁选错工具反而增加延迟和 token 消耗。需要特别注意合规边界。Agent 接入企业数据时要做好脱敏和权限控制确保 Skill 只能访问当前用户有权限读取的内容调用外部 API 时要确认服务条款允许自动化调用如果 Agent 用于对外服务必须在输出环节增加审核机制防止模型生成不实信息。涉及人脸、声音、个人敏感信息的数据处理必须先获得明确授权再进入 Agent 流程。3. 环境准备与前置条件3.1 安装依赖建议使用独立的 Python 虚拟环境避免依赖冲突。python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate pip install -U langchain langchain-openai langchain-text-splitters langchain-community langgraph langserve # 如果使用本地模型额外安装 pip install -U langchain-ollama3.2 配置模型本文示例默认使用 OpenAI 兼容接口。先设置环境变量export OPENAI_API_KEYsk-xxx如果你是本地模型可以使用 Ollama 加载模型from langchain_ollama import ChatOllama model ChatOllama(modelqwen2.5:7b, temperature0)更稳妥的做法是把模型初始化单独放到一个model.py里后续所有案例统一引用。这样可以随时切换远端 API 和本地模型不影响业务代码。4. 案例 1-3技能封装基础4.1 案例一最小 Agent封装一个天气技能这个案例的目标是跑通最小闭环定义技能 - 创建 Agent - 调用成功。from langchain_openai import ChatOpenAI from langchain.tools import tool from langgraph.prebuilt import create_react_agent tool def get_weather(city: str) - str: 查询指定城市的天气返回天气描述。 return f{city} 今天天气晴朗气温 25℃。 model ChatOpenAI(modelgpt-4o-mini, temperature0) agent create_react_agent(model, tools[get_weather]) result agent.invoke({ messages: [{role: user, content: 北京今天天气怎么样}] }) for message in result[messages]: message.pretty_print()运行后可以看到模型的思考过程先判断用户需要天气信息然后调用get_weather最后把结果整理成自然语言回复。判断成功标准是输出里出现 Tool Call 记录并且最后一条消息是带着天气内容的 Assistant 回复。如果模型没有调用工具优先检查技能描述是否清晰以及模型是否支持工具调用。4.2 案例二技能描述优化决定 Agent 什么时候选它同样是工具描述不同Agent 的错误率差异很大。看下面这个例子tool def calculate_fibonacci(n: int) - int: 计算斐波那契数列第 n 项的值。n 从 0 开始0 对应第 0 项。 if n 0: raise ValueError(n 必须是大于等于 0 的整数) if n 1: return n a, b 0, 1 for _ in range(2, n 1): a, b b, a b return b好的技能描述应该包含三层信息这个技能能做什么、参数含义是什么、什么情况下不要使用。比如calculate_fibonacci的描述中明确写了参数范围和起始项模型就不会在用户问“第 1 项”和“第 0 项”时产生歧义。实战中如果发现 Agent 反复调用同一个错误工具最直接的手段不是改 Prompt而是改工具描述。把“何时不用”写进描述里比在系统提示词中强调一百遍更有效。4.3 案例三数据处理技能把 Pandas 能力交给 Agent实际项目里Agent 经常需要处理表格数据。下面这个技能封装了销售报表分析import pandas as pd from langchain.tools import tool tool def analyze_sales_report(csv_path: str) - str: 读取销售报表 CSV 文件返回总销售额、订单数和 TOP3 商品。 文件必须包含 amount 和 product 两列。 df pd.read_csv(csv_path) total_sales float(df[amount].sum()) order_count len(df) top3 ( df.groupby(product)[amount] .sum() .nlargest(3) .to_dict() ) return f总销售额 {total_sales}订单数 {order_count}TOP3 商品{top3}这里的关键点是将数据校验放在技能内部。CSV 缺列、类型不对、路径错误都应该在技能里抛出明确异常而不是让模型去猜。Agent 拿到异常信息后会自动调整调用方式比如重新传参数。这类技能的价值在于隔离复杂度。业务侧不需要知道 Pandas 怎么用Agent 也不需要理解 CSV 结构所有细节都收敛在技能内部。5. 案例 4-6上下文增强5.1 案例四RAG 技能给 Agent 加知识库让 Agent 回答内部制度问题最合理的方式不是把文档全量塞进上下文而是封装一个检索技能。先准备索引from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_core.vectorstores import InMemoryVectorStore from langchain_community.document_loaders import TextLoader loader TextLoader(./docs/company_policy.txt) documents loader.load() splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50 ) chunks splitter.split_documents(documents) vectorstore InMemoryVectorStore.from_documents( chunks, OpenAIEmbeddings() ) retriever vectorstore.as_retriever()然后封装成技能tool def query_company_policy(question: str) - str: 查询公司制度文档回答请假、报销、考勤等政策问题。 docs retriever.invoke(question) return \n\n.join(doc.page_content for doc in docs)RAG 技能和普通工具最大的区别在于它返回的是检索结果而不是最终答案。最终答案仍由 Agent 根据检索内容生成。这样设计的好处是文档更新不需要改代码只需要重新构建索引。5.2 案例五记忆技能让 Agent 记住上下文对话场景下Agent 必须知道用户之前问过什么。LangGraph 的 checkpointer 提供了开箱即用的记忆能力from langgraph.checkpoint.memory import InMemorySaver from langgraph.prebuilt import create_react_agent memory InMemorySaver() agent_with_memory create_react_agent( model, tools[get_weather, query_company_policy], checkpointermemory, ) config {configurable: {thread_id: user-session-001}} result agent_with_memory.invoke( { messages: [ {role: user, content: 深圳明天适合出行吗} ] }, config, ) result agent_with_memory.invoke( { messages: [ {role: user, content: 我刚刚问的是哪个城市} ] }, config, )注意这里的thread_id。它相当于会话主键同一个线程内的消息会被自动保留。生产环境建议改用SqliteSaver或PostgresSaver把内存状态持久化到数据库避免重启后丢失。5.3 案例六多技能路由意图分发当 Agent 挂载多个技能后核心挑战是路由。下面这个案例用 LangGraph 的条件边实现意图分发from typing import TypedDict from langchain_core.messages import HumanMessage from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): messages: list intent: str def decide_intent(state: AgentState): user_text state[messages][-1].content decision model.invoke( 判断用户意图只输出 weather、policy、data 三者之一。\n f用户消息{user_text} ) return {intent: decision.content.strip().lower()} def weather_node(state: AgentState): return {messages: [已进入天气技能分支]} def policy_node(state: AgentState): return {messages: [已进入制度问答分支]} def data_node(state: AgentState): return {messages: [已进入数据分析分支]} graph StateGraph(AgentState) graph.add_node(router, decide_intent) graph.add_node(weather, weather_node) graph.add_node(policy, policy_node) graph.add_node(data, data_node) graph.add_edge(START, router) graph.add_conditional_edges( router, lambda state: state[intent], { weather: weather, policy: policy, data: data, } ) graph.add_edge(weather, END) graph.add_edge(policy, END) graph.add_edge(data, END) app graph.compile()条件路由的价值在于不需要让模型在一个大 Prompt 里完成所有决策。每个分支节点只做自己职责范围内的事逻辑更清晰也更容易加权限控制和日志。6. 案例 7-9组合编排6.1 案例七并行技能调用如果一次任务需要查询多个城市串行调用会明显增加耗时。LangGraph 的SendAPI 支持动态并行分发from langgraph.types import Send def dispatch_parallel(state): cities state[cities] return [ Send(query_weather, {city: city}) for city in cities ]配合图结构每个城市都会独立进入query_weather节点并行执行完后汇总结果。这个模式适合批量数据获取、多文件处理、独立子任务并发等场景。6.2 案例八复合技能技能调用技能单个 Agent 里的技能也可以互相调用。LangChain 的 tool 对象本身支持.invoke()因此可以在一个技能里嵌套另一个技能tool def business_report(city: str) - str: 生成某城市的综合经营日报先查天气再查门店数据。 weather get_weather.invoke({city: city}) return f【{city} 日报】\n天气{weather}\n门店状态营业中复合技能的核心优势是复用。底层技能保持单一职责上层技能负责流程编排。如果天气查询逻辑升级只需要改get_weather所有依赖它的技能自动生效。6.3 案例九多 Agent 协作当任务跨度大、单一 Agent 难以完成时可以拆成多个 Agent。下面是一个“研究员 写作者”的最小协作图from langgraph.prebuilt import create_react_agent researcher create_react_agent( model, tools[query_company_policy], ) writer create_react_agent(model, tools[]) def research_node(state): result researcher.invoke( {messages: state[task]} ) return {research: result[messages][-1].content} def write_node(state): prompt ( 根据以下材料撰写一份周报要求语言简洁\n f{state[research]} ) result writer.invoke( {messages: [{role: user, content: prompt}]} ) return {report: result[messages][-1].content}这个案例里研究员只负责检索素材写作者负责组织语言。职责分离后每个 Agent 的 Prompt 都可以精简错误率也会下降。多 Agent 不是越多越好两个角色能解决的问题不要拆成三个。7. 案例 10-12工程化落地7.1 案例十流式输出用户等待 Agent 执行时流式输出能显著提升体验。LangGraph 的事件流接口天然支持events agent.stream( {messages: [{role: user, content: 北京今天天气怎么样}]}, stream_modeupdates, ) for event in events: for node_name, node_output in event.items(): if messages in node_output: msg node_output[messages][-1] if getattr(msg, content, None): print(node_name, :, msg.content)流式输出的关键不是炫技而是让调用方可以边生成边渲染。在服务端集成时可以换成 FastAPI 的 StreamingResponse把事件流推送前端。7.2 案例十一异常处理与重试Agent 在长时间运行中会遇到工具报错、模型返回格式异常、超过最大步数等问题。一定要显式捕获from langgraph.errors import GraphRecursionError from langchain_core.messages import HumanMessage try: result agent.invoke( {messages: [HumanMessage(content北京天气怎么样)]}, {recursion_limit: 20}, ) except GraphRecursionError: print(Agent 执行超过限制步数请检查技能描述是否清晰)另外建议给每个技能增加超时和异常返回。工具内部不要抛出不可读的异常堆栈而是返回固定格式的错误信息让模型知道下一步该怎么调整。7.3 案例十二LangServe 部署暴露 API 服务最后一步是把 Agent 部署成标准 API。LangServe 提供了完整的 FastAPI 集成from fastapi import FastAPI from langserve import add_routes from langgraph.checkpoint.memory import InMemorySaver from langgraph.prebuilt import create_react_agent app FastAPI( titleAgent Skills API, version1.0.0, ) agent create_react_agent( model, tools[get_weather, query_company_policy], checkpointerInMemorySaver(), ) add_routes(app, agent, path/agent) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动python server.py启动后可以直接访问http://127.0.0.1:8000/agent/playgroundLangServe 会生成一个可视化的调试界面支持手动输入消息、流式查看工具调用过程。8. 批量任务与接口 API8.1 使用 LangServe 的 Invoke 端点服务启动后客户端可以发送 POST 请求curl -X POST http://127.0.0.1:8000/agent/invoke \ -H Content-Type: application/json \ -d { input: { messages: [ {role: user, content: 北京今天天气怎么样} ] }, config: { configurable: {thread_id: demo-001} } }返回结果里面包含完整的消息列表可以提取最后一条 content 作为 Agent 回复。8.2 Python 批量调用与结果汇总批量任务最简单的实现方式是循环调用。以下示例串行处理多个城市并输出结构化结果import time import requests cities [北京, 上海, 广州, 深圳, 杭州] results [] for city in cities: payload { input: { messages: [ {role: user, content: f{city}今天天气怎么样} ] } } resp requests.post( http://127.0.0.1:8000/agent/invoke, jsonpayload, timeout120, ) data resp.json() reply data[output][messages][-1][content] results.append({city: city, reply: reply}) for row in results: print(row[city], -, row[reply])如果 LangServe 版本支持可以直接调用/agent/batch一次传入多条输入。批量任务必须加日志和失败重试。建议把每次调用的thread_id、入参、返回状态、耗时都记录下来后续排查问题时才有依据。8.3 接口安全建议LangServe 默认没有鉴权部署到服务器后接口完全公开。生产环境必须在前面加一层认证或者通过内网访问。至少要做三件事限制访问 IP、增加 API Key 校验、为模型服务单独设置额度上限。9. 性能观察与调试建议9.1 用 LangSmith 追踪每一步Agent 一旦进入多技能组合调试难度会明显上升。建议从开发初期就接入 LangSmith观察每个节点的输入输出、token 消耗和耗时。重点看三类数据路由是否正确Agent 是否把问题分配给了正确的技能。工具调用成功率技能内部是否有报错、是否反复重试。延迟分布耗时花在模型推理上还是工具执行上。9.2 资源占用观察方法如果使用本地模型可以观察显存和内存变化。启动推理服务后使用nvidia-smi查看显存占用使用top或htop查看 CPU 与内存。分辨率、上下文长度、并发数量会明显影响占用实际数值以本机测试为准。9.3 降低开销的常用手段控制上下文长度是最有效的优化方式。长文档先总结再交给 Agent而不是让 Agent 读全文。另一个方式是限制工具数量Agent 每多一个可选项路由判断的开销和出错概率都会增加。10. 常见问题与排查方法问题现象可能原因排查方式解决方案Agent 不调用任何工具技能描述不清晰或模型不支持工具调用检查模型参数与工具描述重写描述明确参数类型和触发条件工具调用报错技能内部抛出了异常查看服务日志中的 Traceback在技能内部捕获异常并返回错误信息多轮对话丢失记忆thread_id 未传或 checkpointer 未配置检查请求中的 config固定 thread_id 并配置持久化存储路由结果不稳定意图分类依赖模型自由输出增加输出格式约束改用结构化输出或强制枚举值请求超时模型推理慢或工具执行阻塞查看耗时统计增加超时时间拆分长任务端口被占用上一次服务未关闭检查端口监听进程换端口或结束旧进程本地模型显存不足模型参数过大或并发过高使用 nvidia-smi 查看降低并发数或换用更小模型11. 最佳实践与使用建议第一次搭建 Agent Skills 架构时不要一上来就设计复杂的图。先写一个最简单的工具跑通create_react_agent再逐步加入 RAG、记忆、路由。每加一个功能都重新验证旧功能是否正常。建议把技能、模型、图结构拆到不同文件管理。技能统一放在skills/目录每个技能一个模块模型初始化放在model.py图编排放在agents/目录。这样团队协作时新增技能不影响主流程。批量任务务必加日志和失败重试。对每次调用记录入参、出参、耗时、错误信息。任务量超过几百条时使用消息队列而不是简单循环。接口服务必须限制访问范围。LangServe 部署在内网即可满足大部分场景。如果必须暴露公网前面必须加 API 网关和鉴权。最后任何涉及企业内部数据、用户隐私或版权素材的 Agent 流程都需要在技能层加入权限校验和审计日志。框架本身不做权限控制这个责任在业务侧。12. 总结Agent Skills 架构的核心并不复杂用 LangChain 封装能力用 LangGraph 编排流程用 LangServe 暴露接口。12 个案例从单个技能到多 Agent 协作覆盖了日常开发中最常见的落地路径。最值得先跑通的是案例一。用几分钟验证模型能正确调用工具后面的路由、记忆、部署才谈得上稳定性。最容易踩的坑是技能描述写得太模糊导致 Agent 反复选错工具实际开发中大部分问题出在这里而不是出在框架本身。后续可以继续扩展的方向包括把 LangServe 替换成自定义 FastAPI 服务精细控制鉴权和限流用 Postgres 替代内存 checkpointer让记忆持久化把复合技能升级为独立 LangGraph 子图支持更复杂的流程控制。先把一个技能跑通再考虑架构扩展。这是最稳妥的路线。