Stream runtime status updates
This commit is contained in:
parent
10131cca48
commit
234a3c957d
|
|
@ -251,6 +251,12 @@ def create_app() -> FastAPI:
|
|||
def sse(event: str, payload: dict[str, Any]) -> str:
|
||||
return f"event: {event}\ndata: {json.dumps(payload, ensure_ascii=False)}\n\n"
|
||||
|
||||
def runtime_status(task_id: str, stage: str, message: str) -> str:
|
||||
return sse(
|
||||
"runtime_status",
|
||||
{"task_id": task_id, "stage": stage, "message": message},
|
||||
)
|
||||
|
||||
async def emit_tool_events(task_id: str, after_sequence: int):
|
||||
events = await event_store.list_events(task_id)
|
||||
visible_types = {
|
||||
|
|
@ -295,9 +301,20 @@ def create_app() -> FastAPI:
|
|||
messages = await runtime.context_builder.build_async_messages(
|
||||
task, history, memory_records
|
||||
)
|
||||
yield runtime_status(
|
||||
task.task_id,
|
||||
"planning",
|
||||
"Планирую, нужны ли локальные действия...",
|
||||
)
|
||||
tool_observations = await runtime._run_action_loop(
|
||||
task.task_id, messages, conversation.workspace
|
||||
)
|
||||
if tool_observations:
|
||||
yield runtime_status(
|
||||
task.task_id,
|
||||
"running_tools",
|
||||
"Локальные действия выполнены, готовлю ответ...",
|
||||
)
|
||||
async for tool_event in emit_tool_events(task.task_id, task_event.sequence):
|
||||
yield tool_event
|
||||
if any(observation.get("requires_approval") for observation in tool_observations):
|
||||
|
|
@ -341,6 +358,11 @@ def create_app() -> FastAPI:
|
|||
+ json.dumps(tool_observations, ensure_ascii=False, indent=2),
|
||||
},
|
||||
]
|
||||
yield runtime_status(
|
||||
task.task_id,
|
||||
"answering",
|
||||
"Формирую ответ...",
|
||||
)
|
||||
await event_store.append(task.task_id, "model_call_started", {"role": "thinker"})
|
||||
async for chunk in model_client.stream_chat("thinker", messages):
|
||||
delta = str(chunk.get("delta") or "")
|
||||
|
|
@ -508,6 +530,11 @@ def create_app() -> FastAPI:
|
|||
)
|
||||
messages = await runtime.context_builder.build_async_messages(task)
|
||||
tool_observations = [tool_observation]
|
||||
yield runtime_status(
|
||||
task_id,
|
||||
"running_tool",
|
||||
"Выполняю разрешённое действие...",
|
||||
)
|
||||
async for tool_event in emit_tool_events(task_id, continued_event.sequence):
|
||||
yield tool_event
|
||||
if has_password_request(tool_observations):
|
||||
|
|
@ -569,6 +596,7 @@ def create_app() -> FastAPI:
|
|||
+ json.dumps(tool_observations, ensure_ascii=False, indent=2),
|
||||
},
|
||||
]
|
||||
yield runtime_status(task_id, "answering", "Формирую ответ...")
|
||||
await event_store.append(task_id, "model_call_started", {"role": "thinker"})
|
||||
async for chunk in model_client.stream_chat("thinker", messages):
|
||||
delta = str(chunk.get("delta") or "")
|
||||
|
|
@ -689,6 +717,11 @@ def create_app() -> FastAPI:
|
|||
)
|
||||
messages = await runtime.context_builder.build_async_messages(task)
|
||||
tool_observations = [tool_observation]
|
||||
yield runtime_status(
|
||||
task_id,
|
||||
"running_tool",
|
||||
"Выполняю действие с переданным паролем...",
|
||||
)
|
||||
async for tool_event in emit_tool_events(task_id, continued_event.sequence):
|
||||
yield tool_event
|
||||
if has_password_request(tool_observations):
|
||||
|
|
@ -724,6 +757,7 @@ def create_app() -> FastAPI:
|
|||
+ json.dumps(tool_observations, ensure_ascii=False, indent=2),
|
||||
},
|
||||
]
|
||||
yield runtime_status(task_id, "answering", "Формирую ответ...")
|
||||
await event_store.append(task_id, "model_call_started", {"role": "thinker"})
|
||||
async for chunk in model_client.stream_chat("thinker", messages):
|
||||
delta = str(chunk.get("delta") or "")
|
||||
|
|
|
|||
|
|
@ -562,6 +562,13 @@ async function handleAssistantStreamEvent(pending, name, data, context) {
|
|||
setStatus("#task-status", data.task_id, "warn");
|
||||
return;
|
||||
}
|
||||
if (name === "runtime_status") {
|
||||
pending.querySelector(".message-meta span").textContent = data.stage || "running";
|
||||
if (!context.contentStarted) {
|
||||
setMessagePending(pending, data.message || "Работаю...");
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (name === "reasoning_delta") {
|
||||
pending.querySelector(".message-meta span").textContent = "reasoning";
|
||||
appendInlineReasoning(pending, data.delta || "");
|
||||
|
|
|
|||
|
|
@ -100,6 +100,9 @@ def test_stream_chat_endpoint_executes_tool_before_streaming_answer(tmp_path, mo
|
|||
body = "".join(response.iter_text())
|
||||
|
||||
assert response.status_code == 200
|
||||
assert 'event: runtime_status\ndata: {"task_id":' in body
|
||||
assert '"stage": "planning"' in body
|
||||
assert body.index('"stage": "planning"') < body.index("event: tool_call_started")
|
||||
assert "event: tool_call_started" in body
|
||||
assert "event: tool_call_finished" in body
|
||||
assert "stream tool content" in body
|
||||
|
|
@ -172,6 +175,8 @@ def test_continue_stream_executes_approved_tool_and_streams_answer(tmp_path, mon
|
|||
|
||||
assert "event: tool_approval_requested" in initial_body
|
||||
assert response.status_code == 200
|
||||
assert '"stage": "running_tool"' in body
|
||||
assert body.index('"stage": "running_tool"') < body.index("event: tool_call_finished")
|
||||
assert "event: tool_call_finished" in body
|
||||
assert "event: content_delta" in body
|
||||
assert "continued after approval" in body
|
||||
|
|
|
|||
Loading…
Reference in New Issue