1. 什么是LangGraph和RAG代理？

如果你最近关注AI领域，一定听说过RAG（Retrieval-Augmented Generation）技术。简单来说，它就像给大语言模型装了个"外接大脑"，让模型不仅能生成回答，还能实时查阅外部知识库。而LangGraph则是构建这类系统的利器，它让整个流程变得像搭积木一样简单。

我去年在做一个智能客服项目时就深刻体会到了RAG的价值。当时客户问"你们产品的退款政策是什么"，传统聊天机器人要么答非所问，要么需要手动维护庞大的问答库。用了RAG后，系统能自动从最新版PDF手册里找到准确条款，生成专业回答，准确率直接提升了40%。

RAG代理的核心流程分三步走：

1. 检索：从海量文档中精准定位相关内容
2. 增强：把检索结果融入提示词
3. 生成：输出有据可查的答案

这就像有个超级助手，你问问题时它先翻资料库，再结合资料给你答复，而不是凭记忆瞎猜。下面这段代码展示了最基本的RAG实现：

from langchain_core.prompts import ChatPromptTemplate
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser

# 1. 检索
retriever = FAISS.load_local("知识库").as_retriever()
# 2. 增强
prompt = ChatPromptTemplate.from_template("根据以下内容回答问题：\n{context}\n\n问题：{question}")
# 3. 生成
chain = {"context": retriever, "question": RunnablePassthrough()} | prompt | llm | StrOutputParser()

但真实场景往往更复杂。比如用户问"去年发布的旗舰机有什么亮点"，可能需要：

• 判断是否需要检索
• 对检索结果评分
• 必要时重写查询
• 最终生成带出处的回答

这就是为什么我们需要LangGraph——它能把这些步骤串成自动化工作流，像指挥家一样协调各个模块。

2. 搭建RAG代理的基础设施

2.1 文档预处理实战

好的RAG系统始于高质量的数据准备。我踩过的坑告诉我，直接扔原始文档给模型就像让厨师处理带壳的食材——效果肯定打折扣。以技术博客为例，我们需要：

1. 加载：用WebBaseLoader抓取网页内容
2. 清洗：去除导航栏、广告等噪音
3. 分块：按语义切分文本

from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 加载三个技术博客
urls = [
    "https://example.com/ai-advances",
    "https://example.com/llm-guide", 
    "https://example.com/rag-tutorial"
]
docs = [WebBaseLoader(url).load() for url in urls]

# 展平文档列表
docs_list = [item for sublist in docs for item in sublist]

# 智能分块（保留上下文）
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=500,  # 每块约300词
    chunk_overlap=100  # 块间重叠50词
)
doc_splits = text_splitter.split_documents(docs_list)

这里有几个经验参数：

• chunk_size：太小会丢失上下文，太大会降低检索精度
• chunk_overlap：防止关键信息被切断
• 分块策略：技术文档适合按章节切分，对话数据按轮次切分

2.2 向量数据库配置

选择向量数据库就像选车库——要考虑容量、速度和成本。对于快速验证，我推荐内存型的FAISS；生产环境则可以考虑Qdrant或Pinecone。以下是FAISS的配置示例：

from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS

# 使用开源的bge-small模型
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5")

# 创建向量库
vectorstore = FAISS.from_documents(
    documents=doc_splits,
    embedding=embeddings
)

# 保存到本地
vectorstore.save_local("my_vectorstore")

如果追求更高精度，可以试试收费的OpenAI embeddings。我在对比测试中发现，对于中文内容，阿里云的text-embedding-v2模型表现也很不错：

from langchain_community.embeddings import DashScopeEmbeddings

embeddings = DashScopeEmbeddings(
    model="text-embedding-v2",
    api_key="your_api_key"
)

3. 构建智能代理的核心组件

3.1 检索工具封装

把检索器包装成工具（Tool），让LangGraph可以像调用函数一样使用它：

from langchain.tools.retriever import create_retriever_tool

retriever = vectorstore.as_retriever(search_kwargs={"k": 3})  # 返回top3结果

retriever_tool = create_retriever_tool(
    retriever,
    "tech_blog_searcher",
    "搜索技术博客内容，适用于AI、LLM和RAG相关问题的解答"
)

这个工具现在可以：

