大数据

langraph入门示例

Posted by

demo01:

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel


# 1. 定义状态类型（所有字段可选，便于增量更新）
class MyState(BaseModel):
    a: str | None = None
    b: str | None = None
    c: str | None = None


# 2. 定义节点函数（仅返回要更新的字段字段）
def step_a(state: MyState):
    print("执行 step_a")
    return {"a": "done"}  # 只更新a


def step_b(state: MyState):
    print("执行 step_b")
    return {"b": "done"}  # 只更新b


def step_c(state: MyState):
    print("执行 step_c")
    return {"c": "done"}  # 只更新c


# 3. 构建图
builder = StateGraph(MyState)  # 指定状态类型
builder.add_node("a", step_a)
builder.add_node("b", step_b)
builder.add_node("c", step_c)

# 定义顺序边
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", "c")
builder.add_edge("c", END)

# 4. 编译图
graph = builder.compile()

# 5. 执行
initial_state = MyState()  # 所有字段为None
final_state = graph.invoke(initial_state)
print("最终状态", final_state)

demo02:

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel
import random


# 1. 定义状态类型（所有字段可选，便于增量更新）
class MyState(BaseModel):
    a: str | None = None
    b: str | None = None
    c: str | None = None

    continue_flag: bool | None = None  # 用于路由的标志


# 2. 定义节点函数（仅返回要更新的字段字段）
# 注意：在 LangGraph 中，当一个节点执行完毕后，系统会使用该节点的返回值来更新全局状态，
# 而不是直接使用函数体内对 state 对象所做的修改。
def step_a(state: MyState):
    print("执行 step_a")
    # 随机决定是否继续
    # 函数体内对 state 属性的直接修改
    # 这些修改会影响传入的 state 对象本身，但不会自动传播到后续节点，
    # 因为节点结束后，系统会丢弃这个被修改的对象，转而使用返回值来构建新的状态
    # 多余的操作，state.continue_flag = random.choice([True, False])
    flag = random.choice([True, False])
    # 对于 Pydantic 状态模型
    # 如果节点返回一个 字典，则字典中的键值对会与当前状态合并（覆盖对应字段，其他字段保持不变）。
    # 如果节点返回 None 或空字典，则状态不发生任何变化。
    return {"a": "done", "continue_flag": flag}  # 只更新a


def step_b(state: MyState):
    print("执行 step_b")
    return {"b": "done"}  # 只更新b


def step_c(state: MyState):
    print("执行 step_c")
    return {"c": "done"}  # 只更新c


# 路由函数：根据状态决定下一个节点
def route_after_b(state: MyState):
    if state.continue_flag:
        return "c"
    else:
        return END


# 3. 构建图
builder = StateGraph(MyState)  # 指定状态类型
builder.add_node("a", step_a)
builder.add_node("b", step_b)
builder.add_node("c", step_c)

# 定义顺序边
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_conditional_edges("b", route_after_b)  # 条件边

# 4. 编译图
graph = builder.compile()

# 5. 执行
initial_state = MyState()  # 所有字段为None
final_state = graph.invoke(initial_state)
print("最终状态", final_state)

demo03: 

from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel
import random


# 1. 定义状态类型（所有字段可选，便于增量更新）
class MyState(BaseModel):
    a: str | None = None
    b: str | None = None
    c: str | None = None

    continue_flag: bool | None = None  # 用于路由的标志


def step_a(state: MyState):
    print(f"[step_a] 刚进来时拿到的 state: {state}")  # 此时全是 None
    flag = random.choice([True, False])
    # 只返回a和flag
    return {"a": "done", "continue_flag": flag}


def step_b(state: MyState):
    # 注意这里的state，它包含了step_a返回的所有字段
    print(f"[step_b] 拿到的 state 包含 a 和 flag: {state}")
    # 读取 step_a 设置的 flag
    if state.continue_flag:
        print("  step_b 发现 continue_flag 为 True，继续工作")
    else:
        print("  step_b 发现 continue_flag 为 False")
    return {"b": "done"}


def step_c(state: MyState):
    print(f"[step_c] 拿到的 state 包含 a, b, flag: {state}")
    return {"c": "done"}


# 3. 构建图
builder = StateGraph(MyState)  # 指定状态类型
builder.add_node("a", step_a)
builder.add_node("b", step_b)
builder.add_node("c", step_c)

# 定义顺序边
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", "c")
builder.add_edge("c", END)

# 4. 编译图
graph = builder.compile()

# 5. 执行
initial_state = MyState()  # 所有字段为None
final_state = graph.invoke(initial_state)
print("最终状态", final_state)

demo04:

import asyncio

from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime
from pydantic import BaseModel


# 1. 定义状态类型（所有字段可选，便于增量更新）
class MyState(BaseModel):
    a: str | None = None
    b: str | None = None


def step_a(state: MyState, runtime: Runtime):
    runtime.stream_writer("step_a 开始处理...")
    return {"a": "done"}


def step_b(state: MyState, runtime: Runtime):
    runtime.stream_writer("step_b 开始处理...")
    return {"b": "done"}


# 3. 构建图
builder = StateGraph(MyState)  # 指定状态类型
builder.add_node("a", step_a)
builder.add_node("b", step_b)

# 定义顺序边
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", END)

# 4. 编译图
graph = builder.compile()

# 流式接收自定义事件
# for chunk in graph.stream(MyState(), stream_mode="custom"):
#     print("收到事件：", chunk)

async def main():
    async for event in graph.astream(
        MyState(),
        stream_mode=["custom", "values"]
    ):
        # event是一个元组(mode, data)
        mode, data = event
        if mode == "custom":
            print(f"自定义事件: {data}")
        elif mode == "values":
            print(f"状态更新: {data}")
            # 最后一条 data 就是最终状态
asyncio.run(main())
上一篇 角色素材库：小帅、小美