LangGraph 极简入门：10分钟开发工作流

🌟 什么是 LangGraph？

根据 LangGraph 官方文档 的描述：

LangGraph 是一个低层级的编排框架和运行时，用于构建、管理和部署长时间运行、有状态的智能体（agents）。

听起来很技术？我们来“翻译”一下：

• “低层级”：意味着它不替你做决策（比如不用预设“必须用 ReAct 模式”），而是给你最大自由度去设计自己的流程。
• “有状态”：你的 AI 应用可以记住上下文，比如“上一步读了哪个文件”、“用户刚修改了什么”。
• “编排”：就像交响乐指挥，LangGraph 负责协调各个“乐手”（节点）何时演奏、如何衔接。

💡 简单说：LangGraph = 用“图”（Graph）的方式组织你的 AI 任务流程。

它被 Klarna、Elastic 等公司用于生产环境，足见其稳定性和强大能力。

🧱 LangGraph 的三大核心概念

要理解 LangGraph，只需掌握三个词：

1. State（状态）

整个工作流的“记忆本”。所有节点共享同一个状态对象，可以读取或更新它。

2. Node（节点）

工作流中的一个“步骤”，比如“读取文件”、“调用大模型”。每个节点是一个函数，接收状态，返回更新后的状态。

3. Edge（边）

定义节点之间的执行顺序。可以是固定顺序（A → B），也可以是条件跳转（如果成功则 A → B，否则 A → C）。

这三者组合起来，就形成了一张有向图——也就是你的 AI 工作流。

🛠️ 动手实践：用 LangGraph 实现“文件摘要”工作流

我们来实现一个具体需求：

用户输入一个文件名 → 程序读取该文件 → 调用大模型总结内容 → 输出摘要。

这个流程天然适合用 四个节点 表示：

[开始] → [读取文件] → [大模型总结] → [结束输出]

下面就是完整代码（可直接在 PyCharm 运行）：

import os
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# —————— 1. 准备工作 ——————
os.environ["OPENAI_API_KEY"] = "你的API密钥" # ← 替换这里！

# —————— 2. 定义状态 ——————
class WorkflowState(TypedDict):
filename: str # 输入的文件名
file_content: str # 读取到的内容
summary: str # 大模型生成的摘要
final_output: str # 最终输出

# —————— 3. 定义节点函数 ——————
def start_node(state: WorkflowState) –> dict:
print(f"[开始] 文件名: {state['filename']}")
return {}

def read_file_node(state: WorkflowState) –> dict:
try:
with open(state["filename"], 'r', encoding='utf-8') as f:
content = f.read()
print(f"[读取] 成功读取 {len(content)} 字符")
return {"file_content": content}
except Exception as e:
error = f"❌ 文件读取失败: {e}"
print(error)
return {"file_content": "", "summary": error}

def llm_summarize_node(state: WorkflowState) –> dict:
if not state["file_content"]:
return {"summary": "无内容可总结"}

print("[LLM] 正在生成摘要…")
llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke([
HumanMessage(content=f"请用中文简洁总结以下内容（3句话内）：\\n\\n{state['file_content']}")
])
return {"summary": response.content.strip()}

def end_node(state: WorkflowState) –> dict:
print("\\n✅ 最终摘要:")
print("=" * 40)
print(state["summary"])
print("=" * 40)
return {"final_output": state["summary"]}

# —————— 4. 构建工作流图 ——————
workflow = StateGraph(WorkflowState)

workflow.add_node("start", start_node)
workflow.add_node("read_file", read_file_node)
workflow.add_node("summarize", llm_summarize_node)
workflow.add_node("end", end_node)

workflow.set_entry_point("start")
workflow.add_edge("start", "read_file")
workflow.add_edge("read_file", "summarize")
workflow.add_edge("summarize", "end")
workflow.add_edge("end", END)

app = workflow.compile()

# —————— 5. 运行！ ——————
if __name__ == "__main__":
# 自动创建测试文件
test_file = "article.txt"
if not os.path.exists(test_file):
with open(test_file, "w") as f:
f.write("LangGraph 让复杂 AI 任务变得像搭积木一样简单。通过定义状态、节点和边，开发者可以清晰地表达业务逻辑，并轻松调试和扩展。")

# 启动工作流
result = app.invoke({"filename": test_file})

🔍 代码解析

运行后，你会看到清晰的步骤日志和最终摘要输出。

可复用的万能开发模板

LangGraph 开发四步法（新手必记）

无论你的需求是什么，都只需按以下四步操作：

✅ 第 1 步：定义状态（State）

问自己：整个流程中，哪些数据需要被记住？
把这些数据列出来，定义成一个 TypedDict。

class MyState(TypedDict):
input_data: str # 用户输入
file_content: str # 读取的文件
analysis_result: str # 分析结果
final_output: str # 最终输出

💡 小技巧：宁可多定义几个字段，也不要漏掉关键信息。

✅ 第 2 步：写节点函数（Node）

每个节点只做一件事！

• 节点 = 一个普通 Python 函数
• 输入：当前状态（state）
• 输出：要更新的状态字段（字典）

def read_file_node(state: MyState) –> dict:
with open(state["input_data"], "r") as f:
content = f.read()
return {"file_content": content} # 只返回要更新的部分

✅ 最佳实践：

• 函数名清晰（如 validate_input, call_llm, send_email）
• 做好异常处理（避免整个流程崩溃）
• 打印日志方便调试

✅ 第 3 步：连边（Edge）

决定执行顺序

• 固定顺序：add_edge(“A”, “B”)
• 条件跳转：add_conditional_edges(“A”, router_function)

# 简单线性流程
workflow.add_edge("start", "read_file")
workflow.add_edge("read_file", "analyze")
workflow.add_edge("analyze", "end")

🔜 下面总结中会教你怎么加“条件分支”。

✅ 第 4 步：运行 & 调试

app = workflow.compile()
result = app.invoke({"input_data": "my_file.txt"})
print(result["final_output"])

总结

核心概念

1. 状态管理（State Management）

LangGraph 使用一个状态对象（state object）来跟踪应用程序当前的状态。该状态可以包含你需要在各个步骤之间持久化的任意数据。

2. 节点（Nodes）

节点是图（graph）的基本构建单元。每个节点都是一个函数，它接收当前状态作为输入，并返回更新后的状态。

3. 边（Edges）

边定义了数据在节点之间的流动方式。边可以是条件性的，从而根据当前状态实现动态路由。

工作流的优势

• 将工作任务拆分成多个工作节点
• 能够将模型单次请求调用视作一个工作节点
• 能够灵活将其他代码逻辑也写入工作节点
• 能够对工作节点进行任务编排
• 能够在工作节点之间进行数据传递

加上条件边代码示例

把上面需求添加一个条件判断逻辑：

如果文件内容中包含 “LangGraph”，则交给大模型总结；否则直接跳过总结，把原文输出。

import os
from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# —————— 1. 准备工作 ——————
os.environ["OPENAI_API_KEY"] = "你的API密钥" # ← 替换这里！

# —————— 2. 定义状态 ——————
class WorkflowState(TypedDict):
filename: str
file_content: str
summary: str
final_output: str

# —————— 3. 定义节点函数 ——————
def start_node(state: WorkflowState) –> dict:
print(f"[开始] 文件名: {state['filename']}")
return {}

def read_file_node(state: WorkflowState) –> dict:
try:
with open(state["filename"], 'r', encoding='utf-8') as f:
content = f.read()
print(f"[读取] 成功读取 {len(content)} 字符")
return {"file_content": content}
except Exception as e:
error = f"❌ 文件读取失败: {e}"
print(error)
return {"file_content": "", "summary": error}

def llm_summarize_node(state: WorkflowState) –> dict:
if not state["file_content"]:
return {"summary": "无内容可总结"}

print("[LLM] 正在生成摘要…")
llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke([
HumanMessage(content=f"请用中文简洁总结以下内容（3句话内）：\\n\\n{state['file_content']}")
])
return {"summary": response.content.strip()}

def end_node(state: WorkflowState) –> dict:
output = state.get("summary") or state.get("file_content", "无内容")
print("\\n✅ 最终输出:")
print("=" * 40)
print(output)
print("=" * 40)
return {"final_output": output}

# —————— 4. 新增：条件路由函数 ——————
def route_after_read(state: WorkflowState) –> Literal["summarize", "end"]:
"""
判断是否包含 'LangGraph'：
– 包含 → 返回 "summarize"
– 不包含 → 返回 "end"
"""
if "LangGraph" in state["file_content"]:
print("[路由] 检测到 'LangGraph'，将调用大模型总结")
return "summarize"
else:
print("[路由] 未检测到 'LangGraph'，直接输出原文")
return "end"

# —————— 5. 构建工作流图 ——————
workflow = StateGraph(WorkflowState)

workflow.add_node("start", start_node)
workflow.add_node("read_file", read_file_node)
workflow.add_node("summarize", llm_summarize_node)
workflow.add_node("end", end_node)

workflow.set_entry_point("start")
workflow.add_edge("start", "read_file")

# 🔥 关键修改：用条件边替代固定边
workflow.add_conditional_edges(
"read_file", # 从哪个节点出发做判断
route_after_read, # 路由函数
{
"summarize": "summarize",
"end": "end"
} # 映射返回值到目标节点名
)

# 注意：summarize 执行完仍要走向 end
workflow.add_edge("summarize", "end")
workflow.add_edge("end", END)

app = workflow.compile()

# —————— 6. 运行测试 ——————
if __name__ == "__main__":
# 测试 1：包含 LangGraph → 应该走 summarize
test_file1 = "with_langgraph.txt"
with open(test_file1, "w", encoding="utf-8") as f:
f.write("LangGraph 是一个强大的 AI 工作流框架，用于构建复杂智能体。")

print(">>> 测试 1：包含 'LangGraph'")
result1 = app.invoke({"filename": test_file1})

print("\\n" + "="*60 + "\\n")

# 测试 2：不包含 LangGraph → 应该直接输出原文
test_file2 = "without_langgraph.txt"
with open(test_file2, "w", encoding="utf-8") as f:
f.write("今天天气很好，适合出去散步。")

print(">>> 测试 2：不包含 'LangGraph'")
result2 = app.invoke({"filename": test_file2})

🔍 关键机制解析

workflow.add_conditional_edges(
"read_file", # 源节点
route_after_read, # 路由函数
{
"summarize": "summarize", # 如果路由返回 "summarize"，就去 summarize 节点
"end": "end" # 如果返回 "end"，就去 end 节点
}
)

💡 注意：字典的 key 是路由函数的返回值，value 是目标节点名。

output = state.get("summary") or state.get("file_content", "无内容")

• 如果走了 summarize 路径 → 有 summary
• 如果直接跳 end → 没有 summary，就用 file_content