01_basic_concepts.py

"""
LangGraph 基础概念教程
====================

本教程介绍 LangGraph 的核心概念：
1. StateGraph - 状态图
2. Nodes - 节点
3. Edges - 边
4. Conditional Edges - 条件边
5. State - 状态
"""

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator

# 1. 定义状态类型
class BasicState(TypedDict):
    """基础状态定义"""
    messages: Annotated[list, operator.add]  # 消息列表，使用操作符进行累加
    counter: int  # 计数器
    result: str  # 结果

# 2. 创建节点函数
def node_a(state: BasicState) -> dict:
    """节点A：处理消息并增加计数器"""
    print("执行节点A")
    return {
        "messages": [{"role": "system", "content": "来自节点A的消息"}],
        "counter": state["counter"] + 1,
        "result": "A完成"
    }

def node_b(state: BasicState) -> dict:
    """节点B：处理消息并增加计数器"""
    print("执行节点B")
    return {
        "messages": [{"role": "user", "content": "来自节点B的消息"}],
        "counter": state["counter"] + 2,
        "result": "B完成"
    }

def node_c(state: BasicState) -> dict:
    """节点C：最终处理节点"""
    print("执行节点C")
    return {
        "messages": [{"role": "assistant", "content": f"最终结果: {state['result']}"}],
        "counter": state["counter"] * 2,
        "result": "处理完成"
    }

# 3. 条件判断函数
def should_continue(state: BasicState) -> str:
    """根据计数器决定下一步"""
    if state["counter"] < 5:
        return "continue_to_b"
    else:
        return "end"

# 4. 构建图
def build_basic_graph():
    """构建基础图"""
    # 创建状态图
    workflow = StateGraph(BasicState)
    
    # 添加节点
    workflow.add_node("node_a", node_a)
    workflow.add_node("node_b", node_b)
    workflow.add_node("node_c", node_c)
    
    # 设置入口点
    workflow.set_entry_point("node_a")
    
    # 添加条件边
    workflow.add_conditional_edges(
        "node_a",
        should_continue,
        {
            "continue_to_b": "node_b",
            "end": "node_c"
        }
    )
    
    # 添加普通边
    workflow.add_edge("node_b", "node_c")
    workflow.add_edge("node_c", END)
    
    # 编译图
    return workflow.compile()

# 5. 运行示例
def run_basic_example():
    """运行基础示例"""
    print("=" * 50)
    print("LangGraph 基础概念示例")
    print("=" * 50)
    
    # 构建图
    graph = build_basic_graph()
    
    # 初始状态
    initial_state = {
        "messages": [],
        "counter": 0,
        "result": "初始状态"
    }
    
    print("\n初始状态:", initial_state)
    
    # 执行图
    result = graph.invoke(initial_state)
    
    print("\n执行结果:")
    print(f"最终计数器: {result['counter']}")
    print(f"最终结果: {result['result']}")
    print(f"消息数量: {len(result['messages'])}")
    
    print("\n所有消息:")
    for i, msg in enumerate(result["messages"]):
        print(f"  {i+1}. {msg['role']}: {msg['content']}")
    
    return result

# 6. 可视化图结构
def visualize_graph():
    """可视化图结构"""
    try:
        import networkx as nx
        import matplotlib.pyplot as plt
        
        graph = build_basic_graph()
        
        # 创建NetworkX图
        G = nx.DiGraph()
        
        # 添加节点
        for node in graph.nodes:
            G.add_node(node)
        
        # 添加边（简化版本）
        G.add_edge("node_a", "node_b")
        G.add_edge("node_a", "node_c")
        G.add_edge("node_b", "node_c")
        G.add_edge("node_c", "END")
        
        # 绘制图
        plt.figure(figsize=(10, 8))
        pos = nx.spring_layout(G, seed=42)
        nx.draw(G, pos, with_labels=True, node_color='lightblue', 
                node_size=3000, font_size=10, font_weight='bold')
        plt.title("LangGraph 基础图结构")
        plt.show()
        
        print("图结构已可视化（需要GUI环境）")
        
    except ImportError:
        print("需要安装 networkx 和 matplotlib 进行可视化")
        print("运行: pip install networkx matplotlib")

if __name__ == "__main__":
    # 运行基础示例
    result = run_basic_example()
    
    print("\n" + "=" * 50)
    print("基础概念总结:")
    print("1. StateGraph: 管理状态和节点的工作流")
    print("2. Nodes: 执行具体任务的函数")
    print("3. Edges: 定义节点之间的流向")
    print("4. Conditional Edges: 根据条件决定下一步")
    print("5. State: 在节点之间传递的数据")
    print("=" * 50)

requirements.txt

LangGraph 核心包

langgraph>=0.2.0
langchain-core>=0.2.0
langchain-openai>=0.1.0

OpenAI SDK

openai>=1.0.0

类型和数据处理

pydantic>=2.0.0
typing-extensions>=4.0.0

可视化和图论

networkx>=3.0
matplotlib>=3.0

Jupyter 支持（可选）

jupyter>=1.0.0

ipykernel>=6.0.0

YAML 支持（可选）

pyyaml>=6.0