菜单
Streams LLM responses chunk-by-chunk from an agent via runner.run_stream(). Use when building real-time UIs, SSE endpoints, WebSocket streaming, or any feature that needs to display tokens as they arrive rather than waiting for the full response.
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
基于 SOC 职业分类
| name | streaming-response-handler |
| description | Streams LLM responses chunk-by-chunk from an agent via runner.run_stream(). Use when building real-time UIs, SSE endpoints, WebSocket streaming, or any feature that needs to display tokens as they arrive rather than waiting for the full response. |
Use
codemap find "SymbolName"to locate any symbol before reading — it gives exact file + line range and is faster than grep across the whole repo.
# from __future__ import annotations is safe here (no @tool definitions)
from __future__ import annotations
from lauren_ai._agents._runner import AgentRunnerBase as AgentRunner
from lauren_ai._config import LLMConfig
from lauren_ai._transport._mock import MockTransport
cfg = LLMConfig(provider="anthropic", model="claude-opus-4-6", api_key="sk-ant-...")
runner = AgentRunner(transport=transport)
async for chunk in await runner.run_stream(my_agent_instance, "Tell me a joke"):
if chunk.delta:
print(chunk.delta, end="", flush=True)
if chunk.stop_reason:
print() # newline at end
| Field | Type | Description |
|---|---|---|
delta | str | Text fragment from this chunk (may be empty string) |
stop_reason | str | None | Set on the final chunk: "end_turn", "tool_use", etc. |
usage | TokenUsage | None | Set on the final chunk with token counts |
thinking_delta | str | None | Anthropic extended thinking fragment |
tool_call_delta | ToolCallDelta | None | Streaming tool-call fragment |
guardrail_override | str | None | Set if a guardrail replaced the output |
async def stream_and_collect(runner, agent_instance, prompt):
full_text = ""
async for chunk in await runner.run_stream(agent_instance, prompt):
if chunk.delta:
full_text += chunk.delta
return full_text
from lauren import sse, EventStream
from lauren_ai._agents._runner import AgentRunnerBase as AgentRunner
@get("/chat/stream")
async def chat_stream(
prompt: Query[str],
agent: MyAgent,
runner: AgentRunner,
) -> EventStream:
async def _generate():
async for chunk in await runner.run_stream(agent, prompt):
if chunk.delta:
yield chunk.delta
return EventStream(_generate())
from lauren_ai._transport import CompletionChunk, TokenUsage
from lauren_ai._transport._mock import MockTransport
mock = MockTransport()
mock.queue_stream([
CompletionChunk(delta="Hello"),
CompletionChunk(delta=" world"),
CompletionChunk(
stop_reason="end_turn",
usage=TokenUsage(input_tokens=5, output_tokens=2),
),
])
chunks = []
async for chunk in await runner.run_stream(my_agent, "hi"):
chunks.append(chunk)
| File | Contents |
|---|---|
src/lauren_ai/_agents/_runner.py | run_stream() implementation |
src/lauren_ai/_transport/__init__.py | CompletionChunk, ToolCallDelta |
src/lauren_ai/_transport/_mock.py | queue_stream() helper |
Adds content safety filters to lauren-ai agents using @use_guardrails for input and output checks. Use when an agent needs prompt injection protection, PII redaction, topic filtering, length limiting, or LLM-powered content evaluation. Also covers creating custom guardrail classes injectable via @guardrail.
Aggregates per-model telemetry from SignalBus events using CostTracker and a custom PerformanceMetrics collector. Use when tracking total LLM calls, token consumption, average latency, error rate, and cost per model across agent runs.
Submit agent tasks to an asyncio queue and process them with a background worker, returning results by task ID. Use when you need non-blocking agent execution in web handlers, batch processing pipelines, or any situation where callers should not wait for agent completion.
Builds AI agents with the lauren-ai framework using @agent, @use_tools, and AgentRunner. Use when defining an agent class, setting up lifecycle hooks, streaming responses, implementing agent delegation, or configuring the agentic loop with tools, guardrails, and memory.
Creates tools for lauren-ai agents using @tool() on async functions or classes. Use when implementing a @tool() function or class, working with ToolContext for DI injection, adding caching, HITL approval, or using built-in skills like WebSearchTool, HttpFetchTool, or CodeExecutionTool.
Implements an in-memory calendar management tool for agents using @tool() class-form. Use when building agents that create, query, or cancel calendar events, with ISO 8601 time handling and date-based filtering. Supports @set_metadata for tool-type annotation.