创建第一个图

from typing import TypedDict

from lingxigraph import END, START, Runtime, StateGraph


class SupportState(TypedDict):
    request: str
    normalized: str
    result: str


class AppContext(TypedDict):
    tenant: str


def normalize(state: SupportState):
    return {"normalized": state["request"].strip().lower()}


def resolve(state: SupportState, runtime: Runtime[AppContext]):
    runtime.stream_writer({"stage": "resolved"})
    return {
        "result": f"{runtime.context['tenant']}: {state['normalized']}"
    }


builder = StateGraph(
    SupportState,
    context_schema=AppContext,
    name="support",
    version="1.0.0",
)
builder.add_node("normalize", normalize)
builder.add_node("resolve", resolve, timeout=30)
builder.add_edge(START, "normalize")
builder.add_edge("normalize", "resolve")
builder.add_edge("resolve", END)
graph = builder.compile()

initial = {"request": "  RESET ACCESS  ", "normalized": "", "result": ""}
print(graph.invoke(initial, context={"tenant": "acme"}))

python support_graph.py

for update in graph.stream(
    initial,
    context={"tenant": "acme"},
    stream_mode="updates",
):
    print(update)