大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。



 
  
    
     
文章是一份全面的LangGraph框架教程，从基础概念到2025-2026年新特性。详细介绍了StateGraph、State、Nodes和Edges等核心组件，提供了对话机器人、工具调用Agent等实战示例。重点讲解了Command API、interrupt()函数等最新功能，以及状态设计、节点设计等**实践。适合开发者学习构建有状态、多步骤的LLM应用和智能Agent系统。
 

---

本教程基于 LangGraph 1.0.8（2026年2月最新版），涵盖所有核心概念和2025-2026年发布的新特性。

1. 基础概念
2. 环境安装
3. 核心概念详解
4. 实战示例
5. 高级特性
6. 2025-2026新特性
7. **实践
8. 常见模式
9. 练习题

1.1 什么是 LangGraph？

LangGraph 是一个用于构建有状态、多步骤 LLM 应用的框架，特别适合构建：

• 代理（Agent）工作流
• 多代理系统
• 复杂的对话系统
• 需要人工审核的工作流

核心理念: 将应用建模为有向图，其中节点执行计算，边定义流程。

1.2 核心组件

GPT plus 代充 只需 145StateGraph (状态图) ├── State (状态) ├── Nodes (节点) ├── Edges (边) │ ├── Normal Edges (普通边) │ └── Conditional Edges (条件边) └── Checkpointer (检查点) 

# 安装基础包 pip install langgraph langchain langchain-openai # 如果使用 Anthropic pip install langchain-anthropic # 新增：Supervisor 库（可选） pip install langgraph-supervisor # 其他可选依赖 pip install langchain-community 

版本检查：

GPT plus 代充 只需 145python -c "import langgraph; print(langgraph.__version__)" # 应该显示 1.0.8 或更高 

3.1 State (状态)

状态是在节点之间传递的数据结构。

from typing import TypedDict, Annotated from langgraph.graph.message import add_messages

方式1: 使用 TypedDict

classAgentState(TypedDict):

GPT plus 代充 只需 145messages: Annotated[list, add_messages] # 自动合并消息 user_info: str current_step: int 

方式2: 使用简单字典

state = { “messages”: [], “counter”: 0 }

重要: Annotated[list, add_messages] 会自动合并新旧消息，而不是覆盖。

3.2 Nodes (节点)

节点是执行实际工作的函数。

defmy_node(state: AgentState) -> dict: “”“

