这次我们来看 LangChain 新版的 Agent Skills 架构。先把结论放在最前面如果你现在还在把所有精力花在“提示词怎么改”上而没有把 Agent 拆成技能层后面做复杂业务一定会被上下文混乱、工具选择错误、批量任务不可控这几个问题反复卡住。Agent Skills 的核心就一句话把 Agent 的能力做成分层、可注册、可调用、可验证的单元再交给统一的编排层去调度。为什么这个主题值得专门写一篇因为 LangChain 生态已经进入 LangGraph 编排时代以前“一条 chain 一把梭”的写法正在被有状态、可中断、可恢复的图式执行替代。而 Skills 恰好是 LangGraph 节点和 Tool 之间缺失的那一层抽象Tool 太细节点太碎Skill 是可以独立描述、独立测试、独立复用的一等公民。做工程的人都关心批量任务和 API 接入Skill 层又天然适合统一封装输入输出后面的部署、监控、批量处理都会顺很多。这篇文章会带读者做四件事先看 Agent Skills 在 LangChain 最新架构中的定位再搭一个最小工程骨架然后用 12 个案例逐一拆解从单技能到多 Agent 协作、从本地脚本到 API 部署的完整链路最后给出一份常见问题排查表和生产化建议。适合人群是已经跑过 LangChain 基础 demo、想往工程化方向走的开发者以及团队里正在做 Agent 技术选型的人。如果你只想看概念可以直接跳到第 2 节如果你想照着操作从第 4 节开始。1. 核心能力速览能力项说明架构定位Agent 能力分层与编排模式介于 LangChain Tool 与 LangGraph 节点之间核心特性技能注册、技能路由、记忆、批量执行、人工审批、多 Agent 协作开发语言Python 为主LangChain 官方同时支持 TypeScript部署方式本地脚本 / LangGraph 服务 / FastAPI 封装 / 容器化批量任务支持通过任务队列和并发限制实现适合批量文本、批量文档处理API 接口支持可封装为 REST API 并配合任务状态轮询硬件要求主要依赖大模型推理端CPU 可跑小模型GPU 可降低延迟无强制 CUDA 要求典型场景客服、知识库问答、批量内容生成、文档处理、工作流自动化学习成本中需要理解 Agent、Tool、Skill、Graph 四个概念扩展性技能可插拔路由可替换底层模型可切换这里有一个容易混淆的点Agent Skills 不是某个单独的 pip 包而是一套把 Agent 能力模块化的设计方式。LangChain 提供组件LangGraph 提供状态编排Skill 则用来封装“可复用的完整能力”。下面先把这几个概念彻底理清。2. Agent Skills 是什么先看清架构定位2.1 Agent、Skill、Tool、Graph 的区别很多 LangChain 初学者会把这些概念混在一起。用一张表就能说清层级粒度职责例子Tool原子操作执行具体动作返回结果搜索、计算、写文件、查天气Skill模块化能力面向任务的完整能力组合文档问答、报告生成、代码审查Agent编排者感知用户目标拆解任务选择 Skill客服 Agent、数据分析 AgentGraph流程控制控制节点间状态流转和恢复LangGraph 中的状态图Tool 的问题是太细。一个对话 Agent 如果同时暴露二十个 Tool模型经常分不清该用哪个调用参数也容易传错。Skill 的作用是把相关 Tool 组合成一个“语义完整的能力包”并且给这个能力包写清楚描述和参数让上层 Agent 更容易决策。2.2 为什么需要 Skills 这一层从工程视角看没有 Skills 层的 Agent 会面临四个问题第一是上下文挤占。每个 Tool 的 description 都会进 promptTool 多了以后光工具描述就能占掉大量 token真正给推理用的空间反而变少。Skill 层可以只暴露技能名和高层描述把具体 Tool 细节下沉。第二是职责不清。Agent 既要负责任务拆解又要负责工具实现细节代码会越来越难维护。Skill 把“用户意图”和“底层执行”隔开Agent 只做路由Skill 内部自己管工具、参数和异常。第三是测试困难。Tool 的单元测试容易写但“整个 Agent 能不能完成一个文档问答需求”很难测。Skill 层天然是一个可独立测试的边界输入结构、输出结构、失败行为都能定义清楚。第四是批量任务和生产部署缺少统一接口。没有 Skill 层批量跑任务时只能针对每个 Agent 写一套临时脚本。有了 Skill 层批量调度、接口暴露、任务状态管理都能共用一套框架。2.3 一个最小架构图用户输入 → Agent编排层 → Skill能力层 → Tool执行层 → 模型输出 ↑ ↑ ↑ 意图路由 参数校验 结果返回这个链路就是后面 12 个案例反复使用的主干。3. 新版 LangChain 与 Agent Skills 的关系3.1 从 AgentExecutor 到 LangGraphLangChain 早期版本里最常用的是AgentExecutor。它内部循环执行“思考 → 调用工具 → 观察结果 → 再思考”逻辑简单但状态管理能力弱一旦遇到中断恢复、人工审批、并行分支改起来非常痛苦。新版 LangChain 已经把重心放到 LangGraph。LangGraph 把 Agent 流程建模成一张有状态图节点之间通过共享状态传递数据支持检查点、中断、恢复。这个能力对 Agent Skills 很重要因为 Skill 可能是一个有状态的执行单元比如“等待用户确认”或“上下文累积到某一步后再继续”。3.2 Skills 在 LangChain 生态中的定位LangChain 官方文档中Agent 通常由模型、工具、记忆和编排逻辑组成。“Skill”这个词在不同的 LangChain 版本和社区项目里出现过不同的含义但更稳妥的理解是Skill 是“工具之上、Agent 之下”的能力封装层。一个 Skill 通常包含以下内容技能名称和一句话描述。输入参数 Schema。执行逻辑内部可以调用多个 Tool 或模型。输出结构定义。失败重试和日志处理。LangChain 提供tool装饰器来定义原子工具提供create_react_agent这样的预构建 Agent也提供 LangGraph 节点来自定义流程。Skill 层则建议自己或团队定义一套统一接口把上面的能力组合起来。下面第 4 节就给出一套可以直接落地的工程骨架。4. 环境准备与最小工程结构4.1 环境准备LangChain 基于 Python建议使用 Python 3.10 或 3.11。创建一个虚拟环境然后安装核心依赖。python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install -U langchain langchain-openai langgraph python-dotenv如果你的模型服务来自其他厂商比如本地 Ollama、智谱、通义、Azure OpenAI对应安装langchain-community或对应的langchain-*扩展包。具体包名以当前官方文档为准。配置模型密钥。在项目根目录创建.env文件OPENAI_API_KEYsk-xxxxxxxx # 如果使用自建模型兼容接口 # OPENAI_API_BASEhttp://127.0.0.1:8000/v1 # OPENAI_MODEL_NAMEyour-model启动方式比较简单直接在项目目录执行 Python 脚本或在 LLM 服务就绪后调用接口服务。4.2 推荐的工程目录结构agent_skills_lab/ ├── skills/ │ ├── __init__.py │ ├── base.py # BaseSkill 抽象类 │ ├── registry.py # 技能注册中心 │ ├── chat_skill.py # 通用对话技能 │ ├── rag_skill.py # 检索增强技能 │ └── batch_skill.py # 批量处理技能 ├── tools/ │ ├── __init__.py │ └── search_tool.py # 原子工具 ├── agents/ │ ├── dispatcher.py # 技能路由 Agent │ └── supervisor.py # 多 Agent 协作编排 ├── tests/ ├── main.py # 本地入口 ├── api_server.py # FastAPI 接口 ├── requirements.txt └── .env这个结构的好处是新加一个技能只需要在skills下新增文件并在注册中心登记不需要改动 Agent 主流程。这一点对后面批量任务和接口接入至关重要。4.3 定义 BaseSkill 接口这是整套架构的地基。所有技能都要实现同一个接口。# skills/base.py from abc import ABC, abstractmethod from typing import Any, Dict class BaseSkill(ABC): 技能接口每个技能都要能描述自己、校验参数并执行。 name: str description: str parameters: Dict[str, Any] {} abstractmethod def execute(self, context: Dict[str, Any]) - Dict[str, Any]: 执行技能返回结构化结果。 raise NotImplementedError三个字段一个方法的定义很关键。name是技能唯一标识description会被上层 Agent 用来做路由parameters描述调用方需要传入的字段execute接收统一格式的 context返回统一格式的结果。这套接口不是某个官方包的强制要求而是从 LangChain Tool 抽象里抽取的通用模式。生产环境可以直接用tool装饰器或 LangGraph 节点替代但设计思路一致。5. 12 个案例逐一拆解 Agent Skills 架构下面这 12 个案例从最简单写到最接近生产环境。每个案例都会说明要解决什么问题、代码怎么写、怎么判断成功。5.1 案例一最小 Agent 闭环第一个案例先跑通“用户输入 → Skill → 模型输出”的最小链路。这个案例的主要目的是验证环境没问题以及证明 Skill 封装之后调用方感知不到内部细节。# skills/chat_skill.py from langchain_openai import ChatOpenAI from skills.base import BaseSkill class ChatSkill(BaseSkill): name chat description 通用对话技能适合闲聊和简单问题回答 parameters {question: {type: string}} def __init__(self, llmNone): self.llm llm or ChatOpenAI(modelgpt-4o-mini, temperature0) def execute(self, context): question context.get(question) response self.llm.invoke(question) return {reply: response.content}使用方法skill ChatSkill() result skill.execute({question: LangChain 的 Skills 是什么}) print(result[reply])判断成功的标准能在终端看到模型返回一段合理回答如果调用失败优先检查 API Key 和网络代理配置。5.2 案例二技能注册中心当技能数量变多Agent 需要一个统一的地方来登记、发现和获取技能。注册中心就是“技能仓库”。# skills/registry.py class SkillRegistry: def __init__(self): self._skills {} def register(self, skill: BaseSkill): self._skills[skill.name] skill def get(self, name: str): return self._skills.get(name) def list_skills(self): return [ {name: skill.name, description: skill.description} for skill in self._skills.values() ]可以再用一个简单装饰器简化注册流程registry SkillRegistry() def register_skill(skill_cls): registry.register(skill_cls()) return skill_cls register_skill class ChatSkill(BaseSkill): ...注册之后上层 Agent 只要调用registry.list_skills()就能拿到所有可用的技能列表再也不用在代码里写死。判断成功标准多个技能注册后列表能完整输出且技能名不重复。5.3 案例三记忆型技能对话类 Agent 没有记忆时每一轮都是“失忆”状态。这个案例把历史消息保存到 Skill 内部并在下一次调用时拼进提示词。# skills/memory_skill.py class MemorySkill(BaseSkill): name memory_chat description 带记忆的多轮对话技能适合客服、助手等连续对话场景 parameters {question: {type: string}, user_id: {type: string}} def __init__(self, llm, max_history6): self.llm llm self.max_history max_history self.memories {} # user_id - list def execute(self, context): user_id context.get(user_id, default) history self.memories.setdefault(user_id, []) prompt self._build_prompt(history, context[question]) reply self.llm.invoke(prompt).content history.append({role: user, content: context[question]}) history.append({role: assistant, content: reply}) if len(history) self.max_history: del history[: len(history) - self.max_history] return {reply: reply} def _build_prompt(self, history, question): history_text \n.join(f{item[role]}: {item[content]} for item in history) return f对话历史\n{history_text}\n\n用户{question}\n助手这个案例的重点是记忆不是 Agent 的专属能力它完全可以被封装成一个 Skill。这样新的客服场景只需要组合记忆技能和问答技能不用重写逻辑。判断成功标准连续多轮对话后Agent 能正确回答涉及历史上下文的问题比如“我刚才问过什么”。5.4 案例四多技能路由现实中一个 Agent 通常有多个技能模型要先判断用户到底要调用哪个再传入参数。# agents/dispatcher.py def dispatch(user_input, skills_meta, llm): skill_list \n.join( f- {m[name]}: {m[description]} for m in skills_meta ) prompt f根据用户需求选择最合适的技能。 可选技能 {skill_list} 用户需求{user_input} 只输出技能名不要解释。 return llm.invoke(prompt).content.strip()路由逻辑并不复杂核心依赖两点技能描述足够准确模型输入输出格式足够简单。实际测试时无论选哪种技能都要在代码里加入“未匹配到任何技能”的兜底分支。这个案例也解释了为什么description这么重要模型不是靠代码逻辑选技能而是靠文字描述做语义匹配。如果你写“适合处理天气问题”但技能内部其实只能查不同城市的温度那模型的匹配准确率就会下降。5.5 案例五检索增强技能RAG Skill知识库问答是 Agent 最常见的落地场景。把检索增强问答封装成 SkillRAG 就从一个流程变成一个可以复用的能力。# skills/rag_skill.py class RagSkill(BaseSkill): name rag_qa description 基于知识库文档回答问题适合产品文档、规章制度、常见问题 parameters {question: {type: string}, top_k: {type: integer}} def __init__(self, retriever, llm): self.retriever retriever self.llm llm def execute(self, context): docs self.retriever.invoke(context[question]) top_k context.get(top_k, 4) context_text \n\n.join(d.page_content for d in docs[:top_k]) prompt f根据资料回答问题\n{context_text}\n\n问题{context[question]} return {reply: self.llm.invoke(prompt).content}这里的retriever可以是向量数据库检索器也可以是 Elasticsearch 检索器甚至是简单的关键词搜索。因为被封装在 Skill 内部上层 Agent 不需要关心检索细节。实战中需要重点观察三个参数检索条数top_k、上下文最大 token 限制、以及文档切分长度。检索条数太大模型容易抓不到重点太小信息不全。建议先小批量测试并观察输出质量再逐步调整。5.6 案例六批量文本处理技能单个技能处理单条数据简单但生产场景经常是“一批文章要总结”“一批合同要提取关键字段”。批量技能把循环和并发收敛到一个入口。# skills/batch_skill.py from concurrent.futures import ThreadPoolExecutor class BatchSummarySkill(BaseSkill): name batch_summary description 批量总结多段文本返回 items 数组 parameters {texts: {type: array}, max_workers: {type: integer}} def __init__(self, llm): self.llm llm def execute(self, context): texts context[texts] max_workers context.get(max_workers, 4) with ThreadPoolExecutor(max_workersmax_workers) as pool: results list(pool.map(self._sum_one, texts)) return {items: results} def _sum_one(self, text): return self.llm.invoke(f用一句话总结这段文本{text}).content这个案例的关键点是把“单条处理”和“批量调度”分离。_sum_one只负责单条逻辑execute只负责并发。后续想要加日志、重试、限流只需要改批量层。判断成功标准传入 5 到 10 条文本返回结果数组与输入顺序一致。如果结果乱序说明并发写回方式有问题在真实场景里会造成数据错位。5.7 案例七API 接入技能企业系统里大量数据通过 HTTP API 暴露。把这个能力封装成 SkillAgent 不需要自己拼请求、处理超时、解析响应。# skills/api_skill.py import requests class WeatherSkill(BaseSkill): name weather_query description 查询指定城市的实时天气 parameters {city: {type: string}} def execute(self, context): city context[city] url fhttps://api.example.com/weather?city{city} resp requests.get(url, timeout10) resp.raise_for_status() return {weather: resp.json()}需要注意这里api.example.com是演示地址实际使用时要替换成你自己的接口。调用外部接口前必须确认接口调用权限、频控限制和数据使用边界。如果是公司内部系统要把敏感字段做脱敏处理后再交给模型。5.8 案例八带并发控制的批量技能批量任务最怕“一次性把所有并发都打出去”大模型 API 通常有速率限制直接并发几十个请求很容易触发限流。# skills/rate_limited_skill.py import asyncio class RateLimitedSkill(BaseSkill): def __init__(self, llm, max_concurrency3, retries2): self.sem asyncio.Semaphore(max_concurrency) self.llm llm self.retries retries async def execute(self, context): async with self.sem: for attempt in range(self.retries): try: return {reply: await self.llm.ainvoke(context[question])} except Exception: if attempt self.retries - 1: raisemax_concurrency3表示最多同时 3 个请求。实际值要根据你使用的模型服务商限制来定。如果 API 返回 429 限流错误优先降低并发数而不是增加重试次数。这个案例的工程价值很明显批量任务从“一把梭”变成“有节流、有重试”系统的稳定性会立刻上一个台阶。5.9 案例九人工审批技能涉及到发邮件、转账、删数据、修改线上配置这些高风险操作Agent 不能直接执行必须先经过人工确认。# skills/approval_skill.py class ApprovalSkill(BaseSkill): name approval description 高风险动作前需要人工确认例如发送邮件、修改数据 parameters {action: {type: string}} def execute(self, context): action context[action] print(f请确认是否执行{action}) decision input(confirm/abort: ).strip().lower() if decision ! confirm: return {status: aborted, action: action} return {status: approved, action: action}生产环境不太可能用input()做审批而是通过消息队列、企业微信/钉钉审批流或者 LangGraph 的中断机制等人力介入方式。但思路一致Agent 在危险动作前停下等人说“通过”再继续。合规提醒涉及资金、隐私、版权或个人敏感信息的操作必须有明确的授权链路。不要为了让 Agent 流畅运行就跳过审批。5.10 案例十多 Agent 协作技能业务复杂到一定程度单个 Agent 无法覆盖所有领域就需要多个专业化 Agent 协作。主管 Agent 负责拆任务子 Agent 负责执行。# agents/supervisor.py class SupervisorAgent: def __init__(self, llm, sub_agents): self.llm llm self.sub_agents sub_agents async def run(self, task): routing_prompt ( f把任务分发给最合适的子Agent{task}\n f子Agent{list(self.sub_agents)} ) chosen (await self.llm.ainvoke(routing_prompt)).content.strip() agent self.sub_agents[chosen] return await agent.run(task)注意这里的多个子 Agent 可能各自又有自己的 Skill 和 Tool。从架构上看Agent 本身也可以被看成是一个可组合的 Skill。这种递归式设计是后来复杂业务能够扩展的基础。在多 Agent 协作场景下首先要避免“套娃式调度”主管 Agent 又把任务抛回给主管 Agent形成死循环。最好在编排层记录调用深度超过预设层数就强制返回。5.11 案例十一可观测性与追踪线上 Agent 不能靠“感觉”判断好坏必须有日志和追踪。这个案例用一个装饰器给技能执行加上耗时和状态记录。# skills/observability.py import logging import time logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) def log_execution(fn): def wrapper(self, context): start time.time() try: result fn(self, context) logging.info(fskill{self.name} statussuccess cost{time.time()-start:.2f}s) return result except Exception as e: logging.error(fskill{self.name} statuserror error{e}) raise return wrapper使用的时候class ChatSkill(BaseSkill): log_execution def execute(self, context): ...记录的数据不止耗时还可以记录输入参数大小、模型调用次数、token 消耗、是否命中缓存。如果团队已经接入 LangSmith 或 OpenTelemetry可以在这个装饰器里上报 trace。日志越完整后续排查批量任务卡住或技能选错原因就越快。5.12 案例十二生产级部署 API最后一个案例把整套 Skill Agent 封装成一个 REST API。这样前端、自动化脚本、其他后端服务都能调用也方便批量任务接入。# api_server.py from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel from skills.registry import registry app FastAPI() tasks {} class TaskRequest(BaseModel): skill_name: str context: dict def run_task(task_id: str, skill_name: str, context: dict): tasks[task_id] {status: running, result: None} skill registry.get(skill_name) if not skill: tasks[task_id] {status: error, error: skill not found} return try: result skill.execute(context) tasks[task_id] {status: done, result: result} except Exception as e: tasks[task_id] {status: error, error: str(e)} app.post(/tasks) def create_task(body: TaskRequest, background: BackgroundTasks): import uuid task_id str(uuid.uuid4()) tasks[task_id] {status: queued} background.add_task(run_task, task_id, body.skill_name, body.context) return {task_id: task_id} app.get(/tasks/{task_id}) def get_task(task_id: str): return tasks.get(task_id, {status: not_found}) app.get(/skills) def list_skills(): return registry.list_skills()启动服务uvicorn api_server:app --host 127.0.0.1 --port 8000这个 API 有两个端点POST /tasks用来创建任务GET /tasks/{task_id}用来查询任务状态。这种异步任务模式比同步返回更适合耗时较长的 Agent 任务前端可以轮询状态批量调度系统也能通过任务 ID 关联后续结果。6. 批量任务与 API 集成6.1 任务状态设计批量任务的第一个问题是“任务到底跑到哪了”。建议每个任务至少包含以下状态状态含义queued已接收等待执行running正在执行done执行完成结果已生成error执行失败附带错误信息not_found任务 ID 不存在或已被清理上面的 FastAPI 示例已经实现了这套状态流转。生产环境还需要考虑状态持久化比如把任务信息写入 Redis 或数据库否则服务重启后任务状态就丢了。6.2 批量任务调用示例调用批量技能时可以先创建大量任务然后用脚本轮询状态。# 创建一个批量总结任务 curl -X POST http://127.0.0.1:8000/tasks \ -H Content-Type: application/json \ -d {skill_name:batch_summary,context:{texts:[文本一,文本二,文本三]}} # 查询任务状态 curl http://127.0.0.1:8000/tasks/task_idPython 轮询示例import time import requests def wait_task(base_url, task_id, timeout600, interval2): start time.time() while time.time() - start timeout: resp requests.get(f{base_url}/tasks/{task_id}, timeout5) status resp.json()[status] if status in (done, error): return resp.json() time.sleep(interval) return {status: timeout}6.3 失败重试建议批量任务失败很常见重试时要避免无脑重试。建议按三类情况区分失败类型处理方式网络超时指数退避重试间隔 1s、2s、4sAPI 限流 429降低并发数等待一段时间后再试参数错误/鉴权失败不要重试直接记录失败原因模型输出格式错误重新生成一次最多重试 2 次批量处理成本控制也要有数。如果每次调用都传完整上下文token 消耗会非常快。建议在批量任务中加入 token 估算和成本统计超出阈值时自动降速。7. 性能与资源占用观察7.1 主要性能瓶颈LangChain 本身只是框架真正的性能瓶颈几乎都在大模型推理端。文本模型常见的瓶颈包括 prompt 长度、生成长度、并发数和服务商限流。CPU 可以跑小模型但生成速度会明显低于 GPU延迟取决于模型权重和量化方式。7.2 该观察哪些指标指标观察方式说明单任务耗时技能日志从用户输入到返回结果的完整耗时模型调用延迟模型服务日志一次 LLM 请求的响应时间token 消耗API 返回用量字段prompt 和 completion 分开统计并发成功率日志统计批量任务的失败率和重试率队列积压数任务状态接口判断调度是否需要扩容7.3 如何优化性能第一控制上下文长度。批量总结场景优先只传相关段落不要每次把整个原文都塞进模型。第二使用缓存。相同或相似问题可以命中语义缓存大幅减少模型调用。第三选择更小的模型做简单任务比如技能路由用轻量模型复杂推理用强模型。第四合理设置并发数既能压满吞吐又不会触发限流。这些优化不一定需要复杂框架先把日志打出来再根据真实的耗时和 token 数据做决定比盲目调参靠谱得多。8. 常见问题与排查方法问题现象可能原因排查方式解决方案安装依赖报错Python 版本不匹配或依赖冲突检查 Python 版本和 pip list使用虚拟环境按当前官方文档安装依赖模型调用超时网络慢或服务商限流查看模型服务响应码和日志增加超时和重试降低并发Agent 选错技能技能描述写得不清楚打印路由输入的 prompt 和路由结果优化 description 和参数示例返回内容不是 JSON模型输出不稳定查看原始输出使用结构化输出方法增加输出校验上下文超长多轮历史或检索内容过多统计每次请求的 token 数量截断历史减少检索条数压缩文本批量任务卡住并发过高或 API 限流查看任务状态和日志调低 max_concurrency增加指数退避FastAPI 服务内存居高不下每次请求创建模型连接观察进程内存变化复用模型实例使用连接池数据泄露或未授权使用没有做权限校验检查接口访问日志增加鉴权敏感操作走人工审批LangGraph 节点状态异常状态键名不一致检查图编译和状态更新逻辑统一状态 Schema加类型校验