LangGraph 基础：用 State、Reducer、Node、Edge 与 Command 管理流程

python -m pip install "langgraph>=1,<2"

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command


class InspectionState(TypedDict):
    device_id: str
    readings: Annotated[list[float], operator.add]
    average: float
    action: str


def collect(state: InspectionState) -> dict:
    # 返回增量；Reducer 会把新读数追加到原列表。
    return {"readings": [68.0, 72.0]}


def analyze(state: InspectionState) -> Command[Literal["review", "__end__"]]:
    average = sum(state["readings"]) / len(state["readings"])
    if average >= 70:
        return Command(update={"average": average, "action": "manual_review"}, goto="review")
    return Command(update={"average": average, "action": "none"}, goto=END)


def review(state: InspectionState) -> dict:
    return {"action": f"review_{state['device_id']}"}


builder = StateGraph(InspectionState)
builder.add_node("collect", collect)
builder.add_node("analyze", analyze)
builder.add_node("review", review)
builder.add_edge(START, "collect")
builder.add_edge("collect", "analyze")
builder.add_edge("review", END)
graph = builder.compile()

result = graph.invoke({
    "device_id": "AX-8",
    "readings": [],
    "average": 0.0,
    "action": "",
})
print(result)