一、为什么需要 LangChain 和 LangGraph

直接调用大模型 API 很容易，但当应用变得复杂时，开发者很快会遇到以下问题：

• Prompt 散落在各个业务函数中；
• 多轮对话上下文难以管理；
• 工具调用逻辑与模型调用逻辑混杂；
• RAG 检索流程难以复用；
• 任务失败后无法恢复；
• 多个 Agent 之间缺乏清晰的协作关系；
• 复杂流程只能通过大量 if else 维护。

LangChain 主要解决的是“大模型应用组件化”问题。它提供了模型封装、Prompt 模板、输出解析器、工具调用、检索器和 Agent 等组件。

LangGraph 则进一步解决“复杂流程编排”问题。它把大模型应用表示为一个有状态的图：

输入
  |
  v
意图识别
  |
  +--> 知识库检索
  |
  +--> 工具调用
  |
  v
答案生成
  |
  v
输出

LangChain 更像是组件库，LangGraph 更像是工作流引擎。两者可以结合使用。

---

二、安装依赖

本文使用 Ollama 作为本地模型服务，也可以替换为 OpenAI、通义千问或其他兼容模型。

安装依赖：

pip install -U langchain langchain-ollama langgraph pydantic

如果需要向量检索，可以额外安装：

pip install chromadb langchain-chroma

下载 Ollama 模型：

ollama pull qwen2.5:7b

---

三、LangChain 调用本地模型

最简单的模型调用代码如下：

from langchain_ollama import ChatOllama

llm = ChatOllama(
    model="qwen2.5:7b",
    temperature=0.2
)

response = llm.invoke("请解释什么是LangChain")
print(response.content)

ChatOllama 返回的是一个消息对象，而不是普通字符串。生成的文本通常位于：

response.content

3.1 使用消息角色

from langchain_core.messages import SystemMessage, HumanMessage

messages = [
    SystemMessage(content="你是一名资深软件架构师。"),
    HumanMessage(content="如何设计一个高可用的消息队列系统？")
]

response = llm.invoke(messages)
print(response.content)

消息角色包括：

• SystemMessage：定义模型角色和行为；
• HumanMessage：用户输入；
• AIMessage：模型输出；
• ToolMessage：工具执行结果。

通过消息类型组织上下文，比手动拼接字符串更加清晰。

---

四、使用 Prompt 模板

将 Prompt 直接写在业务代码里，会导致后续维护困难。LangChain 提供了模板机制。

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    (
        "system",
        "你是一名专业技术作者，请使用清晰、准确的语言回答问题。"
    ),
    (
        "human",
        "请围绕以下主题写一段技术说明：{topic}"
    )
])

chain = prompt | llm

result = chain.invoke({
    "topic": "向量数据库的索引原理"
})

print(result.content)

这里的 | 表示 Runnable 组合。执行过程如下：

输入变量
   |
Prompt模板
   |
ChatOllama
   |
AIMessage

这比手动执行多个函数更加符合声明式编程思想。

---

五、结构化输出

大模型输出自然语言，程序很难直接解析。实际业务中通常需要模型返回固定结构。

定义 Pydantic 模型：

from pydantic import BaseModel, Field
from typing import Literal

class IntentResult(BaseModel):
    intent: Literal["knowledge", "weather", "general"]
    confidence: float = Field(
        ge=0,
        le=1,
        description="意图识别置信度"
    )
    reason: str

让模型输出结构化结果：

structured_llm = llm.with_structured_output(IntentResult)

result = structured_llm.invoke(
    "用户问：今天北京天气怎么样？"
)

print(result)
print(result.intent)
print(result.confidence)

结构化输出的意义在于把模型输出转换为程序可使用的数据对象，后续可以直接根据 intent 选择不同处理流程。

需要注意，结构化输出并不意味着模型绝对可靠。生产环境仍然需要：

• 对字段进行校验；
• 处理解析失败；
• 设置默认值；
• 限制枚举范围；
• 记录异常响应。

---

六、实现一个简单的 RAG Chain

6.1 创建向量库

from langchain_ollama import OllamaEmbeddings
from langchain_chroma import Chroma

embeddings = OllamaEmbeddings(
    model="nomic-embed-text"
)

documents = [
    "系统发布需要经过开发、测试、预发布和生产审批。",
    "员工报销需要在费用发生后30天内提交。",
    "数据库变更必须先执行备份，再执行变更脚本。"
]

