mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-28 20:44:50 +02:00
[eric] agents: extend TTFT probes to full waterfall (msg_received/thinking/first_text/result)
This commit is contained in:
@@ -17,6 +17,7 @@ from backend.apps.agents.manager.session.session_store import load_session_data
|
||||
from backend.apps.agents.manager.session.apply_context_window import apply_context_window
|
||||
from backend.apps.agents.manager.prompt.tool_catalog import get_all_tool_names
|
||||
from backend.apps.agents.manager.prompt.prompt_context import resolve_mode
|
||||
from backend.apps.agents.manager.ttft_probe import ttft_probe
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -44,6 +45,7 @@ class Messaging(AgentManagerProtocol):
|
||||
client_message_id: Optional[str] = None,
|
||||
):
|
||||
"""Send a follow-up message to an existing session."""
|
||||
ttft_probe(session_id, "msg_received")
|
||||
session = self.sessions.get(session_id)
|
||||
if not session:
|
||||
data = load_session_data(session_id)
|
||||
|
||||
@@ -113,6 +113,7 @@ class TurnRunner(AgentManagerProtocol):
|
||||
message, session, session_id, turn, thinking, self.live_partial, self.sessions
|
||||
)
|
||||
elif isinstance(message, ResultMessage):
|
||||
ttft_probe(session_id, "result", chars=turn.assistant_text_chars)
|
||||
await handle_result_message(
|
||||
message, session, session_id, turn, thinking, self.sessions,
|
||||
resolved_model, api_type, global_settings,
|
||||
|
||||
@@ -14,6 +14,7 @@ from backend.apps.agents.core.models import AgentSession
|
||||
from backend.apps.agents.core.ws_manager import ws_manager
|
||||
from backend.apps.agents.manager.streaming.state import ThinkingState, TurnState
|
||||
from backend.apps.agents.manager.streaming.PartialReply import PartialReply
|
||||
from backend.apps.agents.manager.ttft_probe import ttft_probe
|
||||
|
||||
try:
|
||||
from claude_agent_sdk.types import StreamEvent
|
||||
@@ -44,6 +45,7 @@ async def handle_stream_event(
|
||||
|
||||
if block_type == "text":
|
||||
if turn.stream_text_msg_id is None:
|
||||
ttft_probe(session_id, "first_text_block")
|
||||
turn.stream_text_msg_id = uuid4().hex
|
||||
await ws_manager.send_to_session(session_id, "agent:stream_start", {
|
||||
"session_id": session_id,
|
||||
@@ -54,6 +56,8 @@ async def handle_stream_event(
|
||||
|
||||
elif block_type == "thinking":
|
||||
# Reasoning trace from thinking-capable models (GPT-5.3 Codex, Gemini 3 Pro/Flash, Claude with extended thinking). Rendered as a collapsible "thinking" message in the UI via the existing stream infrastructure, the frontend already handles role="thinking" for the DynamicIsland/agent card rendering.
|
||||
if not thinking.block_starts:
|
||||
ttft_probe(session_id, "first_thinking_block")
|
||||
thinking_msg_id = uuid4().hex
|
||||
turn.stream_block_index_map[index] = thinking_msg_id
|
||||
# Server-stamp start so we can accumulate per-turn elapsed_ms across multiple thinking blocks (think → tool → think → answer turns sum correctly).
|
||||
@@ -85,6 +89,8 @@ async def handle_stream_event(
|
||||
|
||||
if msg_id and delta_type == "text_delta":
|
||||
text_chunk = delta.get("text", "")
|
||||
if turn.assistant_text_chars == 0:
|
||||
ttft_probe(session_id, "first_text_delta", chars=len(text_chunk))
|
||||
turn.assistant_text_chars += len(text_chunk)
|
||||
turn.stream_text_accum += text_chunk
|
||||
live_partial[session_id] = PartialReply(
|
||||
|
||||
Reference in New Issue
Block a user