图结构概念

LangGraph 的核心是图（Graph），它定义了智能体的执行流程。理解图结构是掌握 LangGraph 的关键。

📊 核心组件

1. 图 (Graph)

• 定义: 图是节点和边的集合，代表了智能体的工作流程。
• 类型:
  • StateGraph: 最常用的图类型，用于管理状态。
  • MessageGraph: 专门用于处理消息列表的图。

2. 状态 (State)

• 定义: 在图的整个执行过程中共享的数据结构。
• 实现: 通常使用 TypedDict 来定义，确保类型安全。
• 作用:
  • 在节点间传递数据。
  • 记录智能体的上下文和历史。
  • 控制流程的分支和循环。

from typing import TypedDict, List

class AgentState(TypedDict):
    query: str
    messages: List[str]
    result: str

3. 节点 (Nodes)

• 定义: 图中的执行单元，通常是一个 Python 函数或可调用对象。
• 作用:
  • 执行具体的业务逻辑（如调用 LLM、访问 API）。
  • 读取和更新图的状态。
• 签名: 节点函数接收 state 作为参数，并返回一个字典来更新 state。

def my_node(state: AgentState) -> dict:
    # 读取状态
    query = state["query"]
    
    # 执行逻辑
    result = f"处理完成: {query}"
    
    # 返回更新
    return {"result": result}

4. 边 (Edges)

• 定义: 连接节点的有向线，定义了执行的流向。
• 类型:
  • 固定边: workflow.add_edge("node_A", "node_B")
  • 条件边: workflow.add_conditional_edges(...)
  • 结束边: workflow.add_edge("node_C", END)

🏗️ 构建图的步骤

1. 实例化图

from langgraph.graph import StateGraph

workflow = StateGraph(AgentState)

2. 添加节点

workflow.add_node("my_node_name", my_node_function)

3. 设置入口点

workflow.set_entry_point("my_node_name")

4. 添加边

固定边

从一个节点总是流向另一个节点。

workflow.add_edge("node_A", "node_B")

条件边

根据节点的输出或当前状态，决定下一个节点。

def should_continue(state: AgentState) -> str:
    if len(state["messages"]) > 5:
        return "end"
    else:
        return "continue"

workflow.add_conditional_edges(
    "start_node",
    should_continue,
    {
        "continue": "continue_node",
        "end": END,
    },
)

5. 编译图

app = workflow.compile()

编译后的 app 是一个可执行的 LangChain Runnable 对象。

🔄 循环

LangGraph 的一个强大功能是支持循环。通过条件边，可以轻松实现循环逻辑。

示例：多轮对话

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class ChatState(TypedDict):
    messages: List[str]

def user_input_node(state: ChatState) -> dict:
    # 模拟用户输入
    return {"messages": state["messages"] + ["用户: 你好"]}

def agent_response_node(state: ChatState) -> dict:
    # 模拟智能体响应
    return {"messages": state["messages"] + ["智能体: 你好！有什么可以帮忙的吗？"]}

def should_exit(state: ChatState) -> str:
    if len(state["messages"]) >= 4:
        return "exit"
    return "continue"

workflow = StateGraph(ChatState)
workflow.add_node("user", user_input_node)
workflow.add_node("agent", agent_response_node)
workflow.set_entry_point("user")
workflow.add_edge("user", "agent")
workflow.add_conditional_edges(
    "agent",
    should_exit,
    {
        "continue": "user",
        "exit": END,
    },
)

app = workflow.compile()
result = app.invoke({"messages": []})
# result['messages'] 将包含4条消息

🚀 可视化

可以使用 get_graph() 方法生成图的可视化表示。

from IPython.display import Image

# 需要安装 graphviz: pip install graphviz
graph_image = app.get_graph().draw_mermaid_png()
Image(graph_image)

这将生成一个 Mermaid 格式的图，可以直观地看到工作流程。

🎯 总结

• 图: 定义工作流程。
• 状态: 在流程中传递数据。
• 节点: 执行具体任务。
• 边: 控制流程方向。
• 循环: 通过条件边实现。

理解这些核心概念是构建复杂、有状态的智能体应用的基础。

---

💡 下一步: 学习 状态管理 来深入了解如何在图中有效传递和更新数据。