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:

Surface	Primary projection	LangGraph mode	LangChain
Chat / prose	final tokens	messages	astream_events → on_chat_model_stream
Agent inspector / dev UI	step updates	updates	astream_events (on_tool_*, on_chain_*)
Full-state replay / resume	snapshots	values	n/a (rebuild from events)
Long in-tool work	custom progress	custom	dispatched custom events
Debug everything	raw firehose	debug	astream_events (all)

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)

Case 1 · "Stream the tokens, or stream the steps?" — an agent that thinks then answers

• 困境: A research agent runs 4 tools over ~40s, then writes a 2-paragraph answer. If you stream messages only, the user stares at a frozen spinner for 40s, then sees text. If you stream updates only, they see "calling tool X" but the final answer dumps all at once, losing the typing feel.
• 约束: One SSE connection (mobile client). The 40s of tool work is the scary part for the user; the final prose is the payoff.
• 决策步骤:
  1. Reject "pick one mode" — the surface has two audiences in one timeline (progress during work, prose at the end) [lg/stream].
  2. Use stream_mode=["updates","messages"]. During tool work, updates chunks drive a live step list ("Searching… Reading 5 docs… Synthesizing"). When the final responder node starts emitting, messages tokens stream into the bubble (OP-4 demux).
  3. Filter messages to the final node only (OP-8) so the tool-call LLMs don't leak into the answer.
  4. If a tool itself is slow (>5s), add custom progress from inside it (OP-3) so the step list isn't itself frozen.
• 结果: Continuous feedback for the whole 40s, then a smooth typed answer — on one connection, no extra round-trips.
• 可提取的操作: OP-4 + OP-8. The answer to "tokens or steps" is almost always "both, tagged, on one wire" — the question is which is primary when.

Case 2 · "Client disconnects mid-stream — cancel the run or let it finish?"

• 困境: A user kicks off a 90s agent that books a flight (real side effect), then closes the tab at second 30. The stream's consumer is gone. Do you kill the run (and maybe leave a half-booking) or let it complete (burning tokens for a client that may never return)?
• 约束: The booking step is irreversible; tokens cost money; the user might reopen the tab.
• 决策步骤:
  1. Recognize this is the cancel-vs-detach decision (Step 5), and it differs by where in the run the disconnect happens.
  2. Because there's an irreversible side effect, do not hard-cancel mid-action — that's the half-booking risk. Detach: let the current durable step finish under a checkpointer ([[agentsop-langgraph]] Step 6 / HITL ordering — side effects in their own committed step).
  3. Persist the event log with monotonic IDs. On reconnect, replay from Last-Event-ID so the user sees the outcome [mdn/sse].
  4. If, instead, this were a read-only chat with no side effects, do the opposite: cancel immediately on disconnect to stop burning tokens [oai/stream] — that's the cheaper, correct default for chat.
• 结果: Side-effecting runs detach + persist + replay; stateless chat runs cancel. The policy is chosen by reversibility and cost, not by reflex.
• 可提取的操作: OP-6. Disconnect policy is a function of side-effect reversibility and per-step cost — decide it per surface, before shipping, never let it default to "whatever the framework does on socket close".

---

反模式与边界 (Anti-patterns & Boundaries)

• Don't stream everything. Forwarding the raw debug/values firehose to a chat UI floods the client with full-state snapshots and sub-agent tokens. Project to the audience (Step 2); values is heavy by design [lg/stream].
• Don't stream a sub-second call. SSE/WebSocket framing + reconnect + partial- parse bugs for zero perceived-latency gain. Return the whole payload [mdn/sse].
• Don't skip disconnect handling. A stream with no cancel/detach policy leaks zombie runs that burn tokens after the client is gone, or half-completes side effects. Decide in Step 5 [oai/stream].
• Don't interleave semantic streams on one untagged channel. Tokens and tool names on the same unnamed wire are unparseable client-side. Tag with SSE event: / the (mode, chunk) tuple (OP-4) [lg/stream] [mdn/sse].
• Don't leak sub-agent tokens into the final answer. Filter messages by node metadata (OP-8) [lg/stream].
• Don't forget the heartbeat. Long quiet gaps behind a proxy silently kill the connection; the user sees a hang, not an error. Ping every ~15s [mdn/sse].
• Don't assume buffering is off. A buffering proxy batches your tokens and destroys the streaming feel; disable it explicitly (OP-7).
• Don't reach for WebSocket by default. If the client only receives, SSE is simpler and gives reconnect for free [mdn/sse]. Reserve WS for true bidirectional needs.

Hard boundaries (streaming is the wrong tool when):

• Output must be validated/transformed as a whole before the user sees any of it (structured JSON you parse server-side, content that needs a safety pass) — stream nothing until validated, or stream into a parser, never raw to the user.
• The consumer is a machine that wants one atomic JSON object — give it the whole response; partial JSON tokens are a parsing hazard, not a feature.
• No human is waiting (offline batch) — streaming adds cost for no audience.

---

跨框架对照 (Cross-framework Context)

Concern	LangGraph	LangChain (LCEL)	OpenAI SDK	Anthropic SDK
Final tokens	stream_mode="messages" → (token, meta) [lg/stream]	astream_events → on_chat_model_stream [lc/astream-events]	stream=True, iterate delta.content [oai/stream]	client.messages.stream(...), text_stream [anthropic/stream]
Intermediate steps	stream_mode="updates" (per-node diff) [lg/stream]	astream_events (on_tool_*, on_chain_*) [lc/astream-events]	manual: detect tool_calls deltas [oai/stream]	manual: handle content_block_* / tool-use events [anthropic/stream]
Full state	stream_mode="values" (snapshot) [lg/stream]	rebuild from events	n/a	n/a
Custom progress	stream_mode="custom" + get_stream_writer() [lg/stream]	dispatch custom event / callback [lc/astream-events]	hand-rolled out-of-band	hand-rolled out-of-band
Mix modes	list form → tagged (mode, chunk) tuples [lg/stream]	one typed event stream, switch on event [lc/astream-events]	one delta stream, branch on field	one event stream, branch on type
Granularity	node-level + token-level + custom	event-level (richest typed taxonomy)	token + tool-call deltas	event + token (typed blocks)

Heuristics:

• LangGraph — best when you already have a graph and want token+step+custom on one demuxable wire; [lg/stream] modes are the cleanest projection model. See [[agentsop-langgraph]] OP-8 for the orchestration side.
• LangChain LCEL — astream_events gives the richest typed event taxonomy (every on_* lifecycle hook); reach for it when you need fine-grained event routing without a full graph [lc/astream-events].
• Raw OpenAI / Anthropic — you get a single token/delta stream and must derive "steps" yourself from tool-call deltas / content-block events. Choose when you have no orchestration layer and want zero framework weight [oai/stream] [anthropic/stream].

Transport is orthogonal to all four: SSE (default, receive-only), WebSocket (bidirectional), or async generator (internal) wraps any of them. Pick the SDK for what to stream, the transport for how the client consumes it [mdn/sse].

---

附录: 引用速查 (Citation Index)

Short tags → full sources in references/R1-source-evidence.md:

• [lg/stream] = LangGraph streaming concept (values / updates / messages / custom / debug; multi-mode list; get_stream_writer) — distilled in [[agentsop-langgraph]] OP-8 + references/R1.
• [lc/astream-events] = LangChain LCEL astream_events typed event stream.
• [oai/stream] = OpenAI streaming (stream=True, deltas, cancellation).
• [anthropic/stream] = Anthropic Messages streaming (client.messages.stream, text_stream, content-block events).
• [mdn/sse] = MDN Server-Sent Events (EventSource, named events, auto-reconnect, Last-Event-ID, comment heartbeats).