233 lines
10 KiB
Python
233 lines
10 KiB
Python
"""Task-intent agent endpoints (Step 1 of the roundtable flow).
|
|
|
|
Lightweight wrapper around the lead-agent runtime that runs a single built-in
|
|
agent (``roundtable-intent``) on its own thread. The agent's job is to do 2-5
|
|
clarifying turns with the user and then emit a structured
|
|
``{objective, constraints, assumptions}`` summary the frontend hands off to
|
|
the multi-agent roundtable as its initial intent.
|
|
|
|
Same SSE shape as ``multi_agent.py`` (transparent passthrough + appended
|
|
status frame), so the frontend can reuse the same reader pattern.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import Any, AsyncIterator
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, Body, HTTPException, Request
|
|
from fastapi.responses import StreamingResponse
|
|
|
|
# Reuse the building blocks already in multi_agent.py — keeps both routers
|
|
# identical in connection handling, payload shape, and error mapping.
|
|
from app.gateway.routers.multi_agent import (
|
|
_auth_headers,
|
|
_create_thread,
|
|
_find_dispatch_message,
|
|
_find_trailing_ai_text,
|
|
_flatten_content,
|
|
_loopback_base,
|
|
_parse_last_messages,
|
|
_run_payload,
|
|
_stream_upstream,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/intent", tags=["intent"])
|
|
|
|
INTENT_AGENT_ID = "roundtable-intent"
|
|
|
|
# Matches the SOUL-mandated handoff signal: a [INTENT_READY] line followed by
|
|
# a ```json ... ``` block. The model is told to emit this only when it judges
|
|
# the user's task is fully specified.
|
|
_INTENT_READY_PATTERN = re.compile(
|
|
r"\[INTENT_READY\]\s*```json\s*(\{.*?\})\s*```",
|
|
re.DOTALL,
|
|
)
|
|
|
|
|
|
@router.post("/init")
|
|
async def init_intent(request: Request, payload: dict = Body(...)) -> dict[str, Any]:
|
|
"""Create a fresh thread for an intent-clarification session.
|
|
|
|
Returns the stable built-in ``agent_id`` so the frontend doesn't need to
|
|
know it. Unlike multi-agent ``/init``, no coordinator is built — the
|
|
intent agent is long-lived and already registered via the legacy
|
|
agent-sync path on disk.
|
|
"""
|
|
base = _loopback_base(request)
|
|
headers = _auth_headers(request)
|
|
# trust_env=False: this client only talks to the same-process gateway at
|
|
# 127.0.0.1, so honoring HTTP_PROXY / HTTPS_PROXY env vars (set system-wide
|
|
# on the dev machine) makes httpx route loopback calls through an upstream
|
|
# proxy that can't reach 127.0.0.1, costing 40+ seconds of timeout per
|
|
# request before falling back. Symptom in logs: `headers_in=42.160s`.
|
|
async with httpx.AsyncClient(timeout=None, trust_env=False) as client:
|
|
thread_id = await _create_thread(client, base, headers)
|
|
return {
|
|
"status": "success",
|
|
"agent_id": INTENT_AGENT_ID,
|
|
"thread_id": thread_id,
|
|
}
|
|
|
|
|
|
@router.post("/stream")
|
|
async def stream_intent(request: Request, payload: dict = Body(...)) -> StreamingResponse:
|
|
"""Drive one turn of the intent dialogue via SSE.
|
|
|
|
Body: ``{thread_id, message, model?}``. Streams upstream LangGraph frames
|
|
through transparently, then appends one status frame:
|
|
|
|
{
|
|
"status": "asking" | "done" | "error",
|
|
"content": "<final AI text>",
|
|
"summary": {...} | null // only present when status == "done"
|
|
}
|
|
"""
|
|
thread_id = str(payload.get("thread_id") or "").strip()
|
|
message = str(payload.get("message") or "").strip()
|
|
model_name = str(payload.get("model") or "deepseek-chat")
|
|
if not thread_id:
|
|
raise HTTPException(status_code=400, detail="thread_id is required")
|
|
if not message:
|
|
raise HTTPException(status_code=400, detail="message is required")
|
|
|
|
base = _loopback_base(request)
|
|
headers = _auth_headers(request)
|
|
|
|
async def stream_generator() -> AsyncIterator[str]:
|
|
# trust_env=False: same-process loopback to 127.0.0.1 — honoring
|
|
# HTTP_PROXY / HTTPS_PROXY env vars (commonly set on dev machines)
|
|
# causes httpx to route loopback calls through an unreachable upstream
|
|
# proxy and wait ~40 seconds for timeout before falling back.
|
|
async with httpx.AsyncClient(timeout=None, trust_env=False) as client:
|
|
try:
|
|
body = _run_payload(
|
|
agent_name=INTENT_AGENT_ID,
|
|
model_name=model_name,
|
|
thread_id=thread_id,
|
|
new_message=message,
|
|
# Intent agent only ever needs ask_clarification. Block
|
|
# everything else so the model can't drift into solutioning
|
|
# (e.g. saving a design doc with write_file) — the model
|
|
# tends to ignore SOUL prose, hard exclusion is reliable.
|
|
excluded_tools=[
|
|
"web_search",
|
|
"present_files",
|
|
"view_image",
|
|
"agent_orchestration",
|
|
"bash",
|
|
"ls",
|
|
"read_file",
|
|
"write_file",
|
|
"str_replace",
|
|
"memory",
|
|
"hindsight_recall",
|
|
"hindsight_reflect",
|
|
"hindsight_retain",
|
|
"task",
|
|
"write_todos",
|
|
],
|
|
skill_stop_names=[],
|
|
)
|
|
|
|
captured: list[str] = []
|
|
async for line, done in _stream_upstream(client, base, headers, thread_id, body):
|
|
if line is not None:
|
|
yield line if line.endswith("\n") else line + "\n"
|
|
elif done is not None:
|
|
captured = done
|
|
|
|
messages = _parse_last_messages(captured)
|
|
# Find the latest AIMessage with tool_calls (typically
|
|
# ask_clarification) — falls back to the latest plain
|
|
# AIMessage when the model produced no tool call.
|
|
dispatch_msg = _find_dispatch_message(messages)
|
|
content = _flatten_content(dispatch_msg.get("content", ""))
|
|
tool_calls = dispatch_msg.get("tool_calls") or []
|
|
|
|
# Clarification branch: the agent ended its run by calling
|
|
# ask_clarification. ClarificationMiddleware intercepts with
|
|
# Command(goto=END) and stashes the structured question in the
|
|
# tool_call args, so the AIMessage `content` is only the
|
|
# preamble ("下面我用几个问题确认:") and stops there. We splice
|
|
# the question back into the user-facing text so the frontend
|
|
# actually shows what the agent wants to ask.
|
|
clarification_call = next(
|
|
(tc for tc in tool_calls if (tc.get("name") or "").lower() == "ask_clarification"),
|
|
None,
|
|
)
|
|
if clarification_call is not None:
|
|
cargs = clarification_call.get("args") or {}
|
|
question = str(cargs.get("question") or "").strip()
|
|
follow_up = _find_trailing_ai_text(messages, exclude_text=content)
|
|
# Build a single content string the frontend can render
|
|
# as-is: preamble (if any) + the actual question.
|
|
parts: list[str] = []
|
|
if content.strip():
|
|
parts.append(content.strip())
|
|
if follow_up and follow_up != question:
|
|
parts.append(follow_up)
|
|
if question:
|
|
parts.append(question)
|
|
content = "\n\n".join(parts) if parts else content
|
|
|
|
# Scan ALL AI messages for the [INTENT_READY] marker, not just
|
|
# dispatch_msg.content. The model often emits the summary in an
|
|
# earlier AI message and then keeps yapping (or doing extra
|
|
# tool calls) — the marker is the source of truth, wherever it
|
|
# appears in the run.
|
|
summary: dict[str, Any] | None = None
|
|
ready_text: str = ""
|
|
for msg in reversed(messages):
|
|
if not isinstance(msg, dict):
|
|
continue
|
|
if (msg.get("type") or "").lower() not in ("ai", "aimessage", "aimessagechunk"):
|
|
continue
|
|
msg_text = _flatten_content(msg.get("content", ""))
|
|
m = _INTENT_READY_PATTERN.search(msg_text)
|
|
if m:
|
|
try:
|
|
summary = json.loads(m.group(1))
|
|
except json.JSONDecodeError as exc:
|
|
logger.warning("intent summary JSON parse failed: %s", exc)
|
|
summary = None
|
|
ready_text = msg_text
|
|
break
|
|
# When the marker was found in an earlier message, surface that
|
|
# message's text (which contains the user-facing 📋 汇总 + JSON)
|
|
# instead of the final filler AI text like "好的，收到".
|
|
if summary and ready_text:
|
|
content = ready_text
|
|
|
|
status_value = "done" if summary else "asking"
|
|
final_frame: dict[str, Any] = {
|
|
"status": status_value,
|
|
"content": content,
|
|
"summary": summary,
|
|
}
|
|
yield f"data: {json.dumps(final_frame, ensure_ascii=False)}\n\n"
|
|
|
|
except HTTPException as exc:
|
|
logger.warning("intent stream HTTPException: %s %s", exc.status_code, exc.detail)
|
|
err = {"status": "error", "content": f"[{exc.status_code}] {exc.detail}", "summary": None}
|
|
yield f"data: {json.dumps(err, ensure_ascii=False)}\n\n"
|
|
except Exception as exc:
|
|
logger.exception("intent stream unexpected failure")
|
|
err = {"status": "error", "content": f"unexpected: {exc!r}", "summary": None}
|
|
yield f"data: {json.dumps(err, ensure_ascii=False)}\n\n"
|
|
|
|
return StreamingResponse(
|
|
stream_generator(),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"X-Accel-Buffering": "no",
|
|
"Connection": "keep-alive",
|
|
},
|
|
)
|