LangChain新手入门：从零构建你的第一个技能链(Skill Chain)应用

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

为什么需要技能链(Skill Chain)

假设你要开发一个智能客服系统，用户问 ” 如何重置密码？” 时，传统代码可能直接返回预设答案。但真实场景需要：

1. 意图识别：判断用户是想重置密码而非其他操作
2. 信息检索：从知识库找到最新重置流程
3. 回复生成：组合成自然语言回答

用独立脚本实现这些功能会导致代码臃肿，而 LangChain 的模块化设计能像搭积木一样组合这些技能。

传统方案 vs LangChain

传统脚本的问题：

• 功能耦合：改检索逻辑可能影响回复生成
• 难扩展：新增技能要重构主流程
• 缺乏复用：不同项目重复造轮子

LangChain 的优势：

• 每个技能独立开发测试
• 通过标准接口组合（输入 / 输出统一格式）
• 内置常用组件（如 LLM 连接器、记忆模块）

实战：构建三技能链

1. 安装与环境

pip install langchain==0.0.198 python-dotenv

建议使用虚拟环境，并准备 .env 文件存放 API 密钥。

2. 基础技能实现

技能 1：文本清洗

from langchain.chains import TransformChain

def clean_text(inputs):
    text = inputs["text"]
    # 去特殊字符 + 转为小写
    clean = "".join(c for c in text if c.isalnum() or c.isspace()).lower()
    return {"cleaned_text": clean}

clean_chain = TransformChain(input_variables=["text"], 
    output_variables=["cleaned_text"],
    transform=clean_text
)

技能 2：关键词提取（使用虚拟示例）

from typing import Dict

def extract_keywords(inputs: Dict) -> Dict:
    from collections import Counter
    words = inputs["cleaned_text"].split()
    top_words = [w for w,_ in Counter(words).most_common(3)]
    return {"keywords": top_words}

keyword_chain = TransformChain(input_variables=["cleaned_text"],
    output_variables=["keywords"],
    transform=extract_keywords
)

技能 3：模板生成

def make_template(inputs: Dict) -> Dict:
    kw = ",".join(inputs["keywords"])
    return {"output": f"生成内容包含关键词: {kw}"}

template_chain = TransformChain(input_variables=["keywords"],
    output_variables=["output"],
    transform=make_template
)

3. 组合成链

from langchain.chains import SequentialChain

overall_chain = SequentialChain(chains=[clean_chain, keyword_chain, template_chain],
    input_variables=["text"],
    verbose=True  # 打印执行日志
)

# 测试运行
result = overall_chain({"text": "Hello World! This is a DEMO 123."})
print(result["output"])

4. 添加健壮性

错误处理：

from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def safe_extract(inputs):
    try:
        return extract_keywords(inputs)
    except Exception as e:
        print(f"Keyword 提取失败: {e}")
        return {"keywords": []}  # 返回默认值

日志记录：

import logging
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s',
    level=logging.INFO
)

# 在技能函数内添加:
logging.info(f"Processing text: {inputs['text'][:50]}...")

生产环境注意事项

数据格式标准化

• 每个技能的输入 / 输出建议用 Pydantic 模型验证
• 示例：

  from pydantic import BaseModel
  
  class SkillInput(BaseModel):
      text: str
      meta: dict = {}

超时控制

from langchain.chains import LLMChain
from concurrent.futures import TimeoutError

chain = LLMChain(...)
try:
    result = chain.run(
        input="...", 
        timeout=10  # 秒
    )
except TimeoutError:
    print("技能执行超时")

异步优化

import asyncio

async def async_run_chain():
    await asyncio.gather(chain1.arun(...),
        chain2.arun(...)
    )

完整示例

点击访问 Colab Notebook（包含可运行代码）

进阶思考

1. 动态路由：如何根据输入内容决定后续执行哪些技能？（如用户问天气则走 A 链，问商品走 B 链）
2. 性能监控：怎样实时跟踪每个技能的执行耗时和成功率？

通过这个简单的技能链，你已经体验了 LangChain 的核心价值——像流水线一样组合 AI 能力。接下来可以尝试接入真实 API（如 OpenAI），或探索更复杂的 Agent 模式。记住：好的技能链设计就像乐高，每个模块都应该独立且可替换。