GPT plus 代充 只需 145节点函数必须: 1. 接收 state 作为参数 2. 返回字典，用于更新 state """ messages = state["messages"] 

执行某些操作

result = do_something(messages) 

返回状态更新

return { ”messages“: [result], ”current_step“: state[”current_step“] + 1

GPT plus 代充 只需 145} 

3.3 Edges (边)

边定义节点之间的连接关系。

from langgraph.graph import StateGraph, END

创建图

workflow = StateGraph(AgentState)

添加节点

workflow.add_node(”node1“, node1_function) workflow.add_node(”node2“, node2_function)

普通边: 总是从 node1 到 node2

workflow.add_edge(”node1“, ”node2“)

条件边: 根据条件决定下一步

defrouter(state: AgentState) -> str: if state[”current_step“] > 5: return”end“ else: return”continue“

workflow.add_conditional_edges( ”node2“,

GPT plus 代充 只需 145router, # 路由函数 { 

”continue“: ”node1“, ”end“: END

} 

)

设置入口点

workflow.set_entry_point(”node1“)

示例1: 简单的对话机器人

GPT plus 代充 只需 145from langchain_anthropic import ChatAnthropic from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from typing import Annotated, TypedDict

1. 定义状态

classChatState(TypedDict):

messages: Annotated[list, add_messages] 

2. 创建 LLM

llm = ChatAnthropic(model=”claude-3-5-sonnet-“)

3. 定义节点

defchatbot(state: ChatState): return {”messages“: [llm.invoke(state[”messages“])]}

4. 构建图

graph = StateGraph(ChatState) graph.add_node(”chatbot“, chatbot) graph.set_entry_point(”chatbot“) graph.add_edge(”chatbot“, END)

5. 编译

app = graph.compile()

6. 使用

from langchain_core.messages import HumanMessage

response = app.invoke({ ”messages“: [HumanMessage(content=”你好！“)] })

print(response[”messages“][-1].content)

示例2: 带工具调用的 Agent

GPT plus 代充 只需 145from langchain_core.tools import tool from langchain_core.messages import ToolMessage import json

1. 定义工具

@tool defget_weather(location: str) -> str: ”“”获取指定地点的天气“”“

模拟天气数据

returnf”{location}的天气是晴天，温度25度“

@tool defsearch_web(query: str) -> str: ”“”搜索网络“”“ returnf”搜索结果: {query}“

tools = [get_weather, search_web]

2. 绑定工具到 LLM

llm_with_tools = llm.bind_tools(tools)

3. 定义状态

classAgentState(TypedDict):

messages: Annotated[list, add_messages] 

4. 定义节点

defagent(state: AgentState): ”“”调用 LLM“”“ return {”messages“: [llm_with_tools.invoke(state[”messages“])]}

deftool_node(state: AgentState): ”“”执行工具“”“

GPT plus 代充 只需 145messages = state["messages"] last_message = messages[-1] tool_calls = last_message.tool_calls results = [] 

for tool_call in tool_calls:

找到对应的工具

 tool_name = tool_call["name"] tool_to_call = {t.name: t for t in tools}[tool_name] 

调用工具

GPT plus 代充 只需 145 result = tool_to_call.invoke(tool_call["args"]) 

创建工具消息

 results.append( ToolMessage( content=str(result), tool_call_id=tool_call["id"] ) ) 

return {”messages“: results}

5. 路由函数

defshould_continue(state: AgentState) -> str: ”“”判断是否需要继续调用工具“”“

GPT plus 代充 只需 145last_message = state["messages"][-1] 

if hasattr(last_message, ‘tool_calls’) and last_message.tool_calls: return”tools“ return”end“

6. 构建图

workflow = StateGraph(AgentState)

workflow.add_node(”agent“, agent) workflow.add_node(”tools“, tool_node)

workflow.set_entry_point(”agent“)

workflow.add_conditional_edges( ”agent“,

should_continue, { 

”tools“: ”tools“, ”end“: END

GPT plus 代充 只需 145} 

)

workflow.add_edge(”tools“, ”agent“)

app = workflow.compile()

7. 测试

from langchain_core.messages import HumanMessage

result = app.invoke({ ”messages“: [HumanMessage(content=”北京的天气怎么样？“)] })

for msg in result[”messages“]:

print(f"{msg.__class__.__name__}: {msg.content}") 

示例3: 多代理协作

GPT plus 代充 只需 145classMultiAgentState(TypedDict):

messages: Annotated[list, add_messages] current_agent: str task_result: dict 

研究员代理

defresearcher(state: MultiAgentState):

GPT plus 代充 只需 145llm = ChatAnthropic(model="claude-3-5-sonnet-") prompt = f"""你是一个研究员。请针对以下问题进行研究: 

{state[‘messages’][-1].content}

提供详细的研究结果。""" response = llm.invoke([HumanMessage(content=prompt)]) 

return { ”messages“: [response], ”current_agent“: ”writer“, ”task_result“: {”research“: response.content}

GPT plus 代充 只需 145} 

写作代理

defwriter(state: MultiAgentState):

llm = ChatAnthropic(model="claude-3-5-sonnet-") research_data = state["task_result"].get("research", "") prompt = f"""你是一个专业写作者。基于以下研究结果，写一篇文章: 研究结果: 

{research_data}

GPT plus 代充 只需 145请写一篇简洁专业的文章。""" response = llm.invoke([HumanMessage(content=prompt)]) 

return { ”messages“: [response], ”current_agent“: ”reviewer“

} 

审核代理

defreviewer(state: MultiAgentState):

GPT plus 代充 只需 145llm = ChatAnthropic(model="claude-3-5-sonnet-") article = state["messages"][-1].content prompt = f"""你是一个审核者。请审核以下文章，提出改进建议: 

{article}

