LangChain调用Skill实战：如何高效构建模块化AI应用

1次阅读

共计 2963 个字符，预计需要花费 8 分钟才能阅读完成。

在传统 AI 应用开发中，我们经常会遇到紧耦合架构带来的各种问题。比如，每次开发新功能都需要从头开始写代码，导致大量重复劳动；调试困难，因为所有功能都混杂在一起，定位问题像大海捞针；还有更头疼的是，当需要更新某个功能时，可能会影响到其他完全不相关的部分。这些问题不仅降低了开发效率，还让维护变得异常困难。

正是这些痛点，让我们开始思考如何实现模块化开发。而 LangChain 的 Skill 调用机制，就像一股清流，为我们提供了一种高效解耦的方案。通过将每个功能封装成独立的 Skill，我们可以像搭积木一样构建复杂的 AI 应用，既提高了代码复用性，又降低了维护成本。

在决定使用 LangChain Skill 之前，我们需要清楚地了解它与直接调用模型 API 的区别。下面这个对比表格可以帮我们做出更明智的选择：

对比项	直接调用模型 API	LangChain Skill 调用
可维护性	低，代码分散在各处	高，功能模块化封装
扩展性	差，改动影响范围大	好，新增 Skill 不影响现有
复用性	需手动复制粘贴	直接导入调用
错误处理	各自实现，不统一	统一异常处理机制
性能优化	需单独考虑	可集中优化

从表格中可以看出，Skill 调用在多个维度上都优于直接调用 API，特别是在大型项目中，这种优势会更加明显。

使用 @skill 装饰器是注册 Skill 最简便的方式。下面是一个完整示例：

from langchain.skills import skill
from typing import Dict, Any

@skill(name="weather_query", description="查询城市天气信息")
def get_weather(city: str) -> Dict[str, Any]:
    """
    获取指定城市的天气信息
    :param city: 城市名称
    :return: 包含天气信息的字典
    """
    # 这里是模拟实现，实际项目中会调用天气 API
    return {
        "city": city,
        "temperature": "22°C",
        "condition": "晴天"
    }

关键点说明：

使用 @skill 装饰器时，必须指定 name 和description参数
函数参数建议添加类型标注
文档字符串要清晰描述功能

LangChain 提供的 LCEL（LangChain Expression Language）语法让链式调用变得非常简单：

from langchain.agents import AgentExecutor
from langchain.agents.skills import load_skills

# 加载已注册的 Skill
skills = load_skills(["weather_query", "time_converter"])

# 构建执行链
executor = AgentExecutor(
    skills=skills,
    verbose=True
)

# 执行链式调用
result = executor.run({"input": "先查询北京的天气，然后转换成纽约当地时间"})

from langchain.skills import skill
from langchain.agents import AgentExecutor
from langchain.agents.skills import load_skills
from typing import Dict, Any
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

@skill(name="weather_query", description="查询城市天气信息")
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def get_weather(city: str) -> Dict[str, Any]:
    """
    获取指定城市的天气信息
    :param city: 城市名称
    :return: 包含天气信息的字典
    :raises: ValueError 当城市不存在时
    """
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            # 实际项目中替换为真实 API 地址
            response = await client.get(f"https://api.weather.com/{city}")
            response.raise_for_status()
            return response.json()
    except httpx.HTTPStatusError as e:
        raise ValueError(f"获取天气失败: {e}")

# 初始化执行器
skills = load_skills(["weather_query"])
executor = AgentExecutor(
    skills=skills,
    max_execution_time=30,  # 设置超时时间
    verbose=True
)

try:
    result = await executor.arun({"input": "查询北京的天气"})
    print(result)
except Exception as e:
    print(f"执行失败: {e}")
    # 这里可以添加错误处理逻辑，如重试或降级处理

这段代码展示了几个重要实践：

使用 @retry 装饰器实现自动重试
添加了详细的异常处理
使用异步 HTTP 客户端提高性能
设置了合理的超时时间

当多个请求同时调用同一个 Skill 时，可能会遇到资源竞争问题。比如，某个 Skill 需要调用外部 API，而该 API 有调用频率限制。

from asyncio import Semaphore

# 全局并发控制
WEATHER_SEMAPHORE = Semaphore(5)  # 限制最多 5 个并发

@skill(name="weather_query")
async def get_weather(city: str):
    async with WEATHER_SEMAPHORE:
        # 实际业务逻辑
        return await fetch_weather_data(city)

这种方案简单有效，可以防止系统过载。对于更复杂的场景，可以考虑使用令牌桶等算法。