Agentsop Streaming Output

Streaming Tool/Agent Output · SOP (Enhancement Overlay)

Source posture: every non-trivial claim is cited inline. Short tags like [lg/stream], [lc/astream-events], [oai/stream], [anthropic/stream], [mdn/sse] resolve against references/R1-source-evidence.md.

This is an ENHANCE overlay: it sits on top of [[agentsop-langgraph]] (which names the four stream modes but treats streaming as one of ten operations) and [[langchain]]. Read those for the orchestration; read this for the streaming SOP. Cross-link: [[agentsop-langgraph]] OP-8.

何时激活 (Activation Rules)

Activate when any of these fire:

• The run is long (multi-second to multi-minute agent loop, RAG over many docs, multi-tool chain) and the user is waiting — perceived latency, not total latency, is the product metric.
• The user asks to "stream the response", "show a typing effect", "show progress", "show which tool the agent is running", or "show the chain of thought".
• You are building a chat surface (stream final tokens) OR an agent surface (stream intermediate steps: node entered, tool called, partial state) OR a long task surface (stream custom progress like "embedded 40/200 docs").
• You must pick a transport: Server-Sent Events (SSE) vs WebSocket vs plain chunked HTTP, and handle client disconnect / cancellation cleanly.
• You're wiring graph.stream(...) / astream_events / OpenAI stream=True / Anthropic client.messages.stream and need to know which mode and what to forward to the client.