vectorstore = Chroma.from_texts(
    texts=documents,
    embedding=embeddings,
    collection_name="company_knowledge",
    persist_directory="./chroma_data"
)

retriever = vectorstore.as_retriever(
    search_kwargs={"k": 3}
)

6.2 构建检索问答链

from langchain_core.prompts import ChatPromptTemplate

rag_prompt = ChatPromptTemplate.from_template("""
你是企业知识库助手。
请根据上下文回答问题。
如果上下文中没有答案，请明确说明无法从知识库中找到依据。
不要根据常识补充未提供的信息。

上下文：
{context}

问题：
{question}
""")

def format_documents(docs):
    return "\n\n".join(doc.page_content for doc in docs)

def rag_answer(question: str):
    docs = retriever.invoke(question)
    context = format_documents(docs)

    chain = rag_prompt | llm
    response = chain.invoke({
        "context": context,
        "question": question
    })

    return {
        "answer": response.content,
        "sources": [doc.page_content for doc in docs]
    }

result = rag_answer("数据库变更前需要做什么？")
print(result["answer"])

一个可靠的 RAG 系统不应该只返回答案，还应返回来源信息。这样用户可以判断答案是否有依据。

---

七、为什么复杂任务需要 LangGraph

假设我们要构建一个智能助手，它需要完成以下流程：

1. 判断用户意图；
2. 如果是知识库问题，执行知识检索；
3. 如果是天气问题，调用天气工具；
4. 如果是普通问题，直接让模型回答；
5. 检查答案是否存在依据；
6. 必要时重新检索；
7. 最后输出结果。

如果使用普通函数，代码可能变成大量嵌套判断：

if intent == "knowledge":
    docs = retrieve()
    if not docs:
        ...
elif intent == "weather":
    ...
else:
    ...

LangGraph 将这些步骤拆分成节点，并通过边描述节点之间的跳转关系。这样流程更加清晰，也方便增加重试、人工审批和持久化。

---

八、定义 LangGraph 状态

状态是图中所有节点共享的数据结构。

from typing import TypedDict, List
from langchain_core.messages import BaseMessage

class AgentState(TypedDict, total=False):
    question: str
    intent: str
    confidence: float
    documents: List[str]
    answer: str
    grounded: bool
    messages: List[BaseMessage]

total=False 表示字段可以不是一开始就全部存在。不同节点可以逐步补充状态。

---

九、实现意图识别节点

from pydantic import BaseModel
from typing import Literal

class Intent(BaseModel):
    name: Literal["knowledge", "weather", "general"]
    confidence: float

intent_llm = llm.with_structured_output(Intent)

def classify_intent(state: AgentState):
    result = intent_llm.invoke(
        f"请判断下面问题属于哪一类：知识库、天气或普通问题。\n"
        f"问题：{state['question']}"
    )

    return {
        "intent": result.name,
        "confidence": result.confidence
    }

节点函数接收状态，返回需要更新的字段。它不需要手动传递所有变量。

---

十、实现知识检索节点

def retrieve_documents(state: AgentState):
    docs = retriever.invoke(state["question"])

    return {
        "documents": [doc.page_content for doc in docs]
    }

为了避免将过多信息传给模型，可以限制内容大小：

def retrieve_documents(state: AgentState):
    docs = retriever.invoke(state["question"])

    contents = []
    total_length = 0

    for doc in docs:
        text = doc.page_content
        if total_length + len(text) > 5000:
            break

        contents.append(text)
        total_length += len(text)

    return {"documents": contents}

这是一个很重要的工程细节。检索结果越多不一定越好。过长的上下文可能导致：

• 推理速度变慢；
• 关键内容被淹没；
• Token 成本增加；
• 模型回答出现偏离。

---

十一、实现答案生成节点

answer_prompt = ChatPromptTemplate.from_template("""
你是一个严谨的企业智能助手。
请根据提供的上下文回答问题。

上下文：
{context}

用户问题：
{question}

要求：
1. 不要编造上下文之外的事实；
2. 如果资料不足，明确说明；
3. 使用简洁、专业的中文回答。
""")

def generate_answer(state: AgentState):
    context = "\n\n".join(state.get("documents", []))

    response = (answer_prompt | llm).invoke({
        "context": context,
        "question": state["question"]
    })

    return {
        "answer": response.content
    }

---

十二、实现答案校验节点

大模型生成答案后，可以让另一个模型判断答案是否有上下文支持。

class GroundingResult(BaseModel):
    grounded: bool
    reason: str

grounding_llm = llm.with_structured_output(GroundingResult)

def check_grounding(state: AgentState):
    context = "\n\n".join(state.get("documents", []))

    prompt = f"""
请判断下面的答案是否能够从上下文中得到支持。

上下文：
{context}

问题：
{state["question"]}

答案：
{state["answer"]}

请判断答案是否有充分依据。
"""

    result = grounding_llm.invoke(prompt)

    return {
        "grounded": result.grounded
    }

这种“生成 + 校验”的方式可以降低明显幻觉，但它并不能从根本上保证事实正确。校验模型本身也可能判断错误，因此关键业务仍然需要人工审核或规则系统兜底。

---

十三、组装 LangGraph

from langgraph.graph import StateGraph, START, END

builder = StateGraph(AgentState)

builder.add_node("classify", classify_intent)
builder.add_node("retrieve", retrieve_documents)
builder.add_node("generate", generate_answer)
builder.add_node("check", check_grounding)

定义意图路由：

def route_intent(state: AgentState):
    intent = state.get("intent")

    if intent == "knowledge":
        return "retrieve"

    if intent == "weather":
        return "weather"

    return "generate_general"

为了简化示例，天气和普通问题节点如下：

def weather_node(state: AgentState):
    return {
        "answer": "天气工具尚未配置，请先接入实际天气API。"
    }

def general_node(state: AgentState):
    response = llm.invoke(state["question"])
    return {
        "answer": response.content
    }

builder.add_node("weather", weather_node)
builder.add_node("generate_general", general_node)

添加边：

builder.add_edge(START, "classify")

builder.add_conditional_edges(
    "classify",
    route_intent,
    {
        "retrieve": "retrieve",
        "weather": "weather",
        "generate_general": "generate_general"
    }
)

builder.add_edge("retrieve", "generate")
builder.add_edge("generate", "check")
builder.add_edge("weather", END)
builder.add_edge("generate_general", END)

定义校验后的路由：

def route_grounding(state: AgentState):
    if state.get("grounded"):
        return "finish"

    return "retry"

增加重试节点：

def retry_retrieve(state: AgentState):
    docs = retriever.invoke(
        state["question"] + " 请检索更具体、更相关的资料"
    )

    return {
        "documents": [doc.page_content for doc in docs]
    }

builder.add_node("retry", retry_retrieve)

builder.add_conditional_edges(
    "check",
    route_grounding,
    {
        "finish": END,
        "retry": "retry"
    }
)

builder.add_edge("retry", "generate")

编译图：

graph = builder.compile()

执行：

result = graph.invoke({
    "question": "数据库变更前需要执行什么操作？"
})

print(result["answer"])
print("是否有依据：", result.get("grounded"))

完整流程如下：

START
  |
classify
  |
  +--> weather ------> END
  |
  +--> generate_general -> END
  |
retrieve
  |
generate
  |
check
  |
  +--> grounded ------> END
  |
retry
  |
generate

---

十四、使用流式输出观察执行过程

LangGraph 支持流式获取节点执行结果：

for event in graph.stream(
    {"question": "系统发布需要经过哪些流程？"},
    stream_mode="updates"
):
    print(event)

输出可能类似：

{'classify': {'intent': 'knowledge', 'confidence': 0.96}}
{'retrieve': {'documents': ['系统发布需要经过开发、测试、预发布和生产审批。']}}
{'generate': {'answer': '系统发布需要经过开发、测试、预发布和生产审批。'}}
{'check': {'grounded': True}}

流式事件对于调试非常重要。传统的黑盒调用只能看到最终答案，而图流式输出可以观察每个节点的状态变化。

---

十五、接入工具调用

智能 Agent 通常需要调用外部工具。下面定义一个简单的天气工具：

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。"""
    weather_data = {
        "北京": "晴，25摄氏度",
        "上海": "多云，28摄氏度",
        "广州": "小雨，30摄氏度"
    }

    return weather_data.get(city, "暂时没有该城市的天气数据")

模型绑定工具：

tool_llm = llm.bind_tools([get_weather])

response = tool_llm.invoke(
    "请查询北京的天气"
)

print(response.tool_calls)

需要注意，模型返回工具调用请求并不代表工具已经执行。应用程序需要解析 tool_calls，执行对应函数，再把结果返回给模型。

一个简单的执行逻辑：

def execute_tools(message):
    outputs = []

    for call in message.tool_calls:
        if call["name"] == "get_weather":
            result = get_weather.invoke(call["args"])
            outputs.append(result)

    return outputs

在实际项目中，应该建立工具注册表，而不是使用大量 if else：

TOOLS = {
    "get_weather": get_weather
}

def execute_tools(message):
    results = []

    for call in message.tool_calls:
        tool = TOOLS.get(call["name"])

        if tool is None:
            raise ValueError(f"未知工具：{call['name']}")

        result = tool.invoke(call["args"])
        results.append(result)

    return results

---

十六、LangChain 与 LangGraph 的职责划分

可以使用下面的方式理解二者关系：

组件	主要职责
Chat Model	调用大语言模型
Prompt Template	管理输入模板
Output Parser	解析模型输出
Retriever	检索外部知识
Tool	封装外部能力
Chain	串联固定处理步骤
LangGraph	编排条件分支和循环流程
Checkpointer	保存流程状态
Human-in-the-loop	引入人工审批

如果流程是固定的，例如：

输入 -> Prompt -> 模型 -> 解析

使用 LangChain Chain 就足够。

如果流程包含条件、循环、重试、人工审批和多个 Agent，则应当使用 LangGraph。

---

十七、生产环境设计建议

17.1 节点必须尽量保持单一职责

一个节点最好只完成一件事。例如：

• 意图识别节点只负责分类；
• 检索节点只负责检索；
• 生成节点只负责生成；
• 校验节点只负责判断。

这样更容易测试，也容易替换。

17.2 所有外部调用都要设置超时

工具调用、向量数据库调用和模型调用都可能失败。不能让一个节点无限等待。

17.3 对状态进行裁剪

状态会随着流程不断增长。如果把完整对话、所有检索文档和所有工具结果都放进去，长流程可能导致上下文过大。

17.4 使用持久化保存状态

复杂任务可能运行很长时间，或者需要人工确认。此时应该使用 Checkpointer 保存状态，以便中断后继续。

17.5 不要让 Agent 拥有无限权限

工具应该遵循最小权限原则。例如：

• 只读数据库工具与写入工具分开；
• 高风险操作必须人工审批；
• 限制工具参数范围；
• 记录每次工具调用；
• 对敏感数据进行脱敏。

---

十八、测试一个 LangGraph 应用

不要只测试最终答案，还应该测试每个节点。

def test_intent():
    result = classify_intent({
        "question": "数据库变更前需要做什么？"
    })

    assert result["intent"] in {
        "knowledge",
        "weather",
        "general"
    }

测试检索节点：

def test_retrieval():
    result = retrieve_documents({
        "question": "系统发布流程是什么？"
    })

    assert "documents" in result
    assert len(result["documents"]) > 0

测试图的完整执行：

def test_graph():
    result = graph.invoke({
        "question": "数据库变更前需要执行什么操作？"
    })

    assert result.get("answer")

对于大模型应用，测试重点不只是字符串是否完全相等，还包括：

• 是否返回结构化数据；
• 是否调用正确工具；
• 是否包含来源；
• 是否在资料不足时拒答；
• 是否能够从失败节点恢复；
• 是否不会泄露敏感信息。

---

十九、总结

LangChain 解决的是大模型应用开发中的组件复用问题，LangGraph 解决的是复杂任务中的流程编排问题。

一个比较完整的企业级智能应用通常可以拆分为：

用户输入
   |
意图识别
   |
流程路由
   |
检索 / 工具 / 普通问答
   |
结果生成
   |
事实校验
   |
人工审批或最终输出

LangChain 适合封装模型、Prompt、工具和检索器；LangGraph 适合表达条件分支、循环、重试和状态管理。

在实践中，不建议一开始就把所有逻辑都设计成复杂 Agent。更合理的路径是：

1. 先使用 LangChain 实现最小可用链路；
2. 再把固定流程拆成独立节点；
3. 当出现条件分支和循环时引入 LangGraph；
4. 最后补充状态持久化、监控、人工审批和安全控制。

真正可靠的大模型应用，不是简单地把模型接入系统，而是围绕模型建立一套可观测、可恢复、可验证的工程流程。