LangGraph

from typing import TypedDict

class MyState(TypedDict):
    question: str   # 用户问题
    answer: str     # 生成的回答
    retries: int    # 已重试次数

def generate(state: MyState) -> dict:
    # 读取用户问题，生成回答
    return {"answer": f"关于 {state['question']} 的回答"}

def check(state: MyState) -> dict:
    # 累加重试次数
    return {"retries": state["retries"] + 1}

# 条件函数：返回字符串键名，映射到对应节点
def should_retry(state: MyState) -> str:
    if state["retries"] < 3 and state["answer"] == "":
        return "retry"   # 重试
    return "done"        # 结束

pip install -U langgraph

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

# 1. 定义共享状态
class MyState(TypedDict):
    text: str

# 2. 定义节点函数
def step_a(state: MyState) -> dict:
    return {"text": state["text"] + " -> 经过节点A"}

def step_b(state: MyState) -> dict:
    return {"text": state["text"] + " -> 经过节点B"}

# 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()
result = graph.invoke({"text": "开始"})
print(result["text"])

开始 -> 经过节点A -> 经过节点B

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

class RetryState(TypedDict):
    value: int
    count: int

def increment(state: RetryState) -> dict:
    return {"value": state["value"] + 1, "count": state["count"] + 1}

# 条件函数：count < 3 继续循环，否则结束
def check(state: RetryState) -> str:
    return "loop" if state["count"] < 3 else "done"

builder = StateGraph(RetryState)
builder.add_node("inc", increment)
builder.add_edge(START, "inc")
builder.add_conditional_edges("inc", check, {"loop": "inc", "done": END})

graph = builder.compile()
result = graph.invoke({"value": 0, "count": 0})
print(result)  # {'value': 3, 'count': 3}