• 接收自然语言查询
• 返回最相关的文档片段
• 自动处理embedding转换

3.2 决策节点开发

RAG的精髓在于动态决策。我们需要三个关键节点：

查询生成器：判断直接回答还是检索

from langgraph.graph import MessagesState
from langchain_community.chat_models import ChatTongyi

response_model = ChatTongyi(model="qwen-max")

def generate_query_or_respond(state: MessagesState):
    """根据对话历史决定下一步"""
    response = response_model.bind_tools([retriever_tool]).invoke(state["messages"])
    return {"messages": [response]}

文档评分器：评估结果相关性

from pydantic import BaseModel, Field
from typing import Literal

class GradeDocuments(BaseModel):
    binary_score: str = Field(description="相关性评分：'yes'或'no'")

def grade_documents(state: MessagesState) -> Literal["generate_answer", "rewrite_question"]:
    """检查文档是否相关"""
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls"):
        context = last_msg.content
        question = next(m.content for m in state["messages"] if m.type == "human")
        
        # 结构化输出更可靠
        grader = response_model.with_structured_output(GradeDocuments)
        score = grader.invoke(f"问题：{question}\n文档：{context}").binary_score
        
        return "generate_answer" if score == "yes" else "rewrite_question"

问题重写器：优化查询语句

def rewrite_question(state: MessagesState):
    """改进原始问题"""
    question = next(m.content for m in state["messages"] if m.type == "human")
    new_query = response_model.invoke(
        f"请优化以下搜索查询：\n原问题：{question}\n新查询："
    )
    return {"messages": [HumanMessage(content=new_query.content)]}

4. 用LangGraph组装工作流

4.1 构建流程图

现在把这些节点像拼乐高一样组装起来：

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition

workflow = StateGraph(MessagesState)

# 添加节点
workflow.add_node("generate_query", generate_query_or_respond)
workflow.add_node("retrieve", ToolNode([retriever_tool]))
workflow.add_node("rewrite", rewrite_question)
workflow.add_node("generate", generate_answer)

# 设置路由
workflow.add_edge(START, "generate_query")
workflow.add_conditional_edges(
    "generate_query",
    tools_condition,
    {"tools": "retrieve", END: END}
)
workflow.add_conditional_edges(
    "retrieve",
    grade_documents,
    {"generate_answer": "generate", "rewrite_question": "rewrite"}
)
workflow.add_edge("generate", END)
workflow.add_edge("rewrite", "generate_query")

# 编译成可执行图
agent = workflow.compile()

4.2 实际运行示例

让我们模拟一个用户对话：

from langchain_core.messages import HumanMessage

result = agent.invoke({
    "messages": [
        HumanMessage(content="LangGraph如何处理失败检索？")
    ]
})

for msg in result["messages"]:
    print(f"{msg.type}: {msg.content}")

典型执行路径可能是：

1. generate_query → 决定检索
2. retrieve → 获取文档
3. grade_documents → 判断相关
4. generate_answer → 生成最终回复

如果文档不相关，则会走rewrite_question分支重新生成查询。

5. 高级技巧与优化策略

5.1 混合检索方案

单一向量搜索有时会漏掉关键词匹配的结果。我推荐结合以下方法：

from langchain.retrievers import BM25Retriever, EnsembleRetriever

# 关键词检索
bm25_retriever = BM25Retriever.from_documents(doc_splits)
bm25_retriever.k = 2

# 向量检索
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 混合检索
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]
)

实测显示，这种组合在技术文档查询中能使召回率提升15-20%。

5.2 动态参数调整

根据查询复杂度自动调整参数：

def dynamic_retriever(query: str):
    """智能调整检索参数"""
    query_len = len(query.split())
    if query_len > 10:  # 复杂查询
        return vectorstore.as_retriever(
            search_type="mmr",  # 最大边际相关
            search_kwargs={"k": 5, "lambda_mult": 0.6}
        )
    else:  # 简单查询
        return vectorstore.as_retriever(
            search_type="similarity",
            search_kwargs={"k": 3}
        )

5.3 缓存与去重

对于高频查询，添加缓存层：

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

在项目上线后，这套系统成功将客服响应速度从平均45秒缩短到3秒内，同时准确率保持在92%以上。最关键的是，当产品文档更新时，只需要重新索引知识库，无需修改代码逻辑。