feat(streaming): add Executor.execute_stream() returning AsyncIterator[str] for token-by-token delivery #45

Merged

hurui200320 merged 1 commits from feature/streaming-execute-stream into master 2026-06-12 12:29:49 +00:00

Conversation 2 Commits 1 Files Changed 8

+9314 -12

1 Commits

Author	SHA1	Message	Date
hurui200320	d93c32ccee	feat(streaming): add Executor.execute_stream() returning AsyncIterator[str] for token-by-token delivery ... Implements the full streaming execution path for the CleverThis router: - **LLMAgent.stream_message(message, context)** (agents/llm.py): Async generator calling self.chat_model.astream(messages) and yielding str(chunk.content) per chunk. Captures token counts from the final chunk's usage_metadata using the same _safe_int() fallback chain as process_message(). Sets _last_token_usage and last_token_usage_var after exhaustion. - **Node.stream_agent(state)** (langgraph/nodes.py): Async generator that mirrors _execute_agent() but uses agent.stream_message() for LLMAgent instances and falls back to process_message() for non-LLM agents. Captures per-node token usage into self._last_stream_usage after stream exhaustion. - **PureLangGraph.execute_stream()** (langgraph/pure_graph.py): Async generator mirroring execute(). Calls _stream_from_node() which uses astream() for all AGENT nodes (buffering tokens for intermediate nodes, yielding only from the terminal node) and ainvoke() for non-AGENT nodes. Stores final state and node usages in _last_stream_state / _last_stream_node_usages for post-stream access. Same limit enforcement (timeout_ms, max_depth, max_model_calls, max_tool_calls) as _execute_from_node(). - **_execute_llm_stream() / _execute_graph_stream()** (runtime_dispatch.py): Async generator dispatch functions mirroring _execute_llm() / _execute_graph(). Set executor.last_result after stream exhaustion using collected token usage. - **Executor.execute_stream(message)** (runtime.py): Public async generator that dispatches to _execute_llm_stream() for llm actors and _execute_graph_stream() for graph actors. Raises ConfigurationError for tool/multi_actor types. Resets last_result to None at entry; callers must exhaust the iterator. - **Executor.last_result** (runtime.py): ActorResult | None attribute set by execute_stream() after the iterator is exhausted. - Intermediate AGENT nodes use astream() internally (tokens buffered, not yielded to caller); terminal AGENT nodes yield buffered tokens. This avoids double- running the terminal LLM while satisfying AC2. - Non-AGENT (FUNCTION/TOOL) nodes always use ainvoke() via node.execute(). - timeout_ms in execute_stream() buffers all tokens via asyncio.wait_for() around _collect_stream_tokens(), then yields them — this is simpler than wrapping an async generator directly. - Tool and multi_actor actor types raise ConfigurationError (no streaming path). - 60 new BDD scenarios in features/execute_stream.feature covering all acceptance criteria: token delivery, last_result population, token count extraction, limit enforcement (timeout, model_calls, tool_calls), error paths, graph streaming, non-AGENT node paths, parallel execution paths, and edge cases. - All 2395 existing unit tests continue to pass. - Coverage: 96.5% (meets threshold). ISSUES CLOSED: #16	2026-06-12 10:08:12 +00:00