LangChain实战：构建可观测的RAG应用-AI智能范式网

LangChain实战：构建可观测的RAG应用

鄂奎阿·2026-08-14·阅读 23 GEO检测（免费）

LangChainRAG检索增强生成

|难度：普通|阅读约 14 分钟

1. LangChain智能体开发实战：构建可观测的RAG应用

在当今AI应用开发领域，检索增强生成(RAG)已成为连接大语言模型与私有知识库的黄金标准。但很多开发者在原型阶段往往只关注功能实现，忽略了至关重要的可观测性设计。今天我将分享一个完整的开发案例，展示如何从零构建具备完善监控能力的RAG系统。

这个项目源于我最近为某金融客户开发的内部知识问答系统。与简单演示不同，生产级RAG应用需要全程跟踪：从检索质量、生成效果到系统性能。下面我就从环境搭建开始，逐步拆解每个关键环节的实现与监控方案。

需要模型API调用？免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。

2. 环境准备与基础架构

2.1 工具链选型

python复制# 核心依赖
pip install langchain==0.1.0 openai==1.12.0 
pip install langsmith==0.1.0  # 可观测性平台

选择LangChain作为框架核心，主要考虑其三点优势：

对RAG流程的标准化封装
丰富的集成生态（超过100种数据源支持）
原生支持LangSmith监控平台

注意：OpenAI客户端请使用最新v1.x版本，与旧版API不兼容。建议通过环境变量管理API密钥：

bash复制export OPENAI_API_KEY="sk-xxx"

2.2 项目结构设计

code复制/project
├── /data          # 知识库文档
├── /monitoring    # 监控配置
├── app.py         # 主应用
├── retriever.py   # 检索模块
└── eval.py        # 评估脚本

这种模块化设计便于后期扩展。我曾在一个医疗项目中因早期没做分离，导致检索逻辑和业务代码严重耦合，后期维护成本极高。

3. 检索系统实现

3.1 知识库处理

python复制from langchain.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

def prepare_knowledge_base():
    loader = DirectoryLoader('./data', glob="**/*.pdf")
    docs = loader.load()
    
    # 专业文档建议使用小分块
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,
        chunk_overlap=50
    )
    return splitter.split_documents(docs)

分块大小直接影响检索精度：

法律/金融文档：300-500字符
技术文档：500-800字符
通用内容：800-1200字符

3.2 检索器增强

python复制from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS

def create_retriever():
    docs = prepare_knowledge_base()
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    
    # 加入元数据过滤
    vectorstore = FAISS.from_documents(docs, embeddings)
    retriever = vectorstore.as_retriever(
        search_kwargs={"k": 3, "filter": {"department": "finance"}}
    )
    return retriever

这里有几个关键参数优化点：

k=3：平衡召回率与噪声
使用text-embedding-3-small：性价比最高的嵌入模型
元数据过滤：确保检索范围正确

4. RAG链构建与监控

4.1 基础链实现

python复制from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

template = """Answer based on context:
{context}

Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)

def build_rag_chain():
    return (
        {"context": retriever, "question": RunnablePassthrough()}
        | prompt
        | ChatOpenAI(model="gpt-3.5-turbo-0125")
    )

4.2 监控埋点

python复制from langsmith import Client

client = Client()

def log_run(inputs, outputs):
    client.create_run(
        name="rag_chain",
        inputs=inputs,
        outputs=outputs,
        run_type="chain"
    )

# 装饰器方式监控
@traceable(run_type="chain", name="rag_production")
def rag_chain(question):
    chain = build_rag_chain()
    result = chain.invoke(question)
    log_run({"question": question}, {"answer": result})
    return result

监控指标建议：

检索耗时百分位（P99/P95）
生成token数分布
用户反馈评分（通过埋点收集）

5. 生产级优化策略

5.1 检索优化

python复制# 混合检索策略
from langchain.retrievers import BM25Retriever, EnsembleRetriever

bm25_retriever = BM25Retriever.from_documents(docs)
ensemble = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.7, 0.3]
)

在电商客服场景测试中，混合检索使准确率提升了18%。

5.2 生成控制

python复制prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个严谨的金融顾问，回答必须：\n"
              "- 引用具体条款\n"
              "- 标明信息来源\n"
              "- 不超过100字"),
    ("human", "{question}")
])

通过提示词工程可以：

降低幻觉率
控制输出格式
限制生成长度

6. 常见问题排查

6.1 检索失效场景

症状：返回无关内容
排查步骤：

检查嵌入模型是否匹配（text-embedding-3-small ≠ text-embedding-ada-002）
验证分块策略是否合适
测试原始向量相似度得分

python复制# 诊断脚本
query = "如何申报增值税？"
docs = retriever.get_relevant_documents(query)
for doc in docs:
    print(doc.metadata["source"], "| Score:", doc.score)

6.2 生成质量问题

症状：答案不准确
解决方案：

增加上下文窗口

python复制ChatOpenAI(model="gpt-4-1106-preview", max_tokens=4000)

添加验证层

python复制validator = OpenAIModerationChain()
validated_result = validator.run(result)

7. 性能优化实战

7.1 缓存策略

python复制from langchain.cache import SQLiteCache
import langchain
langchain.llm_cache = SQLiteCache(database_path=".langchain.db")

缓存效果实测：

重复查询响应时间从1200ms → 80ms
API成本降低约40%

7.2 异步处理

python复制async def async_rag(questions):
    chain = build_rag_chain()
    return await chain.abatch(questions)

批量处理时，异步版本吞吐量可提升5-8倍。

8. 评估与迭代

8.1 自动化测试

python复制test_cases = [
    {
        "question": "年度报告提交截止日？",
        "expected": "根据2023年规定，应在3月31日前提交",
        "max_score": 5
    }
]

def run_eval():
    grader = load_evaluator("labeled_score_string")
    for case in test_cases:
        result = rag_chain(case["question"])
        print(grader.evaluate(result, case))

8.2 监控看板

LangSmith提供的关键指标：

链式调用拓扑图
各环节耗时热力图
异常查询预警

在最近一次升级中，通过监控发现gpt-4的响应延迟P99从2.1s升至3.4s，及时回滚避免了线上事故。

这个项目让我深刻体会到：可观测性不是后期添加的功能，而是应该从第一天就构建的基础能力。现在我们的RAG系统每天处理超过2万次查询，平均响应时间稳定在800ms以内，准确率保持在92%以上。