LangChain实战：如何用Skill机制解决复杂任务编排难题

1次阅读

没有评论

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

在构建复杂 AI 工作流时，传统的链式调用方式（如 LangChain 的原生 Chain）往往会遇到几个典型问题：

维护性差 ：当业务流程需要频繁调整时（比如电商客服场景中的多条件分支），需要重新设计整个链结构
调试困难 ：错误会沿着调用链向上传播，难以定位具体问题节点
复用率低 ：相似的业务逻辑（如商品查询、物流跟踪）无法跨流程共享

LangChain 的原生 Chain 和 Tool 虽然简单易用，但在复杂场景下存在明显局限：

扩展性对比 ：
Chain：修改需要重建整个流程
Skill：通过组合现有 Skill 即可实现新功能
调试复杂度 ：
Chain：需要跟踪整个执行链路
Skill：每个单元可独立测试
性能表现 ：
Chain：线性执行难以并行化
Skill：支持批量处理和预热

使用 pydantic 确保强类型约束：

from pydantic import BaseModel
from typing import Optional, Any

class SkillInput(BaseModel):
    context: dict
    params: Optional[dict] = None

class SkillOutput(BaseModel):
    result: Any
    metadata: dict
    success: bool

class BaseSkill:
    def __init__(self, name: str):
        self.name = name

    async def execute(self, input: SkillInput) -> SkillOutput:
        raise NotImplementedError

利用 LLM 实现智能路由决策：

from langchain.chat_models import ChatOpenAI

class Router:
    def __init__(self):
        self.llm = ChatOpenAI(temperature=0)
        self.skill_map = {}  # {'skill_name': skill_instance}

    async def route(self, user_input: str) -> str:
        prompt = f"""根据用户输入选择最合适的 Skill:\n\n"""
        # 添加可用的 skill 描述
        for name, skill in self.skill_map.items():
            prompt += f"- {name}: {skill.__doc__}\n"

        prompt += f"\n 用户输入: {user_input}\n 只返回 skill 名称"
        return await self.llm.apredict(prompt)

from functools import wraps
import asyncio
from typing import Callable, TypeVar

T = TypeVar('T')

def retry(max_attempts=3, delay=1):
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> T:
            last_error = None
            for attempt in range(1, max_attempts+1):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    if attempt < max_attempts:
                        await asyncio.sleep(delay)
            raise RuntimeError(f"After {max_attempts} attempts") from last_error
        return wrapper
    return decorator

对于初始化耗时的 Skill（如加载大模型），使用对象池管理：

from queue import Queue

class SkillPool:
    def __init__(self, skill_factory, pool_size=5):
        self.pool = Queue(maxsize=pool_size)
        for _ in range(pool_size):
            self.pool.put(skill_factory())

    def acquire(self):
        return self.pool.get()

    def release(self, skill):
        self.pool.put(skill)

通过组合多个用户请求实现批量处理：

async def batch_process(requests: List[SkillInput], skill: BaseSkill):
    # 假设 skill 支持批量处理
    batch_size = 10
    results = []
    for i in range(0, len(requests), batch_size):
        batch = requests[i:i+batch_size]
        results.extend(await skill.batch_execute(batch))
    return results

过粗：一个 Skill 处理多个不相关功能，违背单一职责原则
过细：每个简单操作都拆成 Skill，增加管理开销
黄金法则 ：
一个 Skill 应该对应业务领域中的一个完整操作
执行时间控制在 50ms-5s 范围内
输入 / 输出参数不超过 5 个

❌ 在 Skill 内部维护全局状态
❌ 通过类属性共享数据
✅ 正确的做法：
通过 SkillInput.context 传递必要状态
使用 Redis 等外部存储共享数据

使用 unittest.mock 隔离依赖：

from unittest.mock import MagicMock

def test_skill_with_mock():
    # 创建模拟依赖
    mock_llm = MagicMock()
    mock_llm.predict.return_value = "mock response"

    # 注入模拟对象
    skill = MySkill(llm=mock_llm)
    result = skill.execute(test_input)

    assert result.success
    mock_llm.predict.assert_called_once()

处理以下典型场景：

订单状态查询
退货申请处理
优惠券咨询

设计三个核心 Skill：

OrderSkill：
输入：订单号
输出：订单状态、物流信息
ReturnSkill：
输入：订单号 + 退货原因
输出：RMA 编号、退货指引
CouponSkill：
输入：用户 ID
输出：可用优惠券列表

# 初始化组件
router = Router()
router.skill_map = {"order": OrderSkill(),
    "return": ReturnSkill(),
    "coupon": CouponSkill()}

# 处理用户请求
async def handle_request(user_input: str):
    skill_name = await router.route(user_input)
    skill = router.skill_map[skill_name]

    # 构造输入（实际场景需要更复杂的解析）skill_input = SkillInput(context={"user_id": "123"},
        params={"query": user_input}
    )

    return await skill.execute(skill_input)

通过 Skill 机制，我们实现了：