如果文章质量很好，回复"APPROVED"。否则提出具体改进建议。""" response = llm.invoke([HumanMessage(content=prompt)]) 

return {”messages“: [response]}

路由逻辑

defroute_agent(state: MultiAgentState) -> str:

GPT plus 代充 只需 145current = state.get("current_agent", "researcher") 

if current == ”researcher“: return”writer“ elif current == ”writer“: return”reviewer“ else:

检查是否通过审核

 last_msg = state["messages"][-1].content 

if”APPROVED“in last_msg: return END else: return”writer“# 返回写作环节修改

构建图

workflow = StateGraph(MultiAgentState)

workflow.add_node(”researcher“, researcher) workflow.add_node(”writer“, writer) workflow.add_node(”reviewer“, reviewer)

workflow.set_entry_point(”researcher“)

workflow.add_edge(”researcher“, ”writer“) workflow.add_edge(”writer“, ”reviewer“)

workflow.add_conditional_edges( ”reviewer“, lambda state: ”end“if”APPROVED“in state[”messages“][-1].content else”writer“,

GPT plus 代充 只需 145{ 

”writer“: ”writer“, ”end“: END

} 

)

app = workflow.compile()

5.1 持久化 (Checkpointing)

持久化允许你保存和恢复图的执行状态。

GPT plus 代充 只需 145from langgraph.checkpoint.memory import MemorySaver

创建检查点

checkpointer = MemorySaver()

编译时添加检查点

app = graph.compile(checkpointer=checkpointer)

使用线程ID来跟踪会话

config = {”configurable“: {”thread_id“: ”user_123“}}

第一轮对话

response1 = app.invoke(

{"messages": [HumanMessage(content="我叫小明")]}, config=config 

)

第二轮对话 - 会记住上下文

response2 = app.invoke(

GPT plus 代充 只需 145{"messages": [HumanMessage(content="我叫什么名字？")]}, config=config 

)

生产环境持久化：

from langgraph.checkpoint.postgres import PostgresSaver

使用 PostgreSQL 持久化

checkpointer = PostgresSaver.from_conn_string( ”postgresql://user:pass@localhost/dbname“ )

app = graph.compile(checkpointer=checkpointer)

5.2 流式输出

GPT plus 代充 只需 145# 流式获取每个节点的输出 for chunk in app.stream({”messages“: [HumanMessage(content=”你好“)]}):

print(chunk) print("---") 

流式获取tokens

asyncfor event in app.astream_events(

GPT plus 代充 只需 145{"messages": [HumanMessage(content="你好")]}, version="v1" 

): if event[”event“] == ”on_chat_model_stream“:

 print(event["data"]["chunk"].content, end="") 

6.1 Command API ⭐ (最重要的新特性)

什么是 Command？

Command 允许节点同时完成状态更新和路由决策，大幅简化代码。

对比：旧方法 vs 新方法

GPT plus 代充 只需 145# ❌ 旧方法：需要分开处理 defmy_node(state): return {”counter“: state[”counter“] + 1}

defrouter(state): if state[”counter“] > 5: return”end“ return”continue“

workflow.add_conditional_edges(”my_node“, router, {…})

✅ 新方法：一步完成

from langgraph.types import Command

defmy_node(state): if state[”counter“] > 5: return Command(

 update={"counter": state["counter"] + 1}, goto=END ) 

else: return Command(

GPT plus 代充 只需 145 update={"counter": state["counter"] + 1}, goto="my_node" ) 

实战示例：智能审批流程

from langgraph.graph import StateGraph, END, START from langgraph.types import Command from typing import TypedDict

classApprovalState(TypedDict):

GPT plus 代充 只需 145request: str approved: bool retry_count: int reason: str 

defanalyze_request(state: ApprovalState): ”“”分析请求并决定是否需要审批“”“

risk_level = assess_risk(state["request"]) 

if risk_level == ”low“:

低风险：自动批准

return Command(

GPT plus 代充 只需 145 update={ 

”approved“: True, ”reason“: ”自动批准 - 低风险操作“

 }, goto="execute" ) 

elif risk_level == ”high“:

高风险：需要人工审批

return Command(

GPT plus 代充 只需 145 update={"approved": False}, goto="human_approval" ) 

else:

中等风险：AI 辅助决策

return Command(goto=”ai_review“)

defhuman_approval(state: ApprovalState): ”“”人工审批“”“

… 等待人工输入

if state[”approved“]: return Command(goto=”execute“) else: if state[”retry_count“] < 3: return Command(

 update={"retry_count": state["retry_count"] + 1}, goto="analyze_request"# 重新分析 ) 

else: return Command(goto=END)

defexecute(state: ApprovalState): ”“”执行操作“”“ if state[”approved“]:

执行逻辑

GPT plus 代充 只需 145 print(f"执行: {state['request']}") 

return {}

构建图 - 不需要 conditional_edges！

workflow = StateGraph(ApprovalState) workflow.add_node(”analyze_request“, analyze_request) workflow.add_node(”human_approval“, human_approval) workflow.add_node(”ai_review“, ai_review_node) workflow.add_node(”execute“, execute)

workflow.set_entry_point(”analyze_request“)

节点自己通过 Command 决定路由！

app = workflow.compile()

Command 的优势：

• ✅ 代码更简洁：状态和路由逻辑在一起
• ✅ 更灵活：每个节点可以动态决定下一步
• ✅ 减少错误：不需要同步多个地方的逻辑
• ✅ 适合复杂流程：轻松实现复杂的决策树

6.2 interrupt() 函数 ⭐ (超级实用)

什么是 interrupt()？

一个革命性的函数，让人工介入（Human-in-the-Loop）变得极其简单。

对比：旧方法 vs 新方法

# ❌ 旧方法：需要在编译时指定 app = workflow.compile(

GPT plus 代充 只需 145checkpointer=checkpointer, interrupt_before=["approval_node"] # 硬编码，不够灵活 

)

✅ 新方法：在节点内部动态中断

from langgraph.types import interrupt, Command

defapproval_node(state):

暂停并请求审批

user_decision = interrupt() 

根据用户决策继续

if user_decision == ”approve“: return Command(goto=”execute“) elif user_decision == ”reject“: return Command(goto=END) else:

GPT plus 代充 只需 145 modifications = interrupt("请输入修改内容：") 

return Command(

 update=, goto="reprocess" ) 

完整示例：邮件审核系统

GPT plus 代充 只需 145from langgraph.types import interrupt, Command from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from typing import TypedDict

classEmailState(TypedDict):

to: str subject: str body: str approved: bool 

defdraft_email(state: EmailState): ”“”AI 起草邮件“”“

GPT plus 代充 只需 145draft = generate_email_with_ai(state["to"]) 

return { ”subject“: draft[”subject“], ”body“: draft[”body“]

} 

defreview_email(state: EmailState): ”“”人工审核邮件“”“

第一次中断：展示邮件给用户

GPT plus 代充 只需 145approval = interrupt({ 

”message“: ”请审核以下邮件“, ”to“: state[”to“], ”subject“: state[”subject“], ”body“: state[”body“], ”actions“: [”send“, ”edit“, ”cancel“]

}) 

if approval[”action“] == ”send“: return Command(

GPT plus 代充 只需 145 update={"approved": True}, goto="send_email" ) 

elif approval[”action“] == ”edit“:

第二次中断：获取编辑内容

 edits = interrupt({ 

”message“: ”请修改邮件内容“, ”current_subject“: state[”subject“], ”current_body“: state[”body“]

GPT plus 代充 只需 145 }) 

return Command(

 update=, goto="review_email"# 重新审核 ) 

else: # cancel return Command(goto=END)

defsend_email(state: EmailState): ”“”发送邮件“”“ if state[”approved“]:

GPT plus 代充 只需 145 send_email_api(state["to"], state["subject"], state["body"]) print(f"✓ 邮件已发送到 {state['to']}") 

return {}

构建图

workflow = StateGraph(EmailState) workflow.add_node(”draft_email“, draft_email) workflow.add_node(”review_email“, review_email) workflow.add_node(”send_email“, send_email)

workflow.set_entry_point(”draft_email“) workflow.add_edge(”draft_email“, ”review_email“)

必须使用 checkpointer

app = workflow.compile(checkpointer=MemorySaver())

使用

config = {”configurable“: {”thread_id“: ”email-123“}}

第一次运行 - 会在审核处暂停

result = app.invoke({ ”to“: ”“, ”subject“: ”“, ”body“: ”“, ”approved“: False }, config)

查看中断信息

print(result.get(”interrupt“))

用户做出决策后继续

result = app.invoke(

Command(resume={"action": "send"}), config 

)

高级用法：工具调用审批

GPT plus 代充 只需 145defreview_tool_calls(state): ”“”逐个审核 AI 要调用的工具“”“

tool_calls = state["messages"][-1].tool_calls approved_calls = [] 

for tool_call in tool_calls:

展示工具调用详情

GPT plus 代充 只需 145 approval = interrupt({ 

”tool“: tool_call[”name“], ”args“: tool_call[”args“], ”question“: f”允许调用 {tool_call[‘name’]} 工�吗？“, ”risk_level“: assess_tool_risk(tool_call)

 }) 

if approval[”decision“] == ”approve“:

用户可能修改了参数

if”modified_args“in approval:

GPT plus 代充 只需 145 tool_call["args"] = approval["modified_args"] approved_calls.append(tool_call) 

elif approval[”decision“] == ”skip“: continue# 跳过这个工具

else: # reject return Command(goto=END)

return Command(

 update={"approved_tool_calls": approved_calls}, goto="execute_tools" ) 

6.3 工具直接更新状态

新特性：工具现在可以返回 Command，直接更新图的状态和控制流程。

GPT plus 代充 只需 145from langchain_core.tools import tool from langgraph.types import Command

@tool defsmart_search(query: str, state) -> Command: ”“”智能搜索 - 可以更新状态和决定路由“”“

results = search_api(query) 

根据结果质量决定下一步

if len(results) == 0: return Command(

GPT plus 代充 只需 145 update={ 

”search_results“: [], ”needs_refinement“: True

 }, goto="refine_query"# 需要优化查询 ) 

elif len(results) > 100: return Command(

GPT plus 代充 只需 145 update={ 

”search_results“: results[:10], ”too_broad“: True

 }, goto="narrow_search" ) 

else: return Command(

GPT plus 代充 只需 145 update=, goto="process_results" ) 

@tool defvalidate_and_route(data: str) -> Command: ”“”验证数据并决定路由“”“

validation_result = validate(data) 

if validation_result[”valid“]: return Command(

GPT plus 代充 只需 145 update={"validated_data": data}, goto="process" ) 

elif validation_result[”can_fix“]: return Command(

 update={"errors": validation_result["errors"]}, goto="auto_fix" ) 

else: return Command(

GPT plus 代充 只需 145 update={"errors": validation_result["errors"]}, goto="human_review" ) 

6.4 Node Caching (性能优化)

缓存节点结果，避免重复计算

from langgraph.graph import StateGraph

defexpensive_analysis(state): ”“”耗时的分析任务“”“

假设这需要30秒

GPT plus 代充 只需 145result = run_heavy_computation(state["data"]) 

return {”analysis“: result}

workflow = StateGraph(State)

启用缓存

workflow.add_node( ”analyze“,

expensive_analysis, cache=True# 缓存结果 

)

编译时配置

app = workflow.compile(

GPT plus 代充 只需 145cache_config={ 

”ttl“: 3600, # 缓存1小时 ”max_size“: 100# 最多100个结果

} 

)

第一次运行：30秒

result1 = app.invoke({”data“: ”same_data“})

第二次运行相同数据：<1秒（从缓存读取）

result2 = app.invoke({”data“: ”same_data“})

高级缓存策略：

GPT plus 代充 只需 145# 自定义缓存键 workflow.add_node( ”process“,

process_function, cache=True, cache_key=lambda state: f"{state['user_id']}:{state['query']}" 

)

条件缓存

workflow.add_node( ”compute“,

GPT plus 代充 只需 145expensive_computation, cache=lambda state: state.get("mode") == "production" 

)

6.5 Deferred Nodes (并行任务同步)

等待所有并行分支完成后再执行

from langgraph.graph import StateGraph, START, END

deffetch_user_data(state): ”“”并行任务1：获取用户数据“”“ return {”user_data“: fetch_from_db(state[”user_id“])}

deffetch_product_data(state): ”“”并行任务2：获取产品数据“”“ return {”product_data“: fetch_from_api(state[”product_id“])}

deffetch_analytics(state): ”“”并行任务3：获取分析数据“”“ return {”analytics“: run_analytics(state[”query“])}

defcombine_all_data(state): ”“”汇总所有数据 - 需要等待所有并行任务“”“

GPT plus 代充 只需 145combined = { 

”user“: state[”user_data“], ”product“: state[”product_data“], ”analytics“: state[”analytics“]

} 

return {”result“: generate_report(combined)}

workflow = StateGraph(State)

添加并行任务

workflow.add_node(”fetch_user“, fetch_user_data) workflow.add_node(”fetch_product“, fetch_product_data) workflow.add_node(”fetch_analytics“, fetch_analytics)

添加聚合节点 - 标记为 deferred

workflow.add_node( ”combine“,

GPT plus 代充 只需 145combine_all_data, deferred=True# 等待所有上游完成 

)

三个任务并行启动

workflow.add_edge(START, ”fetch_user“) workflow.add_edge(START, ”fetch_product“) workflow.add_edge(START, ”fetch_analytics“)

都连接到 combine，但 combine 会等待全部完成

workflow.add_edge(”fetch_user“, ”combine“) workflow.add_edge(”fetch_product“, ”combine“) workflow.add_edge(”fetch_analytics“, ”combine“) workflow.add_edge(”combine“, END)

app = workflow.compile()

6.6 LangGraph Supervisor (多代理协调)

简化层级多代理系统

pip install langgraph-supervisor from langgraph_supervisor import create_supervisor from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model=”claude-3-5-sonnet-“)

定义专家 Agent

researcher = create_agent( ”researcher“,

GPT plus 代充 只需 145llm, tools=[web_search, academic_search], system_prompt="你是研究专家" 

)

coder = create_agent( ”coder“,

llm, tools=[python_repl, code_review], system_prompt="你是编程专家" 

)

writer = create_agent( ”writer“,

GPT plus 代充 只需 145llm, tools=[grammar_check], system_prompt="你是写作专家" 

)

创建 Supervisor 协调所有 Agent

supervisor = create_supervisor(

agents={ 

”researcher“: researcher, ”coder“: coder, ”writer“: writer

GPT plus 代充 只需 145}, workflow_type="dynamic", # 动态决定调用顺序 llm=llm # Supervisor 使用 LLM 做决策 

)

Supervisor 会自动协调

result = supervisor.invoke({ ”task“: ”研究最新的 AI 框架，写代码示例，然后写一篇教程“ })

不同的 Supervisor 模式：

# 1. 顺序模式 supervisor = create_supervisor(

GPT plus 代充 只需 145agents={...}, workflow_type="sequential", sequence=["researcher", "coder", "writer"] 

)

2. 并行模式

supervisor = create_supervisor(

agents={...}, workflow_type="parallel", parallel_tasks=["research", "data_collection", "analysis"] 

)

3. 层级模式

supervisor = create_supervisor(

GPT plus 代充 只需 145agents={ 

”manager“: manager_agent, ”workers“: [worker1, worker2, worker3]

}, workflow_type="hierarchical" 

)

7.1 状态设计

✅ 推荐:

GPT plus 代充 只需 145classGoodState(TypedDict):

messages: Annotated[list, add_messages] # 使用消息列表 metadata: dict # 元数据 step_count: int # 计数器 

❌ 不推荐:

GPT plus 代充 只需 145# 状态过于复杂 classBadState(TypedDict):

msg1: str msg2: str msg3: str 

… 太多字段

7.2 节点设计

✅ 推荐: 单一职责

GPT plus 代充 只需 145deffetch_data(state): ”“”只负责获取数据“”“

data = api_call() 

return {”data“: data}

defprocess_data(state): ”“”只负责处理数据“”“

GPT plus 代充 只需 145processed = process(state["data"]) 

return {”result“: processed}

❌ 不推荐: 节点做太多事情

defdo_everything(state):

获取数据、处理、保存、通知…

pass

7.3 何时使用 Command vs Conditional Edges

使用 Command：

• ✅ 节点内部逻辑决定路由
• ✅ 需要同时更新状态和路由
• ✅ 动态、复杂的决策树

使用 Conditional Edges：

• ✅ 简单的二选一路由
• ✅ 路由逻辑独立于节点
• ✅ 多个节点共用同一路由逻辑

GPT plus 代充 只需 145# 适合用 Command defcomplex_node(state):

result = process(state) 

if result.needs_review: return Command(update={…}, goto=”review“) elif result.needs_retry: return Command(update={…}, goto=”retry“) else: return Command(update={…}, goto=”finalize“)

适合用 Conditional Edges

defsimple_router(state): return”option_a“if state[”flag“] else”option_b“

workflow.add_conditional_edges(”node“, simple_router, {…})

7.4 错误处理

GPT plus 代充 只需 145defrobust_node(state: AgentState): try:

 result = risky_operation(state) 

return {”messages“: [result], ”error“: None} except Exception as e: return { ”messages“: [HumanMessage(content=f”错误: {str(e)}“)], ”error“: str(e)

GPT plus 代充 只需 145 } 

使用 Command 处理错误路由

defrobust_node_with_command(state): try:

 result = risky_operation(state) 

return Command(

GPT plus 代充 只需 145 update={"result": result}, goto="success" ) 

except Exception as e: return Command(

 update={"error": str(e)}, goto="error_handler" ) 

7.5 调试技巧

GPT plus 代充 只需 145# 1. 可视化图结构 from IPython.display import Image, display

display(Image(app.get_graph().draw_mermaid_png()))

2. 打印中间状态

defdebug_node(state):

print(f"当前状态: {state}") 

return state

workflow.add_node(”debug“, debug_node)

3. 使用 LangSmith 追踪

import os os.environ[”LANGCHAIN_TRACING_V2“] = ”true“ os.environ[”LANGCHAIN_API_KEY“] = ”your-api-key“

4. 查看执行历史

config = {”configurable“: {”thread_id“: ”1“}} app.invoke(input, config)

获取状态历史

history = app.get_state_history(config) for state in history:

GPT plus 代充 只需 145print(state) 

模式1: ReAct Agent (推理-行动循环)

defreact_pattern(): ”“”思考 -> 行动 -> 观察 -> 思考…“”“

defagent_think(state):

LLM 决定下一步行动

GPT plus 代充 只需 145 response = llm.invoke(state["messages"]) 

return {”messages“: [response]}

defagent_act(state):

执行工具

 last_message = state["messages"][-1] 

if hasattr(last_message, ‘tool_calls’):

执行工具…

return {”messages“: [tool_result]} return state

GPT plus 代充 只需 145workflow = StateGraph(AgentState) workflow.add_node("think", agent_think) workflow.add_node("act", agent_act) workflow.add_conditional_edges( 

”think“, lambda s: ”act“if has_tool_calls(s) else END,

 {"act": "act", END: END} ) workflow.add_edge("act", "think") 

模式2: Map-Reduce

GPT plus 代充 只需 145defmap_reduce_pattern(): ”“”并行处理多个任务，然后汇总结果“”“

classMapReduceState(TypedDict):

 inputs: list mapped_results: list final_result: str 

defmap_node(state):

并行处理每个输入

GPT plus 代充 只需 145 results = [process(item) for item in state["inputs"]] 

return {”mapped_results“: results}

defreduce_node(state):

汇总结果

 final = combine(state["mapped_results"]) 

return {”final_result“: final}

GPT plus 代充 只需 145workflow = StateGraph(MapReduceState) workflow.add_node("map", map_node) workflow.add_node("reduce", reduce_node) workflow.add_edge("map", "reduce") 

模式3: 审批流（使用 Command + interrupt）

defmodern_approval_workflow(): ”“”现代化审批流 - 使用新特性“”“

defsubmit(state): ”“”提交请求“”“ return Command(

GPT plus 代充 只需 145 update={"status": "pending"}, goto="auto_review" ) 

defauto_review(state): ”“”自动审核“”“

 risk = assess_risk(state["request"]) 

if risk == ”low“: return Command(

GPT plus 代充 只需 145 update={"status": "auto_approved"}, goto="approved" ) 

else: return Command(goto=”human_review“)

defhuman_review(state): ”“”人工审核“”“

 decision = interrupt({ 

”request“: state[”request“], ”risk“: ”high“, ”question“: ”是否批准？“

GPT plus 代充 只需 145 }) 

if decision == ”approve“: return Command(

 update={"status": "approved"}, goto="approved" ) 

else: return Command(

GPT plus 代充 只需 145 update={"status": "rejected"}, goto="rejected" ) 

defapproved(state):

 execute_action(state["request"]) 

return {}

defrejected(state):

GPT plus 代充 只需 145 notify_rejection(state["request"]) 

return {}

workflow = StateGraph(State) workflow.add_node("submit", submit) workflow.add_node("auto_review", auto_review) workflow.add_node("human_review", human_review) workflow.add_node("approved", approved) workflow.add_node("rejected", rejected) workflow.set_entry_point("submit") app = workflow.compile(checkpointer=MemorySaver()) 

练习1: 基础 - 数学教师

创建一个数学教师 Agent：

1. 接收数学问题
2. 解答问题
3. 提供详细步骤

提示: 使用基本的 StateGraph + Tool

练习2: 中级 - 内容审核（使用 Command）

创建一个内容审核系统：

1. 接收用户输入
2. 检查敏感内容
3. 安全则处理，不安全则拒绝
4. 使用 Command 处理路由

练习3: 高级 - 代码审查流程（使用 Command + interrupt）

创建一个自动化代码审查系统：

1. 代码生成 Agent
2. 自动代码审查
3. 如果发现问题，使用 interrupt() 请求人工确认
4. 使用 Command 控制流程
5. 只有通过所有检查才输出

从 0.x 升级到 1.0

GPT plus 代充 只需 145# 升级 pip install –upgrade langgraph

检查版本

python -c ”import langgraph; print(langgraph.version)“

主要变化

1. langgraph.prebuilt 已弃用

# ❌ 旧代码 from langgraph.prebuilt import create_react_agent

✅ 新代码

from langchain.agents import create_react_agent

1. 推荐使用新特性

• 优先使用 Command 而不是复杂的 conditional_edges
• 优先使用 interrupt() 而不是 interrupt_before
• 考虑使用 Node Caching 优化性能

1. 其他都向后兼容

• 所有核心 API 保持不变
• 旧代码依然可以运行

最小化模板（经典版）

GPT plus 代充 只需 145from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated from langgraph.graph.message import add_messages

classState(TypedDict):

messages: Annotated[list, add_messages] 

defmy_node(state: State):

你的逻辑

return {”messages“: […]}

workflow = StateGraph(State) workflow.add_node(”node1“, my_node) workflow.set_entry_point(”node1“) workflow.add_edge(”node1“, END) app = workflow.compile()

result = app.invoke({”messages“: […]})

最小化模板（使用新特性）

GPT plus 代充 只需 145from langgraph.graph import StateGraph, END from langgraph.types import Command, interrupt from typing import TypedDict

classState(TypedDict):

data: str approved: bool 

defprocess_node(state: State):

使用 interrupt 获取审批

GPT plus 代充 只需 145decision = interrupt(f"处理 {state['data']}?") 

使用 Command 控制流程

if decision == ”yes“: return Command(

 update={"approved": True}, goto="execute" ) 

else: return Command(goto=END)

workflow = StateGraph(State) workflow.add_node(”process“, process_node) workflow.add_node(”execute“, execute_node) workflow.set_entry_point(”process“)

from langgraph.checkpoint.memory import MemorySaver app = workflow.compile(checkpointer=MemorySaver())

第1周：基础

• Day 1-2: 理解核心概念（State, Nodes, Edges）
• Day 3-4: 实践示例1和示例2
• Day 5-7: 完成示例3，理解多代理

第2周：新特性

• Day 1-2: 学习 Command API，重构之前的代码
• Day 3-4: 学习 interrupt()，实现 HITL
• Day 5-7: 探索其他新特性（缓存、Supervisor等）

第3周：实战

• 开始实际项目
• 应用**实践
• 完成练习题

• 官方文档: https://langchain-ai.github.io/langgraph/
• GitHub: https://github.com/langchain-ai/langgraph
• Changelog: https://changelog.langchain.com/
• LangSmith (调试工具): https://smith.langchain.com/
• 社区论坛: https://forum.langchain.com/

---

result = app.invoke({“messages”: […]})

GPT plus 代充 只需 145

最小化模板（使用新特性）

plaintext from langgraph.graph import StateGraph, END from langgraph.types import Command, interrupt from typing import TypedDict

classState(TypedDict):

data: str approved: bool 

defprocess_node(state: State):

使用 interrupt 获取审批

GPT plus 代充 只需 145decision = interrupt(f"处理 {state['data']}?") 

使用 Command 控制流程

if decision == ”yes“: return Command(

 update={"approved": True}, goto="execute" ) 

else: return Command(goto=END)

workflow = StateGraph(State) workflow.add_node(”process“, process_node) workflow.add_node(”execute“, execute_node) workflow.set_entry_point(”process“)

from langgraph.checkpoint.memory import MemorySaver app = workflow.compile(checkpointer=MemorySaver())

第1周：基础

• Day 1-2: 理解核心概念（State, Nodes, Edges）
• Day 3-4: 实践示例1和示例2
• Day 5-7: 完成示例3，理解多代理

第2周：新特性

• Day 1-2: 学习 Command API，重构之前的代码
• Day 3-4: 学习 interrupt()，实现 HITL
• Day 5-7: 探索其他新特性（缓存、Supervisor等）

第3周：实战

• 开始实际项目
• 应用**实践
• 完成练习题

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

第一阶段（10天）：初阶应用

该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。

• 大模型 AI 能干什么？
• 大模型是怎样获得「智能」的？
• 用好 AI 的核心心法
• 大模型应用业务架构
• 大模型应用技术架构
• 代码示例：向 GPT-3.5 灌入新知识
• 提示工程的意义和核心思想
• Prompt 典型构成
• 指令调优方法论
• 思维链和思维树
• Prompt 攻击和防范
• …

第二阶段（30天）：高阶应用

该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。

• 为什么要做 RAG
• 搭建一个简单的 ChatPDF
• 检索的基础概念
• 什么是向量表示（Embeddings）
• 向量数据库与向量检索
• 基于向量检索的 RAG
• 搭建 RAG 系统的扩展知识
• 混合检索与 RAG-Fusion 简介
• 向量模型本地部署
• …

第三阶段（30天）：模型训练

恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。

到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？

• 为什么要做 RAG
• 什么是模型
• 什么是模型训练
• 求解器 & 损失函数简介
• 小实验2：手写一个简单的神经网络并训练它
• 什么是训练/预训练/微调/轻量化微调
• Transformer结构简介
• 轻量化微调
• 实验数据集的构建
• …

第四阶段（20天）：商业闭环

对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。

• 硬件选型
• 带你了解全球大模型
• 使用国产大模型服务
• 搭建 OpenAI 代理
• 热身：基于阿里云 PAI 部署 Stable Diffusion
• 在本地计算机运行大模型
• 大模型的私有化部署
• 基于 vLLM 部署大模型
• 案例：如何优雅地在阿里云私有部署开源大模型
• 部署一套开源 LLM 项目
• 内容安全
• 互联网信息服务算法备案
• …

学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。

如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】