alvaro
4b0108ecf0
Opening a report or podcast link now establishes an authenticated starting reference for the conversation. Users can select other accessible content or compare reports while retaining the original link, and TALKFIRST greetings can name the verified report or podcast. Release notes: - Added authenticated report and episode link resolution with bounded lookups and unavailable-content handling - Added explicit conversation selection without restricting searches or changing the reference during comparisons - Added contextual TALKFIRST greetings through both voice-proxy and HTTP reasoning paths - Preserved backend permission checks, conversation isolation, cancellation, and the existing speech protocol Reviewed-on: AquaPortal/s2s#2
109 lines
4.9 KiB
Python
109 lines
4.9 KiB
Python
"""Conversation-aware HTTP reasoning with the same MCP executor as the relay."""
|
|
|
|
import asyncio
|
|
from contextlib import suppress
|
|
|
|
import httpx
|
|
from fastapi import HTTPException, Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from vocero_api.core import config as settings
|
|
from vocero_api.models.mcp import IntegrationError
|
|
from vocero_api.models.requests import ReasoningRequest
|
|
from vocero_api.services import content_context, conversations, mcp_tools
|
|
|
|
|
|
async def response_text(messages: list[dict[str, str]], headers: dict[str, str]) -> str:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=30, follow_redirects=False) as client:
|
|
response = await client.post(
|
|
f"{settings.OPENAI_BASE_URL}/chat/completions",
|
|
headers=headers,
|
|
json={"model": settings.OPENAI_REASONING_MODEL, "messages": messages},
|
|
)
|
|
response.raise_for_status()
|
|
text = response.json()["choices"][0]["message"]["content"]
|
|
if isinstance(text, str) and text.strip() and len(text) <= 64000:
|
|
return text.strip()
|
|
except (httpx.HTTPError, ValueError, KeyError, IndexError, TypeError):
|
|
pass
|
|
raise IntegrationError(502, "El servicio de razonamiento no devolvió una respuesta válida.")
|
|
|
|
|
|
async def run_turn(
|
|
req: ReasoningRequest, conversation: conversations.Conversation, turn: mcp_tools.ToolTurn, headers: dict[str, str]
|
|
) -> str:
|
|
catalog = await content_context.catalog(conversation.backend_token)
|
|
turn.require_current()
|
|
conversation.catalog = catalog
|
|
base = settings.S2S_PRIMARY_PROMPT or req.instructions
|
|
prior = [message for previous_turn in conversation.history for message in previous_turn]
|
|
notes = f"User-provided reference notes:\n{req.user_context[:12000]}\n\n" if req.user_context else ""
|
|
current = [{"role": "user", "content": notes + req.transcript.strip()}]
|
|
while True:
|
|
system = {"role": "system", "content": mcp_tools.instructions(base, conversation)}
|
|
answer = await response_text([system, *prior, *current], headers)
|
|
turn.require_current()
|
|
chosen = mcp_tools.selection(answer)
|
|
current.append({"role": "assistant", "content": answer})
|
|
if chosen is None:
|
|
conversation.greeting_pending = False
|
|
conversation.history.append(current)
|
|
while (
|
|
len(conversation.history) > 6
|
|
or sum(len(message["content"]) for history in conversation.history for message in history) > 96000
|
|
):
|
|
conversation.history.pop(0)
|
|
return answer
|
|
name, arguments = chosen
|
|
result = await turn.call(name, arguments)
|
|
current.append({"role": "user", "content": mcp_tools.result_text(name, result)})
|
|
|
|
|
|
async def reply(req: ReasoningRequest, request: Request, headers: dict[str, str]) -> JSONResponse:
|
|
conversation: conversations.Conversation | None = None
|
|
try:
|
|
conversation = conversations.resolve(request)
|
|
if not req.transcript.strip() or len(req.transcript) > 8000:
|
|
raise IntegrationError(400, "La consulta debe tener entre 1 y 8000 caracteres.")
|
|
try:
|
|
client_turn = int(request.headers.get("x-vocero-turn", "0"))
|
|
except ValueError:
|
|
raise IntegrationError(400, "El turno de la conversación no es válido.") from None
|
|
if client_turn <= conversation.client_turn:
|
|
raise IntegrationError(409, "Este turno ya fue reemplazado.")
|
|
conversation.client_turn = client_turn
|
|
turn = mcp_tools.ToolTurn(conversation, conversation.interrupt())
|
|
task = asyncio.create_task(run_turn(req, conversation, turn, headers))
|
|
conversation.pending = task
|
|
|
|
async def watch_connection() -> None:
|
|
while not task.done():
|
|
if await request.is_disconnected():
|
|
if conversation.pending is task:
|
|
conversation.interrupt()
|
|
else:
|
|
task.cancel()
|
|
return
|
|
await asyncio.sleep(0.1)
|
|
|
|
watcher = asyncio.create_task(watch_connection())
|
|
try:
|
|
async with asyncio.timeout(180):
|
|
answer = await task
|
|
return JSONResponse({"answer": answer}, headers={"Cache-Control": "no-store"})
|
|
finally:
|
|
watcher.cancel()
|
|
with suppress(asyncio.CancelledError):
|
|
await watcher
|
|
if conversation.pending is task:
|
|
conversation.pending = None
|
|
except IntegrationError as exc:
|
|
if exc.status == 401 and conversation is not None:
|
|
conversations.store.remove(conversation.id)
|
|
raise HTTPException(exc.status, exc.message) from None
|
|
except TimeoutError:
|
|
raise HTTPException(504, "La consulta tardó demasiado. Intenta con un período más corto.") from None
|
|
except asyncio.CancelledError:
|
|
return JSONResponse({"detail": "La consulta fue cancelada."}, status_code=409)
|