Do not activate for: a single fast (<1s) completion, a batch/offline job with no waiting human, or a pure front-end animation question (that's CSS, not a backend SOP). Streaming a 300ms call adds protocol overhead for zero UX gain — see 反模式.

核心心智模型 (Core Mental Model)

Stream what the user needs to see, not everything the engine emits. A backend stream is a curated projection of the run's internal event firehose onto exactly three audiences:

1. Chat audience → final tokens. A human reading prose wants character-by- character output of the final assistant message. In LangGraph this is stream_mode="messages" (LLM tokens + metadata); in raw SDKs it's stream=True / .messages.stream [lg/stream] [oai/stream] [anthropic/stream]. They do not want to see tool JSON or scratch nodes.

2. Agent audience → intermediate updates. A developer (or a power-user UI) watching an agent work wants "entered node planner", "calling tool search", "got 5 results" — the state diffs between steps. LangGraph: stream_mode="updates" (per-node diffs) [lg/stream]. LangChain LCEL: astream_events (a typed event stream: on_chat_model_stream, on_tool_start, on_tool_end) [lc/astream-events].

3. Progress audience → custom events. Work happening inside one tool/node (a loop, a long embed, a download) is invisible to the framework's automatic events. You must emit progress yourself: LangGraph stream_mode="custom" via get_stream_writer() [lg/stream]; LCEL via custom callback / dispatched events [lc/astream-events].

The load-bearing insight from the LangGraph docs: stream modes are composable — pass a list (stream_mode=["messages","updates","custom"]) and demultiplex on the client by the tuple tag [lg/stream]. So the real design question is never "can I stream" but "which projection(s) does this surface need, and how do I tag them on one wire?"

Second axiom: a stream is a contract with a client that can vanish. Networks drop, users close tabs, browsers cap connections. The backend must decide, up front, whether a disconnect should cancel the run (stop burning tokens) or detach and let it finish (so a reconnect can replay). That decision is part of the design, not an afterthought — see 困境 Case 2.

SOP 工作流 (Agentic Protocol)

Walk top-down. Each step has a gate.

Step 1 · Confirm streaming is warranted

Gate: is a human waiting on a run that takes >~1–2s? If no (batch job, sub- second call), don't stream — return the whole payload. Streaming a fast call adds SSE/WebSocket framing, reconnect logic, and partial-parse bugs for no UX win [mdn/sse]. Exit here for fast paths.

Step 2 · Classify the surface → pick the projection

Map the surface to one (or more) of the three audiences:

values emits the full state after each step (heavy, good for resume); updates emits only the diff (light, good for live UI) [lg/stream]. Default a chat agent to ["messages","updates"] and add "custom" only when a tool has internal progress worth surfacing [lg/stream] (= [[agentsop-langgraph]] OP-8).

Step 3 · Pick the transport

Gate questions: does the client only receive (server→client), or also need to send mid-stream (interrupt, steer)?

• Receive-only → SSE. Simplest correct default: one long-lived HTTP response, text/event-stream, auto-reconnect + Last-Event-ID built into the browser EventSource [mdn/sse]. This is what most "stream the agent" use cases need.
• Bidirectional → WebSocket. Only when the client must push during the stream (live cancel, mid-run user input, collaborative). Costs you reconnect logic you get free with SSE.
• Server-internal / non-browser → async generator / gRPC stream. If both ends are yours, skip HTTP framing and yield the tuples directly.

Step 4 · Mix token + step streams on one wire

Use the multi-mode form so one connection carries everything; tag each chunk so the client routes it:

async for mode, chunk in graph.astream(
        inp, stream_mode=["messages", "updates", "custom"], config=cfg):
    if mode == "messages":
        token, meta = chunk
        yield sse("token", token.content)            # → append to bubble
    elif mode == "updates":
        yield sse("step", chunk)                      # → "running tool X"
    elif mode == "custom":
        yield sse("progress", chunk)                  # → progress bar

[lg/stream]. SSE event: field is exactly the demux key; the browser's EventSource.addEventListener("token"|"step"|"progress", …) splits it client-side [mdn/sse]. Never interleave two semantic streams on one untagged channel — the client can't tell a token from a tool name.

Step 5 · Decide disconnect policy before shipping

For each surface answer: on client disconnect, cancel or detach?

• Cancel (stop the run) when: every step costs money/tokens, output is useless without the client, no resume planned. Wire it to the request's cancellation signal so the generator is closed and the LLM call aborted [oai/stream].
• Detach + persist when: the run has side effects that must complete, OR the user may reconnect and wants the result. Pair with a checkpointer ([[agentsop-langgraph]] Step 6) and a resumable event log so reconnect replays via Last-Event-ID [mdn/sse]. Default for a chat agent: cancel (cheap, stateless). Default for a long side-effecting pipeline: detach + persist.

Step 6 · Add backpressure + heartbeat before production

• Heartbeat: SSE connections die silently behind proxies; emit a comment ping (: keep-alive\n\n) every ~15s during long quiet stretches [mdn/sse].
• Backpressure: if the client reads slower than the model emits, your buffer grows. Bound the queue; on overflow either drop intermediate updates (keep messages) or apply flow control. Tokens are the audience-critical stream; progress events are droppable.
• Flush: disable response buffering (X-Accel-Buffering: no for nginx) or the proxy batches your tokens and kills the "streaming" feel.

操作模型 (Operation Models)

Format: Trigger → Action → Output → Evidence.

OP-1 · Stream final tokens to a chat client (the 80% case)

• Trigger: User-facing chat; want typing effect on the final answer.
• Action: LangGraph graph.astream(inp, stream_mode="messages") → yield each (token, metadata)'s token.content; filter by metadata so you only stream the final node's LLM, not sub-agent chatter. Raw: OpenAI stream=True iterate chunk.choices[0].delta.content; Anthropic with client.messages.stream(...) as s: for t in s.text_stream.
• Output: Character-by-character final answer; no tool JSON leaks.
• Evidence: [lg/stream] messages mode; [oai/stream]; [anthropic/stream].

OP-2 · Stream intermediate agent steps

• Trigger: Dev/inspector UI; show "which node / which tool, with inputs".
• Action: LangGraph stream_mode="updates" → each chunk is {node_name: state_diff}; render as a step log. LCEL: astream_events(version="v2") and switch on event["event"] (on_tool_start/on_tool_end/on_chain_*).
• Output: Live step trace without the full state weight of values.
• Evidence: [lg/stream] updates mode; [lc/astream-events].

OP-3 · Emit custom in-tool progress

• Trigger: A tool/node does long internal work (embed 200 docs, paginate an API) the framework can't see.
• Action: LangGraph — inside the node, w = get_stream_writer(); w({"progress": i/n}); consume on stream_mode="custom". LCEL — dispatch a custom event / callback that astream_events surfaces.
• Output: A real progress signal instead of a frozen spinner.
• Evidence: [lg/stream] custom mode + stream writer.

OP-4 · Multiplex modes on one SSE connection

• Trigger: One surface needs tokens and steps and progress.
• Action: stream_mode=["messages","updates","custom"]; map each (mode, chunk) tuple to a distinct SSE event: name; client addEventListener per name (Step 4 snippet).
• Output: Single connection, cleanly demuxed; no extra round-trips.
• Evidence: [lg/stream] (list form returns (mode, chunk) tuples); [mdn/sse] (named events).

OP-5 · Choose SSE vs WebSocket

• Trigger: Deciding the transport.
• Action: Receive-only (browser just displays) → SSE (free reconnect + Last-Event-ID). Client must push mid-stream (cancel, steer, collaborate) → WebSocket. Both ends yours / non-HTTP → async generator.
• Output: Right transport; no hand-rolled reconnect for the common case.
• Evidence: [mdn/sse] (EventSource auto-reconnect, server-push only).

OP-6 · Handle client disconnect (cancel vs detach)

• Trigger: Stream may outlive the client's interest.
• Action: Hook the request cancellation token. Cancel: close the async generator → upstream LLM/agent call aborts; release resources [oai/stream]. Detach: keep running under a checkpointer, log events with monotonic IDs so a reconnect replays from Last-Event-ID.
• Output: No zombie runs burning tokens; or a resumable run, by design.
• Evidence: [oai/stream] cancellation; [mdn/sse] Last-Event-ID; [[agentsop-langgraph]] Step 6 (checkpointer for durability).

OP-7 · Keep the connection alive (heartbeat + flush)

• Trigger: Long quiet gaps (a slow tool) cause proxies to drop the stream, or tokens arrive in clumps not smoothly.
• Action: Emit : ping\n\n comments every ~15s; set X-Accel-Buffering: no / disable proxy buffering; flush after each event.
• Output: Connection survives idle periods; tokens render smoothly.
• Evidence: [mdn/sse] (comment lines ignored by client, keep socket warm).

OP-8 · Filter the firehose to the final answer only

• Trigger: A multi-agent graph streams every LLM's tokens; the chat bubble fills with sub-agent noise.
• Action: On messages mode, inspect metadata (langgraph_node, tags) and forward only tokens whose node is the user-facing responder; route the rest to updates (dev view) or drop.
• Output: Clean final answer; sub-agent reasoning stays in the inspector.
• Evidence: [lg/stream] (messages chunks carry node metadata for filtering).

困境决策案例 (Dilemma Cases)