LangGraph入门知识学习笔记 - AI创想

"""
LangGraph 简单案例HelloWorld：
构建一个最小的有向图，流程是：START → 模型节点 → END
LangGraph的灵魂：State(状态) + Nodes(节点) + Edges(边) + Graph(图)
"""
import uuid
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
import os
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
# ========== 1. 定义状态（State） ==========
# 存储对话消息
class AtguiguState(TypedDict):
# messages 是一个消息列表，Annotated + add_messages 表示支持自动追加消息
messages: Annotated[List, add_messages]
# ========== 2. 定义大模型 ==========
llm = init_chat_model(
model="qwen-plus",
model_provider="openai",
api_key=os.getenv("aliQwen-api"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
# ========== 3. 定义节点函数 ==========
# 节点：调用大模型，并把回复加入到 state["messages"] 里
def model_node(state: AtguiguState):
reply = llm.invoke(state["messages"]) # 输入历史消息，调用模型
return {"messages": [reply]} # 返回新消息，自动加到 state
# ========== 4. 构建图结构 ==========
graph = StateGraph(AtguiguState) # 初始化图，指定 State 类型
graph.add_node("model", model_node) # 添加一个节点，名字叫 "model"
graph.add_edge(START, "model") # 从 START 到 "model"
graph.add_edge("model", END) # 从 "model" 到 END
# 打印图的边和节点信息
#print(graph.edges)
print()
#print(graph.nodes)
# ========== 5. 编译==========
app = graph.compile()
# ========== 6. 运行 ==========
#result = app.invoke({"messages": [HumanMessage(content="请用一句话解释什么是 LangGraph。")]})
result = app.invoke({"messages": "请用一句话解释什么是 LangGraph。"})
# 打印模型的最后一条回复
print("模型回答：", result["messages"][-1].content)
print()
# =========================
#1. 打印图的ascii可视化结构
print(app.get_graph().print_ascii())
print("="*50)
#2. 打印图的Mermaid代码可视化结构并通过https://www.processon.com/mermaid编辑器查看
print(app.get_graph().draw_mermaid())
print("="*50)
#3. 生成 PNG并写入文件
# png_bytes = app.get_graph().draw_mermaid_png()
# output_path = "langgraph" + str(uuid.uuid4())[:8] + ".png"
# with open(output_path, "wb") as f:
# f.write(png_bytes)
# print(f"图片已生成：{output_path}")

复制代码

"""
SendDemo.py
LangGraph Map-Reduce 模式演示
通过使用 Send 对象，LangGraph 提供了一种优雅的方式来实现这种动态图结构，
使得我们可以根据运行时状态来决定执行路径。
解释：
（1）首先执行 generate_subjects主题列表节点，生成主题列表：['猫', '狗', '程序员']
（2）然后通过条件边函数 map_subjects_to_jokes 为每个主题创建一个 Send 对象
（3）make_joke 节点被并行执行3次，每次处理一个主题
（4）最终将所有生成的笑话合并到一个列表中
这种模式非常适合处理动态数量的任务
"""
from typing import Annotated, List, Sequence
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
# 定义状态
class OverallState(TypedDict):
subjects: List[str]
jokes: Annotated[List[str], lambda x, y: x + y] # 使用列表合并的方式
# 第一个节点：生成需要处理的主题列表
def generate_subjects(state: OverallState) -> dict:
"""生成需要处理的主题列表"""
print("执行节点: generate_subjects")
subjects = ["猫", "狗", "程序员"]
print(f"生成主题列表: {subjects}")
return {"subjects": subjects}
# Map节点：为每个主题生成笑话
def make_joke(state: OverallState) -> dict:
"""为单个主题生成笑话"""
subject = state.get("subject", "未知")
print(f"执行节点: make_joke，处理主题: {subject}")
# 根据主题生成相应笑话
jokes_map = {
"猫": "为什么猫不喜欢在线购物？因为它们更喜欢实体店！",
"狗": "为什么狗不喜欢计算机？因为它们害怕被鼠标咬！",
"程序员": "为什么程序员喜欢洗衣服？因为他们在寻找bugs！",
"未知": "这是一个关于未知主题的神秘笑话。"
}
joke = jokes_map.get(subject, f"这是一个关于{subject}的即兴笑话。")
print(f"生成笑话: {joke}")
return {"jokes": [joke]}
# 条件边函数：根据主题列表生成Send对象列表
def map_subjects_to_jokes(state: OverallState) -> Sequence[Send]:
"""将主题列表映射到joke生成任务"""
print("执行条件边函数: map_subjects_to_jokes")
subjects = state["subjects"]
print(f"映射主题到joke任务: {subjects}")
# 为每个主题创建一个Send对象，指向make_joke节点
# 每个Send对象包含节点名称和传递给该节点的状态
send_list = [Send("make_joke", {"subject": subject}) for subject in subjects]
print(f"生成Send对象列表: {send_list}")
return send_list
def main():
"""演示Map-Reduce模式"""
print("=== Map-Reduce 模式演示 ===\n")
# 创建图
builder = StateGraph(OverallState)
# 添加节点
builder.add_node("generate_subjects", generate_subjects)
builder.add_node("make_joke", make_joke)
# 添加边
builder.add_edge(START, "generate_subjects")
# 添加条件边，使用Send对象实现map-reduce
builder.add_conditional_edges(
"generate_subjects", # 源节点
map_subjects_to_jokes # 路由函数，返回Send对象列表
)
# 从make_joke到结束
builder.add_edge("make_joke", END)
# 编译图
graph = builder.compile()
print(graph.get_graph().print_ascii())
# 执行图
initial_state = {"subjects": [], "jokes": []}
print("初始状态:", initial_state)
print("\n开始执行图...")
result = graph.invoke(initial_state)
print(f"\n最终结果: {result}")
print("\n=== 演示完成 ===")
if __name__ == "__main__":
main()

复制代码

"""
CommandDemo.py | LangGraph 1.0.6 正式版
Command 基础演示：状态更新+流程控制+动态路由
"""
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
# 全局常量：统一递归限制，便于维护
RECURSION_LIMIT = 50
# 定义状态
class AgentState(TypedDict):
messages: Annotated[list, lambda x, y: x + y] # 自动合并消息
current_agent: str
task_completed: bool
# 决策代理（核心路由节点）
def decision_agent(state: AgentState) -> Command[AgentState]:
"""根据消息内容路由代理，任务完成则直接终止"""
print("执行节点: decision_agent")
# 优先终止流程（核心防循环逻辑）
if state["task_completed"]:
print("✅ 检测到任务已完成，直接终止流程")
return Command(
update={"messages": [("system", "所有任务处理完成，流程正常结束")]},
goto=END
)
# 提取消息文本（兼容空消息）
last_message = state["messages"][-1] if state["messages"] else ("", "")
last_msg_content = last_message[1]
print(f"最新消息文本: {last_msg_content}")
# 动态路由
if "数学" in last_msg_content:
print("✅ 检测到数学任务，路由到数学代理")
return Command(
update={"messages": [("system", "路由到数学代理")], "current_agent": "math_agent"},
goto="math_agent"
)
elif "翻译" in last_msg_content:
print("✅ 检测到翻译任务，路由到翻译代理")
return Command(
update={"messages": [("system", "路由到翻译代理")], "current_agent": "translation_agent"},
goto="translation_agent"
)
else:
print("❌ 未识别任务类型，标记任务完成并终止")
return Command(
update={"messages": [("system", "任务完成")], "task_completed": True},
goto=END
)
# 数学代理（业务节点）
def math_agent(state: AgentState) -> Command[AgentState]:
"""处理数学计算任务，完成后返回决策代理"""
print("执行节点: math_agent")
result = "2 + 2 = 4"
print(f"计算结果: {result}")
return Command(
update={
"messages": [("assistant", f"数学计算结果: {result}")],
"current_agent": "decision_agent",
"task_completed": True
},
goto="decision_agent"
)
# 翻译代理（业务节点）
def translation_agent(state: AgentState) -> Command[AgentState]:
"""处理中英翻译任务，完成后返回决策代理"""
print("执行节点: translation_agent")
translation = "Hello -> 你好"
print(f"翻译结果: {translation}")
return Command(
update={
"messages": [("assistant", f"翻译结果: {translation}")],
"current_agent": "decision_agent",
"task_completed": True
},
goto="decision_agent"
)
def main():
"""演示Command基础用法：状态更新+动态路由+流程终止"""
print("=== Command 基础演示（LangGraph 1.0.6）===\n")
# 1. 构建状态图
builder = StateGraph(AgentState)
builder.add_node("decision_agent", decision_agent)
builder.add_node("math_agent", math_agent)
builder.add_node("translation_agent", translation_agent)
# 2. 定义边（完整节点关系）
builder.add_edge(START, "decision_agent")
builder.add_edge("math_agent", "decision_agent")
builder.add_edge("translation_agent", "decision_agent")
builder.add_edge("decision_agent", END)
# 3. 编译图
graph = builder.compile()
# 测试1：数学任务
print("【测试1: 数学任务】")
initial_state = {"messages": [("user", "我需要计算数学题")], "current_agent": "user", "task_completed": False}
print("初始状态:", initial_state)
result = graph.invoke(initial_state, recursion_limit=RECURSION_LIMIT)
print("最终状态(简化):", {k: v for k, v in result.items() if k != "messages"}) # 简化输出
print("\n" + "-" * 50 + "\n")
# 测试2：翻译任务
print("【测试2: 翻译任务】")
initial_state = {"messages": [("user", "我需要翻译文本")], "current_agent": "user", "task_completed": False}
print("初始状态:", initial_state)
result = graph.invoke(initial_state, recursion_limit=RECURSION_LIMIT)
print("最终状态(简化):", {k: v for k, v in result.items() if k != "messages"})
print("\n" + "-" * 50 + "\n")
# 测试3：未识别任务
print("【测试3: 未识别任务类型】")
initial_state = {"messages": [("user", "你好")], "current_agent": "user", "task_completed": False}
print("初始状态:", initial_state)
result = graph.invoke(initial_state, recursion_limit=RECURSION_LIMIT)
print("最终状态(简化):", {k: v for k, v in result.items() if k != "messages"})
# 新增：可视化图结构（教学演示必备）
print("\n=== 图结构可视化 ===")
print(graph.get_graph().draw_mermaid())
if __name__ == "__main__":
main()

复制代码