进阶主题与部署

7.1 调试与追踪

LLM 应用的调试比传统软件更具挑战性，因为模型的输出具有不确定性。LangChain 提供了多种调试工具。

7.1.1 Callback 机制

LangChain 的 Callback 系统允许你在链执行过程中插入自定义逻辑：

from langchain_openai import ChatOpenAI
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import LLMResult
from typing import Any, List, Dict

# 自定义回调处理器
class DebugCallback(BaseCallbackHandler):
    def on_llm_start(self, serialized: Dict, prompts: List[str], **kwargs):
        print("🔵 LLM 开始调用")
        print(f"   提示词: {prompts[0][:100]}...")

    def on_llm_end(self, response: LLMResult, **kwargs):
        print("🟢 LLM 调用完成")
        print(f"   Token 使用: {response.llm_output.get('token_usage', {})}")

    def on_chain_start(self, serialized: Dict, inputs: Dict, **kwargs):
        print(f"🔵 Chain 开始: {serialized.get('name', 'unnamed')}")

    def on_chain_end(self, outputs: Dict, **kwargs):
        print("🟢 Chain 执行完成")

    def on_tool_start(self, serialized: Dict, inputs: Dict, **kwargs):
        print(f"⚡ 工具开始: {serialized.get('name', 'unnamed')}")

    def on_tool_end(self, output: str, **kwargs):
        print(f"✅ 工具完成: {output[:50]}...")

# 使用回调
llm = ChatOpenAI(model="gpt-4o-mini", callbacks=[DebugCallback()])
response = llm.invoke("你好")
print(f"最终回复: {response.content}")

7.1.2 内置回调工具

from langchain_core.callbacks import StdOutCallbackHandler
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# StdOutCallbackHandler - 打印到标准输出
handler = StdOutCallbackHandler()

llm = ChatOpenAI(model="gpt-4o-mini")
prompt = PromptTemplate.from_template("翻译成{lang}：{text}")

chain = LLMChain(llm=llm, prompt=prompt, callbacks=[handler])

result = chain.invoke({"lang": "日语", "text": "Hello, World!"})

7.2 LangSmith 使用指南

LangSmith 是 LangChain 官方提供的调试、追踪和评估平台，帮助你深入了解 LLM 应用的行为。

7.2.1 快速集成

# 1. 安装
pip install langsmith

# 2. 设置环境变量
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"  # 从 langsmith.ai 获取
os.environ["LANGCHAIN_PROJECT"] = "my-langchain-app"  # 项目名称

# 3. 使用（自动追踪）
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke("你好，请介绍一下自己")

7.2.2 添加数据集和评估

from langchain_openai import ChatOpenAI
from langchain.smith import RunEvalConfig, run_on_dataset
from langchain_core.output_parsers import StrOutputParser

# 定义评估配置
eval_config = RunEvalConfig(
    evaluators=[
        # 检查输出是否包含关键词
        "qa",
        # 自定义评估器
        {
            "criteria": {
                "helpfulness": {
                    "scale": 1,
                    "prompt": "评估回答是否有帮助..."
                }
            }
        }
    ]
)

# 创建 Chain
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()

# 在数据集上运行评估
run_on_dataset(
    client=None,
    dataset_name="qa-test-dataset",
    llm_or_chain_factory=chain,
    evaluation=eval_config,
)

7.3 性能优化

LLM 应用通常延迟较高，以下是常用的优化策略。

7.3.1 缓存优化

from langchain_openai import ChatOpenAI
from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

# 启用 LLM 响应缓存
set_llm_cache(InMemoryCache())

llm = ChatOpenAI(model="gpt-4o-mini")

# 相同输入直接返回缓存结果
response1 = llm.invoke("什么是人工智能")
response2 = llm.invoke("什么是人工智能")  # 几乎瞬时返回

7.3.2 并行处理

from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage
from langchain.utilities import SerpAPIWrapper
import asyncio

llm = ChatOpenAI(model="gpt-4o-mini")

# 并行执行多个独立任务
async def parallel_tasks():
    # 创建多个任务
    tasks = [
        llm.agenerate([[HumanMessage(content="解释 Python")]]),
        llm.agenerate([[HumanMessage(content="解释 JavaScript")]]),
        llm.agenerate([[HumanMessage(content="解释 Rust")]]),
    ]
    
    # 并行执行
    results = await asyncio.gather(*tasks)
    return results

# 运行
results = asyncio.run(parallel_tasks())

7.3.3 Token 控制

7.4 生产环境部署

7.4.1 Docker 部署

FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 设置环境变量
ENV PYTHONUNBUFFERED=1

# 启动命令
CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - LANGCHAIN_TRACING_V2=true
      - LANGCHAIN_API_KEY=${LANGCHAIN_API_KEY}
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

7.4.2 Kubernetes 部署

apiVersion: apps/v1
kind: Deployment
metadata:
  name: langchain-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: langchain-api
  template:
    metadata:
      labels:
        app: langchain-api
    spec:
      containers:
      - name: api
        image: your-registry/langchain-api:latest
        ports:
        - containerPort: 8000
        env:
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: api-secrets
              key: openai-api-key
        resources:
          requests:
            memory: "512Mi"
            cpu: "250m"
          limits:
            memory: "2Gi"
            cpu: "1000m"

7.4.3 API 服务封装

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

app = FastAPI(title="LangChain API", version="1.0")

# 初始化 Chain
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
prompt = ChatPromptTemplate.from_template("{question}")
chain = prompt | llm | StrOutputParser()

class Question(BaseModel):
    question: str

@app.get("/health")
async def health():
    return {"status": "ok"}

@app.post("/ask")
async def ask(question: Question):
    try:
        result = chain.invoke({"question": question.question})
        return {"answer": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

7.5 安全与最佳实践

7.5.1 敏感信息处理

import os
from dotenv import load_dotenv

# 从环境变量读取密钥（安全）
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

# 输入验证和清理
import re
def sanitize_input(user_input: str) -> str:
    """清理用户输入，防止注入"""
    # 移除潜在的危险字符
    cleaned = re.sub(r'[\x00-\x1f\x7f-\x9f]', '', user_input)
    # 限制长度
    return cleaned[:2000]

# 使用清理后的输入
def create_safe_prompt(user_input: str) -> str:
    safe_input = sanitize_input(user_input)
    return f"用户问题: {safe_input}\n请简要回答。"

7.5.2 速率限制

from fastapi import FastAPI, Request
from slowapi import Limiter
from slowapi.util import get_remote_address

app = FastAPI()
limiter = Limiter(key_func=get_remote_address)

@app.post("/ask")
@limiter.limit("10/minute")  # 每分钟10次请求
async def ask(request: Request):
    # 处理请求
    pass

# 或者使用 token bucket 算法
from ratelimit import limits

@limits(calls=10, period=60)  # 60秒内最多10次
def call_llm(prompt: str):
    # 调用 LLM
    pass

7.5.3 最佳实践清单

7.6 本章小结

本章学习了 LangChain 进阶主题与生产环境部署的核心内容：