feat(assistant): #91 steps humains en direct + outils web_search/fetch_url (labels backend, SSE step, garde SSRF)
This commit is contained in:
@@ -317,6 +317,21 @@ def _resolve_provider_name(requested: str | None) -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
def _tool_event_sse(rec) -> str:
|
||||
"""Serialize one executed tool call as an SSE ``tool`` event."""
|
||||
payload = json.dumps(
|
||||
{"name": rec.name, "ok": rec.ok, "arguments": rec.arguments, "step": rec.step},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
return f"event: tool\ndata: {payload}\n\n"
|
||||
|
||||
|
||||
def _thought_event_sse(note: dict) -> str:
|
||||
"""Serialize one intermediate reasoning note as an SSE ``step`` event."""
|
||||
payload = json.dumps({"step": note}, ensure_ascii=False)
|
||||
return f"event: step\ndata: {payload}\n\n"
|
||||
|
||||
|
||||
# ── Endpoints ──
|
||||
|
||||
|
||||
@@ -456,6 +471,8 @@ async def api_bookslm_agent(
|
||||
)
|
||||
|
||||
async def generate_sse():
|
||||
import asyncio
|
||||
|
||||
try:
|
||||
cfg_name = _resolve_provider_name(req.provider)
|
||||
if cfg_name is None:
|
||||
@@ -466,20 +483,43 @@ async def api_bookslm_agent(
|
||||
yield f"event: error\ndata: {error_data}\n\n"
|
||||
return
|
||||
|
||||
result = await run_agent(
|
||||
# Stream tool events live: each executed step is pushed on the
|
||||
# queue by the loop callback and emitted as soon as it happens,
|
||||
# so the UI can grow its « N steps » block while thinking.
|
||||
queue: asyncio.Queue = asyncio.Queue()
|
||||
|
||||
def _on_tool(rec) -> None:
|
||||
queue.put_nowait(("tool", rec))
|
||||
|
||||
run_task = asyncio.create_task(run_agent(
|
||||
messages,
|
||||
ctx=ctx,
|
||||
llm=_llm,
|
||||
resume_messages=req.confirm_messages,
|
||||
confirm_pending=req.confirm,
|
||||
)
|
||||
on_tool_call=_on_tool,
|
||||
on_thought=lambda note: queue.put_nowait(("thought", note)),
|
||||
))
|
||||
|
||||
for rec in result.tool_calls:
|
||||
payload = json.dumps(
|
||||
{"name": rec.name, "ok": rec.ok, "arguments": rec.arguments},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
yield f"event: tool\ndata: {payload}\n\n"
|
||||
# Drain every completed step as soon as it lands, while the agent
|
||||
# keeps running in the background. If the client disconnects, the
|
||||
# generator is cancelled: release the run so it cannot orphan.
|
||||
try:
|
||||
while True:
|
||||
try:
|
||||
kind, item = await asyncio.wait_for(queue.get(), timeout=0.25)
|
||||
except asyncio.TimeoutError:
|
||||
if run_task.done():
|
||||
break
|
||||
continue
|
||||
yield _tool_event_sse(item) if kind == "tool" else _thought_event_sse(item)
|
||||
while not queue.empty():
|
||||
kind, item = queue.get_nowait()
|
||||
yield _tool_event_sse(item) if kind == "tool" else _thought_event_sse(item)
|
||||
result = run_task.result()
|
||||
finally:
|
||||
if not run_task.done():
|
||||
run_task.cancel()
|
||||
|
||||
if result.stopped == "confirmation_required":
|
||||
pending = json.dumps(
|
||||
|
||||
Reference in New Issue
Block a user