LangChain实战入门：从零构建你的第一个AI技能链

2次阅读

没有评论

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

作为一名 Python 开发者，当你第一次尝试构建基于大语言模型 (LLM) 的应用时，可能会遇到各种挑战。本文将带你从零开始，使用 LangChain 框架构建一个完整的天气查询技能链，涵盖从基础概念到生产环境优化的全流程。

在传统 LLM 应用开发中，我们常遇到几个典型问题：

接口耦合严重：业务逻辑与 API 调用代码混杂在一起
状态管理混乱：对话历史和上下文难以维护
扩展成本高：每增加一个新功能都需要重写大量代码

相比之下，LangChain 框架提供了模块化的设计思路，将复杂流程分解为可复用的组件(Chain/Tool/Agent)，大大提升了开发效率。

让我们通过一个简单例子对比两种方式。假设我们要实现天气查询功能：

直接调用 OpenAI API

import openai

def get_weather(city):
    # 这里需要自己处理 API 调用、错误重试、结果解析等
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": f"{city}的天气如何？"}]
    )
    return response.choices[0].message.content

使用 LangChain 实现

from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template("{city}的天气如何？")
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run(city="北京")

可以看到，LangChain 版本更简洁且易于维护。当需求变更时，我们只需要修改相应的组件，而不必重写整个流程。

首先，我们创建一个专门处理天气查询的 Tool(工具)：

from typing import Optional
from langchain.tools import BaseTool
from pydantic import BaseModel, Field

class WeatherToolInput(BaseModel):
    city: str = Field(..., description="需要查询天气的城市名称")

class WeatherTool(BaseTool):
    name = "weather_tool"
    description = "查询指定城市的天气信息"
    args_schema = WeatherToolInput

    def _run(self, city: str) -> str:
        """同步执行天气查询"""
        # 这里实现实际的天气 API 调用
        api_key = self.metadata.get("api_key")
        if not api_key:
            raise ValueError("缺少 API 密钥")

        # 模拟 API 调用
        import requests
        try:
            response = requests.get(f"https://api.weather.com/v1/{city}",
                headers={"Authorization": f"Bearer {api_key}"}
            )
            response.raise_for_status()
            return f"{city}的天气是: {response.json()['weather']}"
        except Exception as e:
            return f"查询天气失败: {str(e)}"

    async def _arun(self, city: str) -> str:
        """异步执行天气查询"""
        # 异步版本的实现
        return await asyncio.get_event_loop().run_in_executor(None, self._run, city)

这个工具类包含了几个关键点：

使用 Pydantic 模型定义输入参数
实现了同步和异步两种执行方式
包含基本的错误处理和 API 密钥验证

为了让 AI 能基于知识库回答更专业的问题，我们需要实现检索增强生成 (RAG) 流程：

flowchart TD
    A[用户问题] --> B[文档加载]
    B --> C[文本分割]
    C --> D[向量存储]
    D --> E[相似度检索]
    E --> F[生成回答]

对应代码实现：

from langchain.document_loaders import WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS

# 1. 文档加载
loader = WebBaseLoader("https://example.com/weather-docs")
documents = loader.load()

# 2. 文本分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
splits = text_splitter.split_documents(documents)

# 3. 向量存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(splits, embeddings)

# 4. 检索链
retriever = vectorstore.as_retriever()
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever
)

LangChain Expression Language (LCEL) 提供了一种声明式的方式来编排复杂流程：

from langchain.schema import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough

weather_prompt = ChatPromptTemplate.from_template(
    """ 根据以下信息回答天气问题:
{context}

问题: {question}"""
)

chain = ({"context": retriever, "question": RunnablePassthrough()}
    | weather_prompt
    | llm
    | StrOutputParser())

result = chain.invoke("北京明天会下雨吗？")

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def safe_weather_query(city: str) -> str:
    return WeatherTool().run(city)

永远不要将 API 密钥硬编码在代码中！推荐使用环境变量或密钥管理服务：

import os
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv("WEATHER_API_KEY")

可以添加简单的指标收集：

import time
from prometheus_client import Summary

REQUEST_TIME = Summary('request_processing_seconds', 'Time spent processing request')

@REQUEST_TIME.time()
def query_weather(city):
    start_time = time.time()
    result = chain.invoke(city)
    duration = time.time() - start_time
    logger.info(f"查询耗时: {duration:.2f}s")
    return result

随着技能链变得越来越复杂，如何管理不同版本的技能成为一个挑战。你可以考虑：