流式输出与人工审批

async for mode, chunk in graph.astream(
    input_state,
    context={"tenant": "acme"},
    stream_mode=["updates", "custom", "messages"],
):
    print(mode, chunk)

def retrieve(state, runtime):
    runtime.stream_writer({"stage": "searching"})
    docs = search(state["query"])
    runtime.stream_writer({"stage": "found", "count": len(docs)})
    return {"documents": docs}

from lingxigraph import interrupt


def approve(state):
    decision = interrupt({
        "kind": "approval",
        "action": state["proposed_action"],
    })
    return {"approved": bool(decision["approved"])}

from lingxigraph import Command

config = {"configurable": {"thread_id": "ticket-42"}}
graph.invoke(initial, config=config)
graph.invoke(Command(resume={"approved": True}), config=config)

curl -X POST http://localhost:8124/v1/runs/$RUN_ID/resume \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"resume":{"approved":true}}'