好的，我们接着来深入探讨LangGraph。它是一个比LangChain更进阶的框架，专为处理复杂的、有状态的AI工作流而设计。
为了让您快速建立起直观印象，下面的表格清晰地对比了LangGraph和LangChain的核心差异。
下面，我们来详细解答您的五个问题。
一、LangGraph是什么？
LangGraph是LangChain团队开发的一个开源框架，用于构建有状态、多步骤的智能体（Agent）和工作流应用程序。它的核心思想是将应用逻辑建模为一个有向图。
您可以把它理解为一个强大的“流程引擎”：
图结构：工作流由节点和边构成。节点代表一个具体的操作（如调用LLM、使用工具），边定义了操作之间的流转路径。
有状态性：图中的一个共享的状态对象贯穿整个工作流，随着每个节点的执行而不断更新，从而保留了执行的上下文和信息。这是LangGraph实现复杂多步推理的关键。
它弥补了LangChain在处理非线性、动态流程控制方面的不足，特别适合构建需要循环、回溯、条件判断的智能系统。
二、为什么要使用LangGraph？
当您的项目需求超越了简单的“提问-回答”模式，变得复杂时，就是考虑LangGraph的时候：
需要循环和分支逻辑：当您的任务需要根据上一步的结果决定下一步做什么，或者需要反复尝试、迭代优化直到满足条件时（例如，生成一篇文章直到质量合格），LangGraph的条件边和循环能力是原生支持。
构建复杂的多智能体系统：如果您需要多个AI智能体（或工具）协同工作，例如一个负责分析、一个负责检索、一个负责撰写，LangGraph可以通过图结构优雅地编排它们之间的协作与状态传递。
实现真正的有状态应用：对于需要长时间运行、并保持上下文记忆的对话系统或任务，LangGraph内置的状态持久化机制可以轻松实现中断恢复和记忆功能。
引入“人在回路”：在关键决策点（如内容发布、金融交易）插入人工审核环节，LangGraph可以很方便地暂停工作流，等待人工输入后再继续。
三、核心概念和关键知识点
理解LangGraph，关键是掌握以下三个核心构件：
状态：一个类似字典的共享数据结构，在工作流开始时创建，并在整个执行过程中传递和更新。它记录了所有需要共享的信息，如用户输入、中间结果、对话历史等。
节点：工作流中的基本计算单元。它是一个函数，接收当前状态作为输入，执行特定任务（如调用API、处理数据），并返回要更新到状态中的部分内容。
边：连接节点的路径，决定了工作流的走向。分为两种：
普通边：无条件地从一个节点指向另一个节点。
条件边：根据当前状态的值，动态决定下一个要执行的节点，从而实现
if-else逻辑。
高级知识点：
持久化检查点：LangGraph可以将工作流的当前状态（包括执行到哪个节点、已有的输出等）保存到数据库或文件中。这使得长时间运行的任务可以暂停和恢复，也实现了对话记忆的功能。
多智能体协作：通过将不同的智能体定义为图中的不同节点，并让它们读写共享状态，可以轻松实现多智能体协同解决问题。
四、可以用它来实现什么样的需求或功能？
LangGraph非常适用于以下场景：
高级对话系统与聊天机器人：实现能记住复杂上下文、主动追问、动态调整策略的智能对话系统。
迭代式内容生成与优化：如自动写作助手，完成“生成草稿 -> 自我审查/人工审核 -> 基于反馈修改”的循环。
复杂问题求解代理：构建能自主规划、使用工具（如搜索引擎、计算器）、并检查结果可靠性的AI代理。
多专家协同系统：让多个 specialized 的AI模型协作，例如一个负责代码生成，一个负责代码审查和安全检查。
企业级业务流程自动化：处理需要多步骤判断、人工审批环节的复杂工作流。
五、如何使用LangGraph & 完整最佳实践
使用LangGraph通常遵循以下步骤：
定义状态结构：明确工作流中需要跟踪哪些信息。
创建节点函数：实现每个步骤的具体逻辑。
构建图：创建
StateGraph，添加节点，并用边连接它们。编译和执行：将图编译成可执行对象，然后传入初始状态运行。
完整最佳实践：构建一个带自我检查的写作助手
下面我们实现一个相对复杂的写作助手，它能生成文案，并自动进行事实性检查。如果检查不通过，它会尝试重新生成或引用更可靠的来源。
import os
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver # 用于持久化状态（记忆）
from langchain_community.tools import DuckDuckGoSearchRun # 搜索工具
from langchain_community.chat_models import ChatOllama # 假设使用本地Ollama模型，您可替换为OpenAI等
# 设置LLM和工具
llm = ChatOllama(model="qwen2.5:1.5b")
search = DuckDuckGoSearchRun()
# 1. 定义状态结构（共享的数据蓝图）
class WriterState(TypedDict):
user_query: str # 用户原始请求
draft: str # 生成的文案草稿
search_results: list # 搜索到的结果
fact_check_feedback: str # 事实检查的反馈
revision_count: int # 修订次数，用于限制循环
final_answer: str # 最终答案
# 2. 定义各个节点（功能步骤）
def retrieve_information(state: WriterState):
"""节点1：信息检索。根据用户查询搜索相关信息。"""
print("【检索节点】正在搜索相关信息...")
results = search.run(state["user_query"])
# 简单模拟，实际可做更复杂的处理
top_results = [results[:500]] # 截取部分内容
return {"search_results": top_results}
def generate_draft(state: WriterState):
"""节点2：生成初稿。基于查询和检索结果生成文案。"""
print("【生成节点】正在生成文案草稿...")
context = "\n".join(state["search_results"]) if state["search_results"] else "无额外背景信息。"
prompt = f"""
请根据以下用户请求和背景信息，生成一段准确、清晰的文案。
用户请求：{state['user_query']}
背景信息：{context}
文案：
"""
response = llm.invoke(prompt)
draft_text = response.content
return {"draft": draft_text, "revision_count": state.get("revision_count", 0) + 1}
def fact_check(state: WriterState):
"""节点3：事实检查。让LLM检查草稿中的事实准确性。"""
print("【检查节点】正在进行事实检查...")
prompt = f"""
请检查以下文案的事实准确性。文案是基于用户查询“{state['user_query']}”生成的。
如果文案中的主要事实与公认知识或提供的背景信息没有明显冲突，请回复“PASS”。
如果发现有可能不准确或缺乏依据的陈述，请回复“FAIL”并简要说明原因。
文案：{state['draft']}
检查结果（直接回复PASS或FAIL）：
"""
response = llm.invoke(prompt)
feedback = response.content.strip()
return {"fact_check_feedback": feedback}
def human_fallback(state: WriterState):
"""节点4（备选）：人工介入。当自动检查多次不通过时，提示人工处理。"""
print("【人工节点】需要人工介入审核。")
# 在实际应用中，这里可以发送邮件、通知系统等
# 此处模拟人工审核通过
return {"final_answer": f"[需人工审核] {state['draft']}"}
def finalize(state: WriterState):
"""节点5：终结点。准备最终答案。"""
print("【终结点】准备最终答案。")
return {"final_answer": state["draft"]}
# 3. 构建图工作流
builder = StateGraph(WriterState)
# 添加所有定义好的节点
builder.add_node("retrieve", retrieve_information)
builder.add_node("generate", generate_draft)
builder.add_node("check", fact_check)
builder.add_node("human_review", human_fallback)
builder.add_node("finalize", finalize)
# 设置工作流入口：从检索信息开始
builder.set_entry_point("retrieve")
# 添加边：检索 -> 生成
builder.add_edge("retrieve", "generate")
# 添加边：生成 -> 检查
builder.add_edge("generate", "check")
# 4. 添加最关键的条件边：根据检查结果决定下一步
def route_after_check(state: WriterState) -> Literal["finalize", "generate", "human_review"]:
"""路由函数：决定检查后的路径"""
feedback = state["fact_check_feedback"].upper()
revision_count = state.get("revision_count", 0)
if "PASS" in feedback:
return "finalize" # 检查通过，走向最终化
elif revision_count >= 3: # 如果已经重试了3次，转人工
return "human_review"
else: # 检查不通过且未超次数，重新生成
return "generate"
builder.add_conditional_edges(
"check", # 从这个节点出发
route_after_check, # 调用路由函数
{
"finalize": "finalize",
"generate": "generate",
"human_review": "human_review"
}
)
# 添加边：人工审核 -> 最终化；最终化 -> 结束
builder.add_edge("human_review", "finalize")
builder.add_edge("finalize", END)
# 编译图，并启用内存检查点以实现状态持久化/记忆
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)
# 5. 执行工作流
if __name__ == "__main__":
print("=== 写作助手开始工作 ===")
# 配置一个会话线程ID，用于保持对话记忆
config = {"configurable": {"thread_id": "writer_agent_001"}}
# 初始查询
initial_query = "解释一下量子计算的基本原理及其潜在应用。"
print(f"用户查询: {initial_query}")
# 调用图执行工作流
final_state = graph.invoke(
{"user_query": initial_query},
config=config
)
print("\n=== 工作流执行完毕 ===")
print(f"最终结果: {final_state['final_answer']}")
print(f"总共修订次数: {final_state['revision_count']}")这个最佳实践示例的核心价值在于：
状态驱动：整个工作流由
WriterState驱动，每个节点都能读取和更新共享状态。循环与条件判断：通过
route_after_check函数和条件边，实现了“生成-检查-再生成”的自动化循环，直到满足条件（PASS或超次）。容错与人性化：设置了重试次数上限，并在自动流程失败时转向人工介入，保证了系统的健壮性。
可持久化：通过
MemorySaver，该助手可以记住每次交互的上下文。如果这是一个多轮对话，下一次用户说“把刚才的解释再简化一下”，助手可以基于之前的状态继续工作。
希望这个详细的解释和示例能帮助你全面理解LangGraph，并为你构建复杂的AI应用提供坚实的基础！
评论区