LangChain 应用开发实战 · 运维手册

适用场景：基于 LLM 构建运维助手、知识库问答、自动化工单系统等 AI 应用。

---

1. LangChain 核心概念

1.1 架构概览

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  LLM        │     │  Memory     │     │  Tools      │
│  (Ollama)   │     │  (Redis)    │     │  (API/CLI)  │
└──────┬──────┘     └──────┬──────┘     └──────┬──────┘
       │                   │                   │
       └───────────────────┼───────────────────┘
                           │
                    ┌──────▼──────┐
                    │   Chain     │
                    │  (编排器)    │
                    └──────┬──────┘
                           │
                    ┌──────▼──────┐
                    │  Output     │
                    │  (结果)      │
                    └─────────────┘

1.2 核心组件

组件	作用	典型用法
LLM	调用大模型	ChatOllama, ChatOpenAI
PromptTemplate	提示词模板	固定结构 + 变量填充
OutputParser	解析输出	JSON, String
Chain	链式调用	LLMChain, SequentialChain
Memory	会话历史	ConversationBufferMemory
Tool	外部工具	Python 函数封装
Agent	自主决策	AgentExecutor

---

2. 最小可用 Chain

2.1 基础聊天链

# basic_chain.py
from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate
from langchain_community.llms import Ollama

# 初始化 LLM
llm = Ollama(model="qwen2.5:7b", temperature=0.7)

# 定义 Prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个运维专家助手。"),
    ("user", "{question}")
])

# 创建 Chain
chain = LLMChain(llm=llm, prompt=prompt)

# 执行
result = chain.invoke({"question": "Kubernetes Pod 处于 Pending 状态怎么办？"})
print(result["text"])

2.2 带结构化输出的 Chain

# structured_chain.py
from langchain.output_parsers import JsonOutputParser
from langchain.chains import LLMChain

# JSON 解析器
parser = JsonOutputParser()

prompt = ChatPromptTemplate.from_template(
    "分析以下运维问题，输出 JSON：\n{question}\n\n{format_instructions}"
)

chain = LLMChain(llm=llm, prompt=prompt)

result = chain.invoke({
    "question": "Nginx 502 错误",
    "format_instructions": parser.get_format_instructions()
})

# 输出结构化 JSON
print(result["text"])

---

3. Memory 会话管理

3.1 内存存储（开发环境）

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 每次对话自动保存历史
chain = LLMChain(llm=llm, prompt=prompt, memory=memory)

3.2 Redis 存储（生产环境）

from langchain_community.storage import RedisStore
from langchain_community.chat_message_histories import RedisChatMessageHistory

def get_memory(session_id: str):
    """获取会话历史"""
    return RedisChatMessageHistory(
        session_id=session_id,
        url="redis://localhost:6379/0"
    )

# 使用时
history = get_memory("user_123")
chain = LLMChain(llm=llm, prompt=prompt, memory=history)

3.3 Memory 清理策略

# 设置 TTL 避免泄漏
import redis

r = redis.Redis()
r.expire(f"chat_history:user_123", 86400)  # 24 小时过期

# 定期清理
def cleanup_expired_memories():
    keys = r.keys("chat_history:*")
    for key in keys:
        if r.ttl(key) < 0:
            r.delete(key)

---

4. Tools 工具集成

4.1 定义 Tool

from langchain.tools import Tool

def check_service(service_name: str) -> str:
    """检查服务状态"""
    import subprocess
    result = subprocess.run(
        ["systemctl", "status", service_name],
        capture_output=True, text=True
    )
    return result.stdout[:1000]

def query_metrics(metric: str) -> str:
    """查询 Prometheus 指标"""
    import requests
    resp = requests.get(
        f"http://prometheus:9090/api/v1/query?query={metric}"
    )
    return resp.text[:2000]

# 注册 Tools
tools = [
    Tool(
        name="check_service",
        func=check_service,
        description="检查 Linux 服务状态"
    ),
    Tool(
        name="query_metrics",
        func=query_metrics,
        description="查询 Prometheus 指标"
    )
]

4.2 Tool 安全沙箱

import subprocess
import time

ALLOWED_COMMANDS = ["systemctl", "docker", "kubectl", "curl"]
MAX_TIMEOUT = 30

def safe_tool_call(func_name: str, **kwargs) -> dict:
    """安全执行 Tool"""
    # 检查命令白名单
    if func_name not in ALLOWED_COMMANDS:
        return {"error": f"Command {func_name} not allowed"}

    # 执行并限制超时
    try:
        result = subprocess.run(
            kwargs.get("cmd", ""),
            shell=True,
            capture_output=True,
            text=True,
            timeout=MAX_TIMEOUT
        )
        return {
            "stdout": result.stdout[:5000],
            "stderr": result.stderr[:5000],
            "returncode": result.returncode
        }
    except subprocess.TimeoutExpired:
        return {"error": "Command timed out"}

---

5. Agent 智能体

5.1 ReAct Agent

from langchain.agents import create_react_agent, AgentExecutor

# 创建 Agent
agent = create_react_agent(
    llm=llm,
    tools=tools,
    prompt=react_prompt  # ReAct 提示词模板
)

executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,
    handle_parsing_errors=True
)

# 执行
result = executor.invoke({
    "input": "Nginx 返回 502，请帮我诊断"
})

5.2 Agent 工作流

用户输入 → Agent 规划 → 调用 Tool → 观察结果 → 继续规划 → 输出答案
                ↑                                      ↓
                └────────── 循环（最多 N 次）──────────┘

---

6. RAG Pipeline

6.1 完整 RAG 链

from langchain.chains import RetrievalQA
from langchain.vectorstores import Chroma
from langchain.embeddings import OllamaEmbeddings

# 初始化向量库
embeddings = OllamaEmbeddings(model="bge-m3")
vectorstore = Chroma(
    persist_directory="/data/chroma/ops-manuals",
    embedding_function=embeddings
)

# 检索器
retriever = vectorstore.as_retriever(
    search_kwargs={"k": 5, "score_threshold": 0.6}
)

# RAG Chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=retriever,
    return_source_documents=True
)

# 查询
result = qa_chain.invoke("Nginx 502 错误如何排查？")
print(result["result"])
print(result["source_documents"])

6.2 检索优化

# 多路检索：向量 + 关键词
from langchain.retrievers import EnsembleRetriever
from langchain.retrievers import BM25Retriever

vector_retriever = vectorstore.as_retriever()
bm25_retriever = BM25Retriever.from_documents(documents)

ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.7, 0.3]
)

---

7. 错误处理与重试

7.1 重试机制

from langchain.globals import set_verbose, set_debug
from tenacity import retry, wait_exponential, stop_after_attempt

@retry(
    wait=wait_exponential(multiplier=1, min=4, max=10),
    stop=stop_after_attempt(3)
)
def call_llm_with_retry(prompt: str) -> str:
    """带重试的 LLM 调用"""
    return llm.invoke(prompt)

7.2 降级策略

def robust_chain_call(question: str) -> dict:
    """健壮调用：失败降级"""
    try:
        result = chain.invoke({"question": question})
        return {"status": "success", "result": result}
    except Exception as e:
        # 降级到简单模式
        try:
            result = simple_llm.invoke(question)
            return {"status": "degraded", "result": result}
        except:
            return {"status": "error", "result": "服务暂时不可用"}

---

8. 监控与日志

8.1 调用链追踪

import uuid
import logging

logger = logging.getLogger(__name__)

def trace_chain_call(question: str):
    trace_id = str(uuid.uuid4())[:8]
    logger.info(f"[TRACE:{trace_id}] 开始处理: {question}")

    try:
        result = chain.invoke({"question": question})
        logger.info(f"[TRACE:{trace_id}] 成功，耗时: {elapsed}ms")
        return result
    except Exception as e:
        logger.error(f"[TRACE:{trace_id}] 失败: {e}")
        raise

8.2 指标收集

from prometheus_client import Counter, Histogram

REQUEST_COUNT = Counter('llm_requests_total', 'LLM request count', ['model', 'status'])
REQUEST_LATENCY = Histogram('llm_request_latency_seconds', 'LLM request latency')

@REQUEST_LATENCY.time()
def tracked_call(question: str):
    try:
        result = chain.invoke({"question": question})
        REQUEST_COUNT.labels(model='qwen2.5', status='success').inc()
        return result
    except:
        REQUEST_COUNT.labels(model='qwen2.5', status='error').inc()
        raise

---

9. 常见问题排查（5 个坑）

问题 1：Memory 泄漏

现象：Redis 占用持续增长

排查：

import redis
r = redis.Redis()
print(r.keys("chat_history:*"))
print(r.info()['used_memory_human'])

解决：设置 TTL，定期清理

---

问题 2：Token 超限

现象：BadRequestError: This model's maximum context length is 8192 tokens

解决：

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=64)
chunks = splitter.split_documents(documents)

---

问题 3：Tool 权限过大

现象：Agent 执行了危险命令

解决：命令白名单 + 沙箱隔离

---

问题 4：幻觉导致错误操作

现象：Agent 编造不存在的命令

解决：增加 RAG 上下文 + 人工确认关键操作

---

问题 5：长运行超时

现象：Agent 循环超过最大迭代次数

解决：设置 max_iterations=5，添加超时回调

---

10. 延伸阅读

• LangChain 官方文档
• LangGraph 状态机
• Production Tips

---

FAQ

LangChain 和 LangGraph 有什么区别？ LangChain 是线性 Chain 编排，LangGraph 支持循环和状态机。复杂多步骤流程用 LangGraph，简单 Pipeline 用 LangChain。

Memory 怎么管理会话历史？ 使用 ChatMessageHistory + ConversationBufferMemory。生产环境建议存 Redis，设置 TTL 避免泄漏。每个会话独立 namespace。

Token 超限怎么处理？ ① 截断旧消息 ② 压缩历史记录 ③ 用长上下文模型（128k+）④ 分块处理长文档。

Tools 权限如何限制？ 命令白名单 + 沙箱执行 + 结果审核。高危操作（删除/重启）必须人工确认。

---