Day26:LangGraph核心筑基！State+Node + 路由 + 循环，打造可控式 AI 工作流

🔥 引言

还在被 LangChain 原生 Agent 的不可控、乱跳转、无记忆折磨？LangGraph 就是 AI Agent 的流程总控中心，把 AI 的思考、执行变成可视化流水线，想让 AI 怎么走就怎么走！

咱们一站式吃透 LangGraph核心底层逻辑，从基础、进阶、高级三阶段，极简工作流到迭代式思考 AI，全程代码可跑、案例易懂，通义千问 + 本地 Qwen 双版本全覆盖，零基础也能直接上手！

🎯 第一阶段：基础篇｜State+Node+Edge，搭建 LangGraph 极简工作流

核心概念白话拆解

• State（状态）：AI 的共享记事本，所有节点都能读写，存问题、答案、执行记录，全局通用不丢失

• Node（节点）：AI 的专属打工人，每个节点只做一件事（调模型、处理数据），单一职责好维护

• Edge（边）：AI 的执行路线图，规定节点执行顺序，从开始到结束一步不乱

环境一键安装

pip install langgraph langchain langchain-community dashscope ollama -i https://pypi.tuna.tsinghua.edu.cn/simple

实战案例（2 个，通义千问 + 本地 Qwen 双版本）

案例 1：单节点极简 AI 问答工作流

# 导入核心依赖
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
import os

# ===================== 模型二选一 =====================
# 1. 通义千问在线版（新手首选）
os.environ["DASHSCOPE_API_KEY"] = "你的阿里云API Key"
from langchain_community.llms import Tongyi
llm = Tongyi(model="qwen-plus", temperature=0.1)

# 2. 本地Qwen-7B离线版（无API Key用这个）
# from langchain_community.llms import Ollama
# llm = Ollama(model="qwen:7b")

# 1. 定义State：AI的共享记事本
class AgentState(TypedDict):
    user_question: str
    ai_answer: str

# 2. 定义节点：大模型回答
def llm_node(state: AgentState):
    question = state["user_question"]
    answer = llm.invoke(f"简洁回答：{question}")
    return {"ai_answer": answer}

# 3. 构建工作流
workflow = StateGraph(AgentState)
workflow.add_node("ai_answer", llm_node)
workflow.add_edge(START, "ai_answer")
workflow.add_edge("ai_answer", END)
# 编译生效
app = workflow.compile()

# 运行测试
if __name__ == "__main__":
    res = app.invoke({"user_question": "LangGraph是做什么的？"})
    print(f"用户问题：{res['user_question']}")
    print(f"AI回答：{res['ai_answer']}")

案例 2：多节点顺序执行工作流（预处理 + 问答）

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
import os

# 模型配置同案例1
os.environ["DASHSCOPE_API_KEY"] = "你的API Key"
from langchain_community.llms import Tongyi
llm = Tongyi(model="qwen-plus", temperature=0.1)

# 扩展State：新增预处理后问题
class AgentState(TypedDict):
    user_question: str
    clean_question: str
    ai_answer: str

# 节点1：问题预处理（去空格、标点）
def preprocess_node(state: AgentState):
    clean_q = state["user_question"].strip().replace("？", "")
    return {"clean_question": clean_q}

# 节点2：大模型回答
def llm_node(state: AgentState):
    answer = llm.invoke(f"回答：{state['clean_question']}")
    return {"ai_answer": answer}

# 构建顺序工作流
workflow = StateGraph(AgentState)
workflow.add_node("preprocess", preprocess_node)
workflow.add_node("ai_answer", llm_node)
workflow.add_edge(START, "preprocess")
workflow.add_edge("preprocess", "ai_answer")
workflow.add_edge("ai_answer", END)
app = workflow.compile()

# 测试
res = app.invoke({"user_question": "   什么是AI Agent？？？   "})
print(f"预处理后：{res['clean_question']}")
print(f"AI回答：{res['ai_answer']}")

基础阶段避坑指南

1. State 必须用TypedDict定义，是 LangGraph 强制规范

2. 节点函数必须返回字典，键名和 State 完全一致

3. 必须执行compile()编译，工作流才会生效

---

🎯 第二阶段：进阶篇｜条件路由，让 AI 学会自主选择执行路径

核心概念白话拆解

条件路由 = AI 的路口红绿灯，根据 State 里的内容判断下一步走哪个节点，实现if-else 可视化，让 AI 不再一根筋，能根据问题类型自主决策。

实战案例（2 个）

案例 1：二分支路由（计算问题 / 常识问题）

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain_core.tools import tool
import os

# 模型配置
os.environ["DASHSCOPE_API_KEY"] = "你的API Key"
from langchain_community.llms import Tongyi
llm = Tongyi(model="qwen-plus", temperature=0.1)

# 定义计算器工具
@tool
def add(a: int, b: int) -> int:
    """两数相加计算"""
    return a + b

# 定义State
class AgentState(TypedDict):
    user_question: str
    need_calc: bool
    result: str

# 节点1：判断是否需要计算
def judge_node(state: AgentState):
    q = state["user_question"]
    need_calc = any(i in q for i in ["加", "+","减","-"])
    return {"need_calc": need_calc}

# 节点2：计算节点
def calc_node(state: AgentState):
    return {"result": f"计算结果：{add.invoke({'a':10, 'b':20})}"}

# 节点3：大模型节点
def llm_node(state: AgentState):
    return {"result": llm.invoke(state["user_question"])}

# 构建条件路由
workflow = StateGraph(AgentState)
workflow.add_node("judge", judge_node)
workflow.add_node("calc", calc_node)
workflow.add_node("llm", llm_node)

workflow.add_edge(START, "judge")
# 路由函数
def router(state: AgentState):
    return "calc" if state["need_calc"] else "llm"

workflow.add_conditional_edges("judge", router, ["calc", "llm"])
workflow.add_edge("calc", END)
workflow.add_edge("llm", END)
app = workflow.compile()

# 测试计算问题
print(app.invoke({"user_question": "10+20等于多少？"}))
# 测试常识问题
print(app.invoke({"user_question": "什么是大模型？"}))

案例 2：三分支路由（计算 / 检索 / 常识）

# 扩展分支，新增RAG检索节点
class AgentState(TypedDict):
    user_question: str
    q_type: str  # calc/rag/llm
    result: str

# 检索节点（模拟）
def rag_node(state: AgentState):
    return {"result": "RAG检索：2026年LPR为3.45%"}

# 三分支路由函数
def router(state: AgentState):
    q = state["user_question"]
    if any(i in q for i in ["加", "+"]):
        return "calc"
    elif "LPR" in q:
        return "rag"
    else:
        return "llm"

# 构建三分支工作流（其余代码同案例1，直接运行）
workflow = StateGraph(AgentState)
workflow.add_node("judge", judge_node)
workflow.add_node("calc", calc_node)
workflow.add_node("rag", rag_node)
workflow.add_node("llm", llm_node)
workflow.add_edge(START, "judge")
workflow.add_conditional_edges("judge", router, ["calc", "rag", "llm"])
app = workflow.compile()

# 测试检索问题
print(app.invoke({"user_question": "2026年LPR是多少？"}))

进阶阶段避坑指南

1. 路由函数返回值必须是已添加的节点名

2. 所有分支节点最终都要绑定 END，避免流程卡死

3. 分支判断逻辑要清晰，减少误判

---

🎯 第三阶段：高级篇｜循环 + 中断，打造迭代式思考 AI Agent

核心概念白话拆解

• 循环：让 AI 在节点间反复执行，直到满足结束条件，实现反复优化回答

• 中断：暂停工作流，等待用户输入，实现人机交互闭环

实战案例（2 个）

案例 1：迭代重试循环（AI 优化回答 3 次）

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
import os

# 模型配置
os.environ["DASHSCOPE_API_KEY"] = "你的API Key"
from langchain_community.llms import Tongyi
llm = Tongyi(model="qwen-plus", temperature=0.1)

# 定义State：新增重试次数
class AgentState(TypedDict):
    user_question: str
    ai_answer: str
    retry_count: int

# 思考优化节点
def think_node(state: AgentState):
    retry = state["retry_count"] + 1
    answer = llm.invoke(f"第{retry}次优化：{state['user_question']}，更简洁")
    return {"ai_answer": answer, "retry_count": retry}

# 循环路由：重试≤2次继续优化
def loop_router(state: AgentState):
    return "think" if state["retry_count"] < 2 else END

# 构建循环工作流
workflow = StateGraph(AgentState)
workflow.add_node("think", think_node)
workflow.add_edge(START, "think")
workflow.add_conditional_edges("think", loop_router)
app = workflow.compile()

# 测试
res = app.invoke({"user_question": "介绍LangGraph", "retry_count": 0})
print(f"最终优化回答：{res['ai_answer']}")
print(f"重试次数：{res['retry_count']}")

案例 2：用户中断循环（人机交互确认）

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
import os

# 模型配置同案例1
os.environ["DASHSCOPE_API_KEY"] = "你的API Key"
from langchain_community.llms import Tongyi
llm = Tongyi(model="qwen-plus", temperature=0.1)

class AgentState(TypedDict):
    user_input: str
    is_satisfied: bool

# AI回答节点
def ai_node(state: AgentState):
    ans = llm.invoke(state["user_input"])
    print(f"AI回答：{ans}")
    return {}

# 用户确认中断节点
def user_node(state: AgentState):
    s = input("满意吗？(y/n)：")
    return {"is_satisfied": s == "y"}

# 循环路由
def router(state: AgentState):
    return "ai_node" if not state["is_satisfied"] else END

# 构建中断循环工作流
workflow = StateGraph(AgentState)
workflow.add_node("ai_node", ai_node)
workflow.add_node("user_node", user_node)
workflow.add_edge(START, "ai_node")
workflow.add_edge("ai_node", "user_node")
workflow.add_conditional_edges("user_node", router)
app = workflow.compile()

# 测试
app.invoke({"user_input": "什么是LangGraph循环", "is_satisfied": False})

高级阶段避坑指南

1. 循环必须设置结束条件，防止死循环

2. 中断节点做好异常处理，避免输入崩溃

3. 重试次数限制，减少 token 消耗

总结

一篇吃透 LangGraph基础组件 + 条件路由 + 循环中断，搞定 AI 工作流的核心逻辑，从极简流程到迭代式 AI，完全掌握 LangGraph 的底层灵魂！