共计 1484 个字符,预计需要花费 4 分钟才能阅读完成。
LangChain 技能模块基础
LangChain 的技能(Skill)是构建 AI 代理的核心组件,可以理解为一个个可复用的功能单元。每个技能封装了特定能力,比如文本处理、API 调用或数据转换。通过组合不同技能,我们可以像搭积木一样构建复杂的 AI 代理。

- 技能的本质:包含输入参数处理、核心逻辑、输出格式化的 Python 类
- 典型应用场景:问答系统需要组合 ” 搜索 ”、” 摘要 ” 和 ” 回答生成 ” 三个技能
- 核心优势:标准化接口使得不同开发者创建的技能可以无缝组合
新手三大痛点解析
- 技能组合顺序错误:比如先执行需要前置数据的技能,导致流程中断
- 参数传递混乱:未正确处理技能间的输入输出映射关系
- 异常处理缺失:网络请求类技能没有设置超时和重试机制
实战:创建天气预报技能
from langchain.skills import BaseSkill
from typing import Dict, Any
import requests
class WeatherSkill(BaseSkill):
"""获取指定城市的天气信息"""
def __init__(self):
super().__init__(
name="weather",
description="查询城市天气",
input_schema={"city": "string"}, # 定义输入格式
output_schema={"weather": "string"}
)
async def execute(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
city = inputs["city"]
# 实际项目应该用环境变量管理 API 密钥
api_url = f"https://api.weatherapi.com/v1/current.json?key=YOUR_KEY&q={city}"
try:
response = requests.get(api_url, timeout=5)
data = response.json()
return {"weather": data["current"]["condition"]["text"]}
except Exception as e:
return {"error": str(e)}
性能优化三原则
- 异步调用 :用
async/await避免 I / O 阻塞,特别是需要调用外部 API 时 - 缓存策略:对频繁查询且结果变化不频繁的技能添加内存缓存
- 批量处理:设计支持批量输入的技能减少网络请求次数
避坑指南
- 技能命名冲突:为自定义技能添加项目前缀(如
myproject_weather) - 未处理空输入:在所有技能入口添加参数验证逻辑
- 忽略技能版本:当升级技能时,保持向后兼容或明确版本控制
- 过度复杂技能:单个技能应专注单一功能,复杂逻辑拆分为子技能
- 缺少日志记录:关键步骤添加日志,便于调试技能链
理解技能链原理
技能链 (Skill Chain) 实质是有向无环图(DAG),LangChain 会自动解析技能间的依赖关系。例如当技能 B 需要技能 A 的输出时,框架会确保执行顺序为 A→B。依赖管理通过两种方式实现:
- 显式声明:在技能定义时指定前置依赖
- 自动推断:通过输入输出参数名自动匹配
复合技能实践任务
尝试构建一个组合技能:
- 调用天气 API 获取英文天气描述
- 将结果通过翻译技能转为中文
- 最终输出格式:
{"city": "北京", "weather_zh": "晴朗"}
提示步骤:
- 先分别实现独立的天气和翻译技能
- 创建新技能组合两者,注意处理错误情况
- 测试不同城市的天气查询
通过这个练习,你会掌握技能组合的核心方法。当遇到问题时,记得查看 LangChain 的调试日志,它通常会给出明确的依赖解析信息。
正文完
