diff --git a/backend/.env.example b/backend/.env.example index 000cc95..c943b72 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -18,6 +18,7 @@ CORS_ORIGINS=http://localhost:3000,http://127.0.0.1:3000 # 用独立随机值签名长期 API Key 摘要和短期 Client Secret；生产环境必须修改。 REALTIME_TOKEN_SECRET=replace-with-a-long-random-secret REALTIME_CLIENT_SECRET_TTL_SECONDS=60 +RESPONSES_REQUEST_TIMEOUT_SECONDS=120 # ---- RustFS / S3-compatible storage ---- # Use 127.0.0.1 (not localhost) if HTTP_PROXY is set — proxies often return 502 for localhost. diff --git a/backend/app.py b/backend/app.py index 2ddfa66..a57614f 100644 --- a/backend/app.py +++ b/backend/app.py @@ -25,6 +25,7 @@ import settings import uvicorn from db.session import sync_default_tools, sync_interface_definitions from services.knowledge import recover_interrupted_documents +from services.openai_responses.state import recover_interrupted_responses from services.post_call.worker import recover_interrupted_analyses, run_analysis_worker from services.webhooks.worker import recover_interrupted_webhooks, run_webhook_worker from services.test_runs.orchestrator import ( @@ -45,6 +46,7 @@ from routes import ( model_registry, node_types, openai_realtime, + openai_responses, realtime_api_keys, site_settings, test_cases, @@ -64,6 +66,7 @@ async def lifespan(_app: FastAPI): await recover_interrupted_analyses() await recover_interrupted_webhooks() await recover_interrupted_test_runs() + await recover_interrupted_responses() analysis_worker = asyncio.create_task( run_analysis_worker(), name="post-call-analysis-worker" ) @@ -102,6 +105,7 @@ app.include_router(mcp_servers.router) app.include_router(model_registry.router) app.include_router(node_types.router) app.include_router(openai_realtime.router) +app.include_router(openai_responses.router) app.include_router(realtime_api_keys.router) app.include_router(site_settings.router) app.include_router(test_cases.router) diff --git a/backend/db/models.py b/backend/db/models.py index f83313d..0e39fdc 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -424,6 +424,52 @@ class RealtimeApiKey(Base): ) +class OpenAIResponse(Base): + """Persisted state for one public Responses API generation.""" + + __tablename__ = "openai_responses" + + id: Mapped[str] = mapped_column(String(40), primary_key=True) + api_key_id: Mapped[str] = mapped_column( + String(40), + ForeignKey("realtime_api_keys.id", ondelete="CASCADE"), + index=True, + ) + assistant_id: Mapped[str | None] = mapped_column( + String(40), + ForeignKey("assistants.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + conversation_id: Mapped[str] = mapped_column( + String(40), + ForeignKey("conversation_sessions.id", ondelete="CASCADE"), + index=True, + ) + previous_response_id: Mapped[str | None] = mapped_column( + String(40), + ForeignKey("openai_responses.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + status: Mapped[str] = mapped_column(String(16), index=True, default="in_progress") + input_items: Mapped[list] = mapped_column(JSONB, default=list) + output_items: Mapped[list] = mapped_column(JSONB, default=list) + context_messages: Mapped[list] = mapped_column(JSONB, default=list) + runtime_state: Mapped[dict] = mapped_column(JSONB, default=dict) + error: Mapped[dict | None] = mapped_column(JSONB, nullable=True) + dynamic_variables: Mapped[dict] = mapped_column(JSONB, default=dict) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + completed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + + class WebhookDelivery(Base): """Immutable event snapshot plus its asynchronous delivery state.""" diff --git a/backend/migrations/versions/20260826_0024_add_openai_responses.py b/backend/migrations/versions/20260826_0024_add_openai_responses.py new file mode 100644 index 0000000..dadf7bd --- /dev/null +++ b/backend/migrations/versions/20260826_0024_add_openai_responses.py @@ -0,0 +1,106 @@ +"""add stateful OpenAI Responses records + +Revision ID: 20260826_0024 +Revises: 20260818_0023 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +revision: str = "20260826_0024" +down_revision: str | Sequence[str] | None = "20260818_0023" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "openai_responses", + sa.Column("id", sa.String(length=40), nullable=False), + sa.Column("api_key_id", sa.String(length=40), nullable=False), + sa.Column("assistant_id", sa.String(length=40), nullable=True), + sa.Column("conversation_id", sa.String(length=40), nullable=False), + sa.Column("previous_response_id", sa.String(length=40), nullable=True), + sa.Column( + "status", sa.String(length=16), server_default="in_progress", nullable=False + ), + sa.Column( + "input_items", + postgresql.JSONB(astext_type=sa.Text()), + server_default=sa.text("'[]'::jsonb"), + nullable=False, + ), + sa.Column( + "output_items", + postgresql.JSONB(astext_type=sa.Text()), + server_default=sa.text("'[]'::jsonb"), + nullable=False, + ), + sa.Column( + "context_messages", + postgresql.JSONB(astext_type=sa.Text()), + server_default=sa.text("'[]'::jsonb"), + nullable=False, + ), + sa.Column( + "runtime_state", + postgresql.JSONB(astext_type=sa.Text()), + server_default=sa.text("'{}'::jsonb"), + nullable=False, + ), + sa.Column( + "error", postgresql.JSONB(astext_type=sa.Text()), nullable=True + ), + sa.Column( + "dynamic_variables", + postgresql.JSONB(astext_type=sa.Text()), + server_default=sa.text("'{}'::jsonb"), + nullable=False, + ), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.ForeignKeyConstraint( + ["api_key_id"], ["realtime_api_keys.id"], ondelete="CASCADE" + ), + sa.ForeignKeyConstraint( + ["assistant_id"], ["assistants.id"], ondelete="SET NULL" + ), + sa.ForeignKeyConstraint( + ["conversation_id"], ["conversation_sessions.id"], ondelete="CASCADE" + ), + sa.ForeignKeyConstraint( + ["previous_response_id"], ["openai_responses.id"], ondelete="SET NULL" + ), + sa.PrimaryKeyConstraint("id"), + ) + for column in ( + "api_key_id", + "assistant_id", + "conversation_id", + "previous_response_id", + "status", + ): + op.create_index( + f"ix_openai_responses_{column}", "openai_responses", [column] + ) + + +def downgrade() -> None: + op.drop_table("openai_responses") diff --git a/backend/routes/openai_responses.py b/backend/routes/openai_responses.py new file mode 100644 index 0000000..e85595f --- /dev/null +++ b/backend/routes/openai_responses.py @@ -0,0 +1,358 @@ +"""Public stateful OpenAI Responses-compatible text endpoints.""" + +from __future__ import annotations + +import asyncio +import json +from collections.abc import AsyncIterator +from typing import Any + +import settings +from db.session import SessionLocal +from fastapi import APIRouter, Request +from fastapi.responses import JSONResponse, StreamingResponse +from loguru import logger +from services.openai_realtime.auth import ( + RealtimeAuthError, + RealtimeCredential, + authenticate_bearer, + bearer_from_authorization, +) +from services.openai_responses.protocol import ( + ResponseRequest, + ResponsesAPIError, + error_payload, + new_id, + parse_request, + response_payload, +) +from services.openai_responses.runner import ResponseRunError, ResponsesTextRunner +from services.openai_responses.state import ( + PreparedResponse, + finish_response, + get_response, + prepare_response, +) + + +router = APIRouter(prefix="/v1/responses", tags=["openai-responses"]) + + +async def _credential(request: Request) -> RealtimeCredential: + token = bearer_from_authorization(request.headers.get("authorization")) + async with SessionLocal() as db: + credential = await authenticate_bearer(db, token) + if credential.ephemeral: + raise RealtimeAuthError("A long-lived API key is required") + return credential + + +def _json_error(error: ResponsesAPIError) -> JSONResponse: + return JSONResponse(status_code=error.status_code, content=error_payload(error)) + + +def _sse(event: dict[str, Any]) -> bytes: + return ( + f"event: {event['type']}\n" + f"data: {json.dumps(event, ensure_ascii=False)}\n\n" + ).encode("utf-8") + + +def _text_item(item_id: str, text: str) -> dict[str, Any]: + return { + "id": item_id, + "type": "message", + "status": "completed", + "role": "assistant", + "content": [ + { + "type": "output_text", + "text": text, + "annotations": [], + } + ], + } + + +def _output_items( + *, + text: str, + text_item_id: str, + function_calls: list[dict[str, Any]], +) -> list[dict[str, Any]]: + items: list[dict[str, Any]] = [] + if text: + items.append(_text_item(text_item_id, text)) + items.extend(function_calls) + return items + + +async def _run_prepared( + prepared: PreparedResponse, + *, + on_delta=None, + text_item_id: str, +): + try: + async with asyncio.timeout(settings.RESPONSES_REQUEST_TIMEOUT_SECONDS): + result = await ResponsesTextRunner().run( + prepared.config, + history_messages=prepared.history_messages, + current_messages=prepared.current_messages, + runtime_state=prepared.runtime_state, + client_tool_results=prepared.client_tool_results, + on_delta=on_delta, + ) + return await finish_response( + prepared.row.id, + status="completed", + output_items=_output_items( + text=result.text, + text_item_id=text_item_id, + function_calls=result.function_calls, + ), + context_messages=result.context_messages, + dynamic_variables=result.dynamic_variables, + runtime_state=result.runtime_state, + ) + except TimeoutError as exc: + await finish_response( + prepared.row.id, + status="failed", + output_items=[], + error={"code": "response_timeout", "message": "Response timed out"}, + ) + raise ResponsesAPIError( + "Response timed out", code="response_timeout", status_code=504 + ) from exc + except asyncio.CancelledError: + await finish_response( + prepared.row.id, + status="cancelled", + output_items=[], + error={"code": "client_disconnected", "message": "Client disconnected"}, + ) + raise + except ResponseRunError as exc: + await finish_response( + prepared.row.id, + status="failed", + output_items=[], + error={"code": "response_failed", "message": str(exc)}, + ) + raise ResponsesAPIError( + "Response generation failed", + code="response_failed", + status_code=500, + ) from exc + except Exception as exc: # noqa: BLE001 + logger.exception(f"Responses generation failed: {exc}") + await finish_response( + prepared.row.id, + status="failed", + output_items=[], + error={"code": "internal_error", "message": "Response generation failed"}, + ) + raise ResponsesAPIError( + "Response generation failed", code="internal_error", status_code=500 + ) from exc + + +async def _stream_response( + prepared: PreparedResponse, + *, + text_item_id: str, +) -> AsyncIterator[bytes]: + queue: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue() + text_started = False + + async def on_delta(delta: str) -> None: + await queue.put({"kind": "delta", "delta": delta}) + + async def generate() -> None: + try: + row = await _run_prepared( + prepared, + on_delta=on_delta, + text_item_id=text_item_id, + ) + await queue.put({"kind": "completed", "row": row}) + except ResponsesAPIError as exc: + await queue.put({"kind": "failed", "error": exc}) + finally: + await queue.put(None) + + task = asyncio.create_task(generate(), name=f"stream-{prepared.row.id}") + try: + created = response_payload(prepared.row) + yield _sse({"type": "response.created", "response": created}) + yield _sse({"type": "response.in_progress", "response": created}) + while True: + event = await queue.get() + if event is None: + break + if event["kind"] == "delta": + if not text_started: + text_started = True + pending_item = _text_item(text_item_id, "") + pending_item["status"] = "in_progress" + yield _sse( + { + "type": "response.output_item.added", + "output_index": 0, + "item": pending_item, + } + ) + yield _sse( + { + "type": "response.content_part.added", + "item_id": text_item_id, + "output_index": 0, + "content_index": 0, + "part": { + "type": "output_text", + "text": "", + "annotations": [], + }, + } + ) + yield _sse( + { + "type": "response.output_text.delta", + "item_id": text_item_id, + "output_index": 0, + "content_index": 0, + "delta": event["delta"], + } + ) + continue + if event["kind"] == "failed": + error: ResponsesAPIError = event["error"] + yield _sse({"type": "error", **error_payload(error)}) + failed = await get_response( + prepared.row.id, api_key_id=prepared.row.api_key_id + ) + yield _sse( + {"type": "response.failed", "response": response_payload(failed)} + ) + continue + + row = event["row"] + payload = response_payload(row) + output_index = 0 + for item in row.output_items or []: + if item.get("type") == "message": + text = str((item.get("content") or [{}])[0].get("text") or "") + if not text_started: + yield _sse( + { + "type": "response.output_item.added", + "output_index": output_index, + "item": { + **item, + "status": "in_progress", + "content": [], + }, + } + ) + yield _sse( + { + "type": "response.content_part.added", + "item_id": item["id"], + "output_index": output_index, + "content_index": 0, + "part": { + "type": "output_text", + "text": "", + "annotations": [], + }, + } + ) + yield _sse( + { + "type": "response.output_text.done", + "item_id": item["id"], + "output_index": output_index, + "content_index": 0, + "text": text, + } + ) + yield _sse( + { + "type": "response.content_part.done", + "item_id": item["id"], + "output_index": output_index, + "content_index": 0, + "part": item["content"][0], + } + ) + yield _sse( + { + "type": "response.output_item.done", + "output_index": output_index, + "item": item, + } + ) + else: + yield _sse( + { + "type": "response.output_item.added", + "output_index": output_index, + "item": item, + } + ) + yield _sse( + { + "type": "response.output_item.done", + "output_index": output_index, + "item": item, + } + ) + output_index += 1 + yield _sse({"type": "response.completed", "response": payload}) + finally: + if not task.done(): + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + +@router.post("") +async def create_response(request: Request): + try: + credential = await _credential(request) + body = await request.json() + parsed: ResponseRequest = parse_request(body) + prepared = await prepare_response(parsed, api_key_id=credential.api_key_id) + text_item_id = new_id("msg") + if parsed.stream: + return StreamingResponse( + _stream_response(prepared, text_item_id=text_item_id), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) + row = await _run_prepared(prepared, text_item_id=text_item_id) + return response_payload(row) + except RealtimeAuthError as exc: + return _json_error( + ResponsesAPIError(str(exc), code="invalid_api_key", status_code=401) + ) + except (json.JSONDecodeError, UnicodeDecodeError): + return _json_error(ResponsesAPIError("Request body must be valid JSON")) + except ResponsesAPIError as exc: + return _json_error(exc) + except ValueError as exc: + return _json_error(ResponsesAPIError(str(exc))) + + +@router.get("/{response_id}") +async def retrieve_response(response_id: str, request: Request): + try: + credential = await _credential(request) + row = await get_response(response_id, api_key_id=credential.api_key_id) + return response_payload(row) + except RealtimeAuthError as exc: + return _json_error( + ResponsesAPIError(str(exc), code="invalid_api_key", status_code=401) + ) + except ResponsesAPIError as exc: + return _json_error(exc) diff --git a/backend/services/brains/dify_brain.py b/backend/services/brains/dify_brain.py index bee51e1..d7778a7 100644 --- a/backend/services/brains/dify_brain.py +++ b/backend/services/brains/dify_brain.py @@ -24,6 +24,8 @@ class DifyBrain(BaseBrain): def __init__(self): self._user_id = f"ai-video-{uuid4().hex}" self._client: AsyncClient | None = None + self._llm: DifyLLMService | None = None + self._restored_conversation_id = "" def _get_client(self, cfg: AssistantConfig) -> AsyncClient: if self._client is None: @@ -54,8 +56,25 @@ class DifyBrain(BaseBrain): return cfg.greeting def build_llm(self, cfg: AssistantConfig, context: LLMContext) -> FrameProcessor: - return DifyLLMService( + self._llm = DifyLLMService( cfg, client=self._get_client(cfg), user_id=self._user_id, + conversation_id=self._restored_conversation_id, ) + return self._llm + + def restore_text_state(self, snapshot: dict) -> None: + if snapshot.get("type") != "dify": + raise ValueError("Responses runtime state does not belong to Dify") + self._user_id = str(snapshot.get("user_id") or self._user_id) + self._restored_conversation_id = str(snapshot.get("conversation_id") or "") + + def text_state_snapshot(self) -> dict: + return { + "type": "dify", + "user_id": self._user_id, + "conversation_id": ( + self._llm.conversation_id if self._llm is not None else "" + ), + } diff --git a/backend/services/brains/dify_llm.py b/backend/services/brains/dify_llm.py index 849d36e..ac4c6de 100644 --- a/backend/services/brains/dify_llm.py +++ b/backend/services/brains/dify_llm.py @@ -54,6 +54,7 @@ class DifyLLMService(LLMService): *, client: AsyncClient | None = None, user_id: str | None = None, + conversation_id: str = "", ): super().__init__( settings=LLMSettings( @@ -75,7 +76,11 @@ class DifyLLMService(LLMService): api_base=normalize_api_base(cfg.dify_api_url), ) self._user_id = user_id or f"ai-video-{uuid4().hex}" - self._conversation_id = "" + self._conversation_id = conversation_id + + @property + def conversation_id(self) -> str: + return self._conversation_id async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) diff --git a/backend/services/brains/fastgpt_brain.py b/backend/services/brains/fastgpt_brain.py index 3d0e416..8733f98 100644 --- a/backend/services/brains/fastgpt_brain.py +++ b/backend/services/brains/fastgpt_brain.py @@ -66,3 +66,11 @@ class FastGPTBrain(BaseBrain): def build_llm(self, cfg: AssistantConfig, context: LLMContext) -> FrameProcessor: return FastGPTLLMService(cfg, chat_id=self._chat_id) + + def restore_text_state(self, snapshot: dict) -> None: + if snapshot.get("type") != "fastgpt": + raise ValueError("Responses runtime state does not belong to FastGPT") + self._chat_id = str(snapshot.get("chat_id") or self._chat_id) + + def text_state_snapshot(self) -> dict: + return {"type": "fastgpt", "chat_id": self._chat_id} diff --git a/backend/services/brains/prompt_brain.py b/backend/services/brains/prompt_brain.py index bd48ed6..6017f83 100644 --- a/backend/services/brains/prompt_brain.py +++ b/backend/services/brains/prompt_brain.py @@ -83,6 +83,7 @@ class PromptBrain(BaseBrain): self._startup_failed = False self._greeting_pending = False self._entry_dispatched = False + self._restored_text_state: dict[str, Any] | None = None async def greeting(self, cfg: AssistantConfig) -> str: # The built-in opening Message owns the greeting so speech and the @@ -127,6 +128,12 @@ class PromptBrain(BaseBrain): self._startup_failed = False self._greeting_pending = False self._entry_dispatched = False + if self._restored_text_state is not None: + state = self._restored_text_state + self._preflight_finished = bool(state.get("preflight_finished")) + self._opening_finished = bool(state.get("opening_finished")) + self._opening_started = self._opening_finished + self._entry_dispatched = bool(state.get("entry_dispatched")) schemas: list[FunctionSchema] = [] registered_names: set[str] = set() binding_by_id = {binding.tool_id: binding for binding in cfg.tool_bindings} @@ -171,6 +178,21 @@ class PromptBrain(BaseBrain): ) runtime.set_tools(schemas) + def restore_text_state(self, snapshot: dict[str, Any]) -> None: + """Restore Prompt startup progress for a later Responses API turn.""" + if snapshot.get("type") != "prompt": + raise ValueError("Responses runtime state does not belong to a prompt") + self._restored_text_state = dict(snapshot) + + def text_state_snapshot(self) -> dict[str, Any]: + """Prevent completed startup actions from running again next turn.""" + return { + "type": "prompt", + "preflight_finished": self._preflight_finished, + "opening_finished": self._opening_finished, + "entry_dispatched": self._entry_dispatched, + } + async def setup_realtime( self, cfg: AssistantConfig, diff --git a/backend/services/brains/workflow_brain.py b/backend/services/brains/workflow_brain.py index 41597cd..476c666 100644 --- a/backend/services/brains/workflow_brain.py +++ b/backend/services/brains/workflow_brain.py @@ -187,6 +187,7 @@ class WorkflowBrain(BaseBrain): self._next_message_token = 1 self._pending_message: _MessageContinuation | None = None self._realtime_controller: WorkflowRealtimeController | None = None + self._restored_text_node_id: str | None = None async def greeting(self, _cfg: AssistantConfig) -> str: """Workflow opening speech belongs to an explicit Message or Agent.""" @@ -274,6 +275,18 @@ class WorkflowBrain(BaseBrain): if self._realtime_controller is not None: await self._realtime_controller.start() return + if self._restored_text_node_id: + node_id = self._restored_text_node_id + self._restored_text_node_id = None + self._state.enter(node_id, WorkflowStatus.WAITING_USER) + await self._emit_node_active(node_id) + await self._emit_variables(reason="restored", node_id=node_id) + if self._manager is None: + raise RuntimeError("Workflow FlowManager 尚未初始化") + node_config = await self._resolve_path(node_id) + await self._activate_node_config(node_config, initialize=True) + logger.info(f"工作流文本状态已恢复: 当前节点={self._manager.current_node}") + return self._state.enter(self._engine.start_id, WorkflowStatus.STARTING) await self._emit_node_active(self._engine.start_id) await self._emit_variables( @@ -287,6 +300,33 @@ class WorkflowBrain(BaseBrain): await self._activate_node_config(node_config, initialize=True) logger.info(f"工作流模式启用: 当前节点={self._manager.current_node}") + def restore_text_state(self, snapshot: dict[str, Any]) -> None: + """Restore one stable Responses API workflow node before connecting.""" + if not snapshot: + return + if snapshot.get("type") != "workflow": + raise ValueError("Responses runtime state does not belong to a workflow") + if snapshot.get("revision") != self._engine.revision: + raise ValueError("Workflow definition changed after the previous response") + node_id = str(snapshot.get("current_node_id") or "") + if not self._engine.node_type(node_id): + raise ValueError("Previous workflow node no longer exists") + self._restored_text_node_id = node_id + + def text_state_snapshot(self) -> dict[str, Any]: + """Return the minimum durable state needed by the next text request.""" + node_id = ( + str(self._manager.current_node) + if self._manager is not None and self._manager.current_node + else self._state.current_node_id + ) + return { + "type": "workflow", + "revision": self._engine.revision, + "current_node_id": node_id, + "status": self._state.status.value, + } + async def _initial_node_config(self) -> NodeConfig: """Only a default-only Start advances before the first user turn.""" outgoing = self._engine.outgoing(self._engine.start_id) diff --git a/backend/services/openai_responses/__init__.py b/backend/services/openai_responses/__init__.py new file mode 100644 index 0000000..bba2138 --- /dev/null +++ b/backend/services/openai_responses/__init__.py @@ -0,0 +1 @@ +"""Stateful OpenAI Responses-compatible text API.""" diff --git a/backend/services/openai_responses/protocol.py b/backend/services/openai_responses/protocol.py new file mode 100644 index 0000000..2be2937 --- /dev/null +++ b/backend/services/openai_responses/protocol.py @@ -0,0 +1,241 @@ +"""Validation and wire helpers for the supported Responses API subset.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any +from uuid import uuid4 + +from db.models import OpenAIResponse + + +LOCKED_FIELDS = {"instructions", "tools", "tool_choice"} +ALLOWED_FIELDS = { + "input", + "model", + "previous_response_id", + "store", + "stream", + "x_interactive_media", +} +UNSUPPORTED_FIELDS = { + "background", + "conversation", + "include", + "max_output_tokens", + "parallel_tool_calls", + "reasoning", + "temperature", + "text", + "top_p", + "truncation", +} + + +@dataclass +class ResponsesAPIError(ValueError): + message: str + code: str = "invalid_request_error" + param: str | None = None + status_code: int = 400 + + def __str__(self) -> str: + return self.message + + +@dataclass(frozen=True) +class ResponseRequest: + model: str | None + previous_response_id: str | None + input_items: list[dict[str, Any]] + stream: bool + dynamic_variables: dict[str, Any] + + +def error_payload(error: ResponsesAPIError) -> dict[str, Any]: + return { + "error": { + "type": "invalid_request_error", + "code": error.code, + "message": error.message, + "param": error.param, + } + } + + +def _text_content(value: object, *, role: str) -> list[dict[str, str]]: + expected = "input_text" if role == "user" else "output_text" + if isinstance(value, str): + text = value.strip() + if not text: + raise ResponsesAPIError("Message content cannot be empty", param="input") + return [{"type": expected, "text": text}] + if not isinstance(value, list) or not value: + raise ResponsesAPIError("Message content must be text", param="input") + result: list[dict[str, str]] = [] + for part in value: + if not isinstance(part, dict) or part.get("type") != expected: + raise ResponsesAPIError( + f"Only {expected} content is supported", param="input" + ) + text = str(part.get("text") or "").strip() + if not text: + raise ResponsesAPIError("Message text cannot be empty", param="input") + result.append({"type": expected, "text": text}) + return result + + +def normalize_input(value: object) -> list[dict[str, Any]]: + if isinstance(value, str): + return [ + { + "type": "message", + "role": "user", + "content": _text_content(value, role="user"), + } + ] + if not isinstance(value, list) or not value: + raise ResponsesAPIError("input must be a non-empty string or array", param="input") + + items: list[dict[str, Any]] = [] + for index, raw in enumerate(value): + param = f"input[{index}]" + if not isinstance(raw, dict): + raise ResponsesAPIError("Input item must be an object", param=param) + item_type = str(raw.get("type") or "message") + if item_type == "message": + role = str(raw.get("role") or "user") + if role not in {"user", "assistant"}: + raise ResponsesAPIError( + "Message role must be user or assistant", param=f"{param}.role" + ) + items.append( + { + "type": "message", + "role": role, + "content": _text_content(raw.get("content"), role=role), + } + ) + continue + if item_type == "function_call_output": + call_id = str(raw.get("call_id") or "").strip() + if not call_id: + raise ResponsesAPIError( + "function_call_output requires call_id", + param=f"{param}.call_id", + ) + output = raw.get("output") + if not isinstance(output, str): + output = json.dumps(output, ensure_ascii=False, default=str) + items.append( + { + "type": "function_call_output", + "call_id": call_id, + "output": output, + } + ) + continue + raise ResponsesAPIError( + f"Unsupported input item type: {item_type}", + code="unsupported_parameter", + param=f"{param}.type", + ) + return items + + +def parse_request(body: object) -> ResponseRequest: + if not isinstance(body, dict): + raise ResponsesAPIError("Request body must be an object") + unknown = sorted(set(body) - ALLOWED_FIELDS - LOCKED_FIELDS - UNSUPPORTED_FIELDS) + if unknown: + raise ResponsesAPIError( + f"Unsupported parameter: {unknown[0]}", + code="unsupported_parameter", + param=unknown[0], + ) + locked = sorted(LOCKED_FIELDS.intersection(body)) + if locked: + raise ResponsesAPIError( + f"{locked[0]} is owned by the selected assistant", + code="immutable_configuration", + param=locked[0], + ) + unsupported = sorted(UNSUPPORTED_FIELDS.intersection(body)) + if unsupported: + raise ResponsesAPIError( + f"Unsupported parameter: {unsupported[0]}", + code="unsupported_parameter", + param=unsupported[0], + ) + if "store" in body and body["store"] is not True: + raise ResponsesAPIError( + "store must be true for this stateful API", + code="unsupported_parameter", + param="store", + ) + stream = body.get("stream", False) + if not isinstance(stream, bool): + raise ResponsesAPIError("stream must be a boolean", param="stream") + extension = body.get("x_interactive_media") or {} + if not isinstance(extension, dict): + raise ResponsesAPIError( + "x_interactive_media must be an object", param="x_interactive_media" + ) + variables = extension.get("variables") or {} + if not isinstance(variables, dict): + raise ResponsesAPIError( + "variables must be an object", + param="x_interactive_media.variables", + ) + raw_model = body.get("model") + if raw_model is not None and not isinstance(raw_model, str): + raise ResponsesAPIError("model must be a string", param="model") + raw_previous = body.get("previous_response_id") + if raw_previous is not None and not isinstance(raw_previous, str): + raise ResponsesAPIError( + "previous_response_id must be a string", param="previous_response_id" + ) + model = (raw_model or "").strip() or None + previous = (raw_previous or "").strip() or None + if not model and not previous: + raise ResponsesAPIError( + "model is required for the first response", param="model" + ) + if "input" not in body: + raise ResponsesAPIError("input is required", param="input") + return ResponseRequest( + model=model, + previous_response_id=previous, + input_items=normalize_input(body["input"]), + stream=stream, + dynamic_variables=dict(variables), + ) + + +def new_id(prefix: str) -> str: + return f"{prefix}_{uuid4().hex}" + + +def response_payload(row: OpenAIResponse) -> dict[str, Any]: + return { + "id": row.id, + "object": "response", + "created_at": int(row.created_at.timestamp()), + "completed_at": ( + int(row.completed_at.timestamp()) if row.completed_at else None + ), + "status": row.status, + "error": row.error, + "incomplete_details": None, + "instructions": None, + "model": f"assistant:{row.assistant_id}" if row.assistant_id else None, + "output": list(row.output_items or []), + "parallel_tool_calls": False, + "previous_response_id": row.previous_response_id, + "store": True, + "tool_choice": "auto", + "tools": [], + "metadata": {}, + "usage": None, + } diff --git a/backend/services/openai_responses/runner.py b/backend/services/openai_responses/runner.py new file mode 100644 index 0000000..e625bfd --- /dev/null +++ b/backend/services/openai_responses/runner.py @@ -0,0 +1,607 @@ +"""Transportless production runner for one Responses API turn.""" + +from __future__ import annotations + +import asyncio +from copy import deepcopy +import json +import time +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from typing import Any + +from db.session import SessionLocal +from loguru import logger +from models import AssistantConfig +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.frames.frames import ( + BotStoppedSpeakingFrame, + CancelFrame, + DataFrame, + EndFrame, + ErrorFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMMessagesAppendFrame, + LLMTextFrame, + ManuallySwitchServiceFrame, + TTSSpeakFrame, +) +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.worker import PipelineParams, PipelineWorker +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + LLMContextAggregatorPair, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.services.llm_service import FunctionCallParams +from pipecat.workers.runner import WorkerRunner +from services.brains import BrainRuntime, build_brain +from services.client_tools import ClientToolResponseWaitMode +from services.knowledge import search as search_knowledge +from services.openai_responses.protocol import new_id +from services.pipecat.call_lifecycle import CallEndCoordinator +from services.pipecat.processors import ( + KnowledgeRetrievalProcessor, + UserTurnRoutingProcessor, +) +from services.pipecat.service_factory import config_with_resource +from services.pipecat.workflow_services import build_workflow_llm_switcher + + +IDLE_SETTLE_SECONDS = 0.2 +WORKER_START_TIMEOUT_SECONDS = 5.0 + + +class ResponseRunError(RuntimeError): + pass + + +DeltaCallback = Callable[[str], Awaitable[None]] + + +@dataclass(frozen=True) +class ResponseRunResult: + text: str + function_calls: list[dict[str, Any]] + context_messages: list[dict[str, Any]] + dynamic_variables: dict[str, Any] + runtime_state: dict[str, Any] + + +@dataclass +class _ResponseWindow: + active_llm: int = 0 + active_assistant: int = 0 + pending_tools: set[str] = field(default_factory=set) + activity_count: int = 0 + last_activity_at: float = field(default_factory=time.monotonic) + + def touch(self) -> None: + self.activity_count += 1 + self.last_activity_at = time.monotonic() + + @property + def idle(self) -> bool: + return self.active_llm == 0 and self.active_assistant == 0 and not self.pending_tools + + +@dataclass +class _BoundaryFrame(DataFrame): + reached: asyncio.Event + + +class _CaptureProcessor(FrameProcessor): + def __init__(self, on_delta: DeltaCallback | None) -> None: + super().__init__() + self.on_delta = on_delta + self.window = _ResponseWindow() + self.deltas: list[str] = [] + self.outputs: list[str] = [] + self.error: str | None = None + + def reset(self) -> None: + self.deltas.clear() + self.outputs.clear() + self.error = None + self.window.activity_count = 0 + self.window.last_activity_at = time.monotonic() + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, LLMFullResponseStartFrame): + self.window.active_llm += 1 + self.window.touch() + elif isinstance(frame, LLMTextFrame): + if frame.text: + self.deltas.append(frame.text) + self.window.touch() + if self.on_delta is not None: + await self.on_delta(frame.text) + elif isinstance(frame, LLMFullResponseEndFrame): + self.window.active_llm = max(0, self.window.active_llm - 1) + self.window.touch() + await self.push_frame(frame, direction) + await self.push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM) + return + elif isinstance(frame, FunctionCallInProgressFrame): + self.window.pending_tools.add(frame.tool_call_id) + self.window.touch() + elif isinstance(frame, FunctionCallResultFrame): + self.window.pending_tools.discard(frame.tool_call_id) + self.window.touch() + elif isinstance(frame, TTSSpeakFrame): + text = frame.text.strip() + if text: + self.outputs.append(text) + self.window.touch() + return + elif isinstance(frame, ErrorFrame): + self.error = frame.error or "Pipeline execution failed" + self.window.pending_tools.clear() + self.window.touch() + elif isinstance(frame, _BoundaryFrame): + frame.reached.set() + self.window.touch() + elif isinstance(frame, (EndFrame, CancelFrame)): + self.window.touch() + await self.push_frame(frame, direction) + + +class ResponsesClientToolPort: + """Turn a configured client-side tool invocation into a Response item.""" + + def __init__(self, results: list[dict[str, Any]] | None = None) -> None: + self.calls: list[dict[str, Any]] = [] + self.called = asyncio.Event() + self._results = list(results or []) + + def supports(self, _function_name: str) -> bool: + return True + + async def call( + self, + function_name: str, + arguments: dict[str, Any], + *, + timeout_seconds: float, + wait_for_response: bool = True, + response_wait_mode: ClientToolResponseWaitMode = "timeout", + interrupt_on_result: bool = False, + ) -> dict[str, Any]: + del timeout_seconds, wait_for_response, response_wait_mode, interrupt_on_result + for index, pending in enumerate(self._results): + if pending.get("name") == function_name: + self._results.pop(index) + return dict(pending.get("result") or {}) + item = { + "id": new_id("fc"), + "type": "function_call", + "status": "completed", + "call_id": new_id("call"), + "name": function_name, + "arguments": json.dumps(arguments, ensure_ascii=False), + } + self.calls.append(item) + self.called.set() + # The HTTP response carries this call back to the caller. The worker is + # cancelled by ResponsesTextRunner after collecting parallel calls. + await asyncio.Future() + raise AssertionError("unreachable") + + +def _context_snapshot( + context: LLMContext, + *, + function_calls: list[dict[str, Any]], + fallback_messages: list[dict[str, Any]] | None = None, +) -> list[dict[str, Any]]: + """Keep model-visible state and align provider call IDs with public IDs.""" + messages = [ + deepcopy(message) + for message in context.get_messages() + if message.get("role") != "system" + ] + if fallback_messages: + messages.extend(deepcopy(fallback_messages)) + for public_call in function_calls: + matched = False + for message in reversed(messages): + for tool_call in reversed(message.get("tool_calls") or []): + function = tool_call.get("function") or {} + if function.get("name") != public_call["name"]: + continue + tool_call["id"] = public_call["call_id"] + matched = True + break + if matched: + break + if not matched: + messages.append( + { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": public_call["call_id"], + "type": "function", + "function": { + "name": public_call["name"], + "arguments": public_call["arguments"], + }, + } + ], + } + ) + return json.loads(json.dumps(messages, ensure_ascii=False, default=str)) + + +def _public_variables(brain, cfg: AssistantConfig) -> dict[str, Any]: + store = getattr(brain, "_store", None) + public_values = getattr(store, "public_values", None) + if callable(public_values): + return dict(public_values()) + return { + name: value + for name, value in (cfg.dynamic_variables or {}).items() + if not name.startswith(("system__", "secret__")) + } + + +def _runtime_state(brain) -> dict[str, Any]: + snapshot = getattr(brain, "text_state_snapshot", None) + return dict(snapshot()) if callable(snapshot) else {} + + +async def _wait_until_idle( + capture: _CaptureProcessor, + runner_task: asyncio.Task[None], + *, + require_activity: bool, + client_tools: ResponsesClientToolPort | None = None, +) -> str: + while True: + if capture.error: + raise ResponseRunError(capture.error) + if client_tools is not None and client_tools.called.is_set(): + # Let calls emitted in the same provider batch enter the port. + await asyncio.sleep(0.05) + return "function_call" + if runner_task.done(): + await runner_task + if capture.outputs or capture.deltas: + return "text" + raise ResponseRunError("Pipeline ended before producing a response") + active = capture.window.activity_count > 0 + settled = time.monotonic() - capture.window.last_activity_at >= IDLE_SETTLE_SECONDS + if capture.window.idle and settled and (active or not require_activity): + return "text" + await asyncio.sleep(0.05) + + +class ResponsesTextRunner: + async def run( + self, + cfg: AssistantConfig, + *, + history_messages: list[dict[str, Any]], + current_messages: list[dict[str, Any]], + runtime_state: dict[str, Any] | None = None, + client_tool_results: list[dict[str, Any]] | None = None, + on_delta: DeltaCallback | None = None, + ) -> ResponseRunResult: + if cfg.runtimeMode != "pipeline": + raise ResponseRunError("Responses API only supports pipeline assistants") + if not current_messages: + raise ResponseRunError("The current response has no usable input") + + brain = build_brain(cfg) + restore_state = getattr(brain, "restore_text_state", None) + if runtime_state and callable(restore_state): + restore_state(runtime_state) + knowledge_config = cfg.knowledge_retrieval_config or {} + knowledge_mode = str(knowledge_config.get("mode") or "automatic") + + def with_knowledge_hint(text: str) -> str: + if cfg.type != "prompt" or not cfg.knowledge_base_id: + return text + mode_text = ( + "系统会在每轮用户问题前自动提供相关资料。" + if knowledge_mode == "automatic" + else "需要业务资料时先调用 search_knowledge_base。" + ) + return f"{text}\n\n你已连接内部知识库。{mode_text}" + + system_prompt = with_knowledge_hint(brain.system_prompt(cfg)) + context = LLMContext( + messages=( + list(history_messages) + if cfg.type == "workflow" + else [ + {"role": "system", "content": system_prompt}, + *history_messages, + ] + ) + ) + graph_settings = cfg.graph.get("settings") or {} + default_resource = cfg.workflow_model_resources.get( + str(graph_settings.get("defaultLlmResourceId") or "") + ) + llm_cfg = ( + config_with_resource(cfg, default_resource) + if cfg.type == "workflow" and default_resource is not None + else cfg + ) + llm = brain.build_llm(llm_cfg, context) + llm_services: dict[str, FrameProcessor] = {} + current_llm_service: FrameProcessor = llm + if cfg.type == "workflow": + llm, llm_services, current_llm_service = build_workflow_llm_switcher( + cfg, llm + ) + + aggregators = LLMContextAggregatorPair(context) + user_aggregator = aggregators.user() + assistant_aggregator = aggregators.assistant() + capture = _CaptureProcessor(on_delta) + knowledge = KnowledgeRetrievalProcessor( + cfg.knowledge_base_id if knowledge_mode == "automatic" else None, + top_n=int(knowledge_config.get("top_n", knowledge_config.get("topN", 5))), + score_threshold=float( + knowledge_config.get( + "score_threshold", knowledge_config.get("scoreThreshold", 0.0) + ) + ), + ) + pipeline = Pipeline( + [ + user_aggregator, + UserTurnRoutingProcessor(brain), + knowledge, + llm, + capture, + assistant_aggregator, + ] + ) + worker = PipelineWorker( + pipeline, + params=PipelineParams(enable_metrics=False), + enable_rtvi=False, + enable_turn_tracking=False, + idle_timeout_secs=None, + ) + runner = WorkerRunner(handle_sigint=False, check_dangling_tasks=False) + worker_started = asyncio.Event() + + @worker.event_handler("on_pipeline_started") + async def on_pipeline_started(_worker, _frame): + worker_started.set() + + @worker.event_handler("on_pipeline_error") + async def on_pipeline_error(_worker, frame): + capture.error = frame.error or "Pipeline execution failed" + capture.window.touch() + + client_tools = ResponsesClientToolPort(client_tool_results) + workflow_vision_scope: dict[str, Any] = {"enabled": False} + + async def queue_call_end(reason: str) -> None: + await worker.queue_frame(EndFrame(reason=reason)) + + call_end = CallEndCoordinator(queue_call_end) + current_service = current_llm_service + + async def switch_services( + llm_resource_id: str | None, + _asr_resource_id: str | None, + _tts_resource_id: str | None, + ) -> None: + nonlocal current_service + target = ( + llm_services.get(llm_resource_id) + if llm_resource_id + else current_llm_service + ) + if target is None: + raise ValueError(f"Workflow LLM resource not found: {llm_resource_id}") + if target is current_service: + return + await worker.queue_frame(ManuallySwitchServiceFrame(service=target)) + current_service = target + + def set_system_prompt(text: str) -> None: + messages = context.get_messages() + text = with_knowledge_hint(text) + if messages and messages[0].get("role") == "system": + messages[0] = {"role": "system", "content": text} + else: + messages.insert(0, {"role": "system", "content": text}) + + knowledge_schema: FunctionSchema | None = None + if cfg.type == "prompt" and cfg.knowledge_base_id and knowledge_mode == "on_demand": + knowledge_schema = FunctionSchema( + name="search_knowledge_base", + description="在当前助手绑定的知识库中检索资料。", + properties={ + "query": {"type": "string", "description": "完整问题或检索关键词"} + }, + required=["query"], + ) + + async def search_bound_knowledge(params: FunctionCallParams) -> None: + query = str(params.arguments.get("query") or "").strip() + if not query: + await params.result_callback( + {"status": "error", "message": "检索问题为空"} + ) + return + try: + async with SessionLocal() as session: + results = await search_knowledge( + session, + cfg.knowledge_base_id or "", + query, + top_k=int( + knowledge_config.get( + "top_n", knowledge_config.get("topN", 5) + ) + ), + score_threshold=float( + knowledge_config.get( + "score_threshold", + knowledge_config.get("scoreThreshold", 0.0), + ) + ), + ) + await params.result_callback({"status": "ok", "results": results}) + except Exception as exc: # noqa: BLE001 + logger.warning(f"Responses knowledge search failed: {exc}") + await params.result_callback( + {"status": "error", "message": "知识库检索暂时不可用"} + ) + + llm.register_function("search_knowledge_base", search_bound_knowledge) + + def set_tools(schemas=None) -> None: + visible = list(schemas or []) + if knowledge_schema is not None: + visible.append(knowledge_schema) + if visible: + context.set_tools(ToolsSchema(standard_tools=visible)) + else: + context.set_tools() + + await brain.setup( + cfg, + BrainRuntime( + context=context, + llm=llm, + queue_frame=worker.queue_frame, + set_system_prompt=set_system_prompt, + set_tools=set_tools, + call_end=call_end, + session_id=cfg.conversation_id, + client_tools=client_tools, + worker=worker, + context_aggregator=aggregators, + transport=None, + switch_services=switch_services, + set_knowledge_scope=knowledge.set_scope, + set_vision_scope=lambda scope: workflow_vision_scope.update(scope), + vision_function=None, + set_input_enabled=lambda _enabled: None, + apply_turn_config=lambda _enabled, _config: asyncio.sleep(0), + flow_global_functions=[], + ), + ) + greeting = str(await brain.greeting(cfg) or "").strip() + if greeting and brain.spec.owns_context: + brain.prepare_greeting_context(greeting, context) + + @assistant_aggregator.event_handler("on_assistant_turn_started") + async def on_assistant_turn_started(_aggregator): + capture.window.active_assistant += 1 + capture.window.touch() + await brain.on_assistant_text_start(new_id("turn")) + + @assistant_aggregator.event_handler("on_assistant_turn_stopped") + async def on_assistant_turn_stopped(_aggregator, message): + text = str(message.content or "").strip() + capture.window.active_assistant = max( + 0, capture.window.active_assistant - 1 + ) + if text: + capture.outputs.append(text) + capture.window.touch() + await brain.on_assistant_text_end( + new_id("turn"), + text, + bool(getattr(message, "interrupted", False)), + ) + if call_end.ending: + await call_end.finish() + + runner_task: asyncio.Task[None] | None = None + try: + await brain.run_preflight() + await runner.add_workers(worker) + runner_task = asyncio.create_task( + runner.run(), name=f"responses-{new_id('run')[:20]}" + ) + await asyncio.wait_for( + worker_started.wait(), timeout=WORKER_START_TIMEOUT_SECONDS + ) + await brain.on_connected(greeting_pending=False) + await brain.on_client_ready() + + boundary = asyncio.Event() + await worker.queue_frame(_BoundaryFrame(reached=boundary)) + await asyncio.wait_for(boundary.wait(), timeout=WORKER_START_TIMEOUT_SECONDS) + startup_outcome = await _wait_until_idle( + capture, + runner_task, + require_activity=False, + client_tools=client_tools, + ) + if startup_outcome == "function_call": + startup_text = ( + capture.outputs[-1] + if capture.outputs + else "".join(capture.deltas) + ).strip() + return ResponseRunResult( + text=startup_text, + function_calls=list(client_tools.calls), + context_messages=_context_snapshot( + context, + function_calls=client_tools.calls, + fallback_messages=current_messages, + ), + dynamic_variables=_public_variables(brain, cfg), + runtime_state=_runtime_state(brain), + ) + capture.reset() + + await worker.queue_frame( + LLMMessagesAppendFrame(messages=current_messages, run_llm=True) + ) + outcome = await _wait_until_idle( + capture, + runner_task, + require_activity=True, + client_tools=client_tools, + ) + if outcome == "function_call": + text = ( + capture.outputs[-1] + if capture.outputs + else "".join(capture.deltas) + ).strip() + return ResponseRunResult( + text=text, + function_calls=list(client_tools.calls), + context_messages=_context_snapshot( + context, + function_calls=client_tools.calls, + ), + dynamic_variables=_public_variables(brain, cfg), + runtime_state=_runtime_state(brain), + ) + text = (capture.outputs[-1] if capture.outputs else "".join(capture.deltas)).strip() + if not text: + raise ResponseRunError("Model returned no text") + return ResponseRunResult( + text=text, + function_calls=[], + context_messages=_context_snapshot(context, function_calls=[]), + dynamic_variables=_public_variables(brain, cfg), + runtime_state=_runtime_state(brain), + ) + finally: + if not worker.has_finished(): + await worker.cancel(reason="responses_turn_complete") + if runner_task is not None: + await asyncio.gather(runner_task, return_exceptions=True) diff --git a/backend/services/openai_responses/state.py b/backend/services/openai_responses/state.py new file mode 100644 index 0000000..e41eec8 --- /dev/null +++ b/backend/services/openai_responses/state.py @@ -0,0 +1,472 @@ +"""Persistence and context reconstruction for stateful Responses calls.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any + +from db.models import ConversationMessage, ConversationSession, OpenAIResponse +from db.session import SessionLocal +from models import AssistantConfig +from services.openai_realtime.events import assistant_id_from_model +from services.openai_responses.protocol import ( + ResponseRequest, + ResponsesAPIError, + new_id, +) +from services.realtime.launcher import resolve_assistant_config +from sqlalchemy import func, select, update + + +CONTINUABLE_STATUSES = {"completed"} + + +@dataclass(frozen=True) +class PreparedResponse: + row: OpenAIResponse + config: AssistantConfig + history_messages: list[dict[str, Any]] + current_messages: list[dict[str, Any]] + runtime_state: dict[str, Any] + client_tool_results: list[dict[str, Any]] + + +def _content_text(item: dict[str, Any]) -> str: + return "\n".join( + str(part.get("text") or "") + for part in item.get("content") or [] + if isinstance(part, dict) + ).strip() + + +def items_to_messages(items: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Convert the supported Responses item subset to OpenAI LLM messages.""" + messages: list[dict[str, Any]] = [] + pending_calls: list[dict[str, Any]] = [] + + def flush_calls() -> None: + if not pending_calls: + return + messages.append( + {"role": "assistant", "content": "", "tool_calls": list(pending_calls)} + ) + pending_calls.clear() + + for item in items: + item_type = item.get("type") + if item_type == "function_call": + pending_calls.append( + { + "id": str(item.get("call_id") or ""), + "type": "function", + "function": { + "name": str(item.get("name") or ""), + "arguments": str(item.get("arguments") or "{}"), + }, + } + ) + continue + flush_calls() + if item_type == "message": + messages.append( + { + "role": str(item.get("role") or "user"), + "content": _content_text(item), + } + ) + elif item_type == "function_call_output": + messages.append( + { + "role": "tool", + "tool_call_id": str(item.get("call_id") or ""), + "content": str(item.get("output") or ""), + } + ) + flush_calls() + return messages + + +async def _load_chain( + response_id: str, + *, + api_key_id: str, +) -> list[OpenAIResponse]: + chain: list[OpenAIResponse] = [] + seen: set[str] = set() + async with SessionLocal() as db: + current_id: str | None = response_id + while current_id: + if current_id in seen or len(chain) >= 1000: + raise ResponsesAPIError( + "Response chain is invalid", code="invalid_response_chain" + ) + seen.add(current_id) + row = await db.get(OpenAIResponse, current_id) + if row is None or row.api_key_id != api_key_id: + raise ResponsesAPIError( + "Response not found", code="response_not_found", status_code=404 + ) + chain.append(row) + current_id = row.previous_response_id + chain.reverse() + return chain + + +def _validate_tool_outputs( + chain: list[OpenAIResponse], + current_items: list[dict[str, Any]], +) -> None: + unresolved: dict[str, dict[str, Any]] = {} + for row in chain: + for item in row.output_items or []: + if item.get("type") == "function_call": + unresolved[str(item.get("call_id") or "")] = item + for item in row.input_items or []: + if item.get("type") == "function_call_output": + unresolved.pop(str(item.get("call_id") or ""), None) + + supplied = [ + str(item.get("call_id") or "") + for item in current_items + if item.get("type") == "function_call_output" + ] + if len(set(supplied)) != len(supplied): + raise ResponsesAPIError( + "A function call output can only be supplied once", + code="duplicate_tool_output", + param="input", + ) + unknown = sorted(set(supplied) - set(unresolved)) + if unknown: + raise ResponsesAPIError( + f"Unknown or already resolved call_id: {unknown[0]}", + code="invalid_tool_output", + param="input", + ) + if unresolved and set(supplied) != set(unresolved): + missing = sorted(set(unresolved) - set(supplied)) + raise ResponsesAPIError( + f"Missing function_call_output for call_id: {missing[0]}", + code="missing_tool_output", + param="input", + ) + + +def _client_tool_results( + chain: list[OpenAIResponse], + current_items: list[dict[str, Any]], +) -> list[dict[str, Any]]: + calls = { + str(item.get("call_id") or ""): item + for row in chain + for item in row.output_items or [] + if item.get("type") == "function_call" + } + results: list[dict[str, Any]] = [] + for item in current_items: + if item.get("type") != "function_call_output": + continue + call = calls.get(str(item.get("call_id") or "")) or {} + output = str(item.get("output") or "") + try: + value = json.loads(output) + except json.JSONDecodeError: + value = output + if not isinstance(value, dict) or "status" not in value: + value = {"status": "ok", "data": value} + results.append( + { + "call_id": item.get("call_id"), + "name": call.get("name"), + "result": value, + } + ) + return results + + +async def prepare_response( + request: ResponseRequest, + *, + api_key_id: str, +) -> PreparedResponse: + chain: list[OpenAIResponse] = [] + parent: OpenAIResponse | None = None + if request.previous_response_id: + chain = await _load_chain( + request.previous_response_id, + api_key_id=api_key_id, + ) + parent = chain[-1] + if parent.status not in CONTINUABLE_STATUSES: + raise ResponsesAPIError( + f"Previous response is {parent.status}", + code="response_not_ready", + param="previous_response_id", + status_code=409, + ) + if (parent.runtime_state or {}).get("status") == "ended": + raise ResponsesAPIError( + "The workflow conversation has ended", + code="conversation_ended", + param="previous_response_id", + status_code=409, + ) + if not parent.assistant_id: + raise ResponsesAPIError( + "The assistant used by the previous response no longer exists", + code="invalid_model", + param="previous_response_id", + ) + assistant_id = parent.assistant_id + if request.model: + requested_id = assistant_id_from_model(request.model) + if requested_id != assistant_id: + raise ResponsesAPIError( + "model must match the previous response", + code="invalid_model", + param="model", + ) + conversation_id = parent.conversation_id + variables = dict(parent.dynamic_variables or {}) + variables.update(request.dynamic_variables) + _validate_tool_outputs(chain, request.input_items) + client_tool_results = _client_tool_results(chain, request.input_items) + runtime_state = dict(parent.runtime_state or {}) + else: + try: + assistant_id = assistant_id_from_model(request.model) + except ValueError as exc: + raise ResponsesAPIError( + str(exc), code="invalid_model", param="model" + ) from exc + conversation_id = new_id("conv")[:40] + variables = dict(request.dynamic_variables) + client_tool_results = [] + runtime_state = {} + if any( + item.get("type") == "function_call_output" + for item in request.input_items + ): + raise ResponsesAPIError( + "function_call_output requires previous_response_id", + code="invalid_tool_output", + param="input", + ) + + try: + config = await resolve_assistant_config( + assistant_id, + dynamic_variables=variables, + conversation_id=conversation_id, + ) + except ValueError as exc: + raise ResponsesAPIError( + str(exc), code="invalid_model", param="model" + ) from exc + if config.runtimeMode != "pipeline": + raise ResponsesAPIError( + "Responses API only supports pipeline assistants", + code="unsupported_runtime_mode", + param="model", + ) + if config.type not in {"prompt", "workflow", "dify", "fastgpt"}: + raise ResponsesAPIError( + f"Assistant type {config.type} is not supported by Responses API", + code="unsupported_assistant_type", + param="model", + ) + if config.conversation_id != conversation_id: + config = config.model_copy(update={"conversation_id": conversation_id}) + effective_variables = { + name: value + for name, value in (config.dynamic_variables or {}).items() + if not name.startswith(("system__", "secret__")) + } + + history_items: list[dict[str, Any]] = [] + for item in chain: + history_items.extend(item.input_items or []) + history_items.extend(item.output_items or []) + history_messages = ( + list(parent.context_messages or []) + if parent is not None and parent.context_messages + else items_to_messages(history_items) + ) + + row = OpenAIResponse( + id=new_id("resp")[:40], + api_key_id=api_key_id, + assistant_id=assistant_id, + conversation_id=conversation_id, + previous_response_id=request.previous_response_id, + status="in_progress", + input_items=request.input_items, + output_items=[], + context_messages=[], + runtime_state={}, + dynamic_variables=effective_variables, + ) + async with SessionLocal() as db: + if parent is None: + db.add( + ConversationSession( + id=conversation_id, + assistant_id=assistant_id, + assistant_name=config.name, + channel="responses", + runtime_mode="pipeline", + status="active", + message_count=0, + analysis_status="none", + analysis_data=( + {"plan": config.analysis_config} + if config.analysis_config.get("enabled") + else {} + ), + extra={}, + ) + ) + else: + conversation = await db.get(ConversationSession, conversation_id) + if conversation: + conversation.status = "active" + conversation.ended_at = None + db.add(row) + await db.commit() + await db.refresh(row) + + return PreparedResponse( + row=row, + config=config, + history_messages=history_messages, + current_messages=items_to_messages(request.input_items), + runtime_state=runtime_state, + client_tool_results=client_tool_results, + ) + + +async def get_response( + response_id: str, + *, + api_key_id: str, +) -> OpenAIResponse: + async with SessionLocal() as db: + row = await db.get(OpenAIResponse, response_id) + if row is None or row.api_key_id != api_key_id: + raise ResponsesAPIError( + "Response not found", code="response_not_found", status_code=404 + ) + return row + + +async def finish_response( + response_id: str, + *, + status: str, + output_items: list[dict[str, Any]], + context_messages: list[dict[str, Any]] | None = None, + dynamic_variables: dict[str, Any] | None = None, + runtime_state: dict[str, Any] | None = None, + error: dict[str, Any] | None = None, +) -> OpenAIResponse: + now = datetime.now(UTC) + async with SessionLocal() as db: + row = await db.get(OpenAIResponse, response_id) + if row is None: + raise RuntimeError(f"Response disappeared while running: {response_id}") + row.status = status + row.output_items = output_items + row.context_messages = context_messages or [] + if dynamic_variables is not None: + row.dynamic_variables = dynamic_variables + if runtime_state is not None: + row.runtime_state = runtime_state + row.error = error + row.completed_at = now + + conversation = ( + await db.execute( + select(ConversationSession) + .where(ConversationSession.id == row.conversation_id) + .with_for_update() + ) + ).scalar_one_or_none() + if conversation: + max_sequence = await db.scalar( + select(func.max(ConversationMessage.sequence)).where( + ConversationMessage.session_id == row.conversation_id + ) + ) + sequence = int(max_sequence or 0) + for item in row.input_items or []: + if item.get("type") != "message" or item.get("role") != "user": + continue + text = _content_text(item) + if not text: + continue + sequence += 1 + db.add( + ConversationMessage( + id=new_id("msg")[:40], + session_id=row.conversation_id, + sequence=sequence, + role="user", + content_type="text", + content=text, + extra={"response_id": row.id}, + ) + ) + for item in output_items: + if item.get("type") == "message": + text = _content_text(item) + elif item.get("type") == "function_call": + text = f"[function_call] {item.get('name') or ''}" + else: + continue + if not text: + continue + sequence += 1 + db.add( + ConversationMessage( + id=new_id("msg")[:40], + session_id=row.conversation_id, + sequence=sequence, + role="assistant", + content_type="text", + content=text, + extra={"response_id": row.id, "item": item}, + ) + ) + conversation.message_count = sequence + conversation.status = "completed" if status == "completed" else status + conversation.ended_at = now + await db.commit() + await db.refresh(row) + return row +async def recover_interrupted_responses() -> None: + """Mark generations interrupted by a process restart as cancelled.""" + now = datetime.now(UTC) + async with SessionLocal() as db: + conversation_ids = select(OpenAIResponse.conversation_id).where( + OpenAIResponse.status == "in_progress" + ) + await db.execute( + update(ConversationSession) + .where(ConversationSession.id.in_(conversation_ids)) + .values(status="cancelled", ended_at=now) + ) + await db.execute( + update(OpenAIResponse) + .where(OpenAIResponse.status == "in_progress") + .values( + status="cancelled", + error={ + "code": "server_restarted", + "message": "Response was interrupted by a server restart", + }, + completed_at=now, + ) + ) + await db.commit() diff --git a/backend/services/realtime/launcher.py b/backend/services/realtime/launcher.py index ef67184..8733d47 100644 --- a/backend/services/realtime/launcher.py +++ b/backend/services/realtime/launcher.py @@ -17,6 +17,7 @@ async def resolve_assistant_config( assistant_id: str, *, dynamic_variables: dict[str, Any] | None = None, + conversation_id: str | None = None, ) -> AssistantConfig: async with SessionLocal() as session: config = await resolve_runtime_config(session, assistant_id) @@ -24,6 +25,7 @@ async def resolve_assistant_config( config, dynamic_variables or {}, assistant_id=assistant_id, + conversation_id=conversation_id, ) diff --git a/backend/services/runtime_variables.py b/backend/services/runtime_variables.py index 0382939..c6a7239 100644 --- a/backend/services/runtime_variables.py +++ b/backend/services/runtime_variables.py @@ -256,6 +256,7 @@ def prepare_dynamic_config( client_values: dict[str, Any] | None, *, assistant_id: str | None, + conversation_id: str | None = None, timezone: str = "Asia/Shanghai", trusted_values: dict[str, Any] | None = None, ) -> AssistantConfig: @@ -288,7 +289,7 @@ def prepare_dynamic_config( merged[name] = value merged.update(supplied) - conversation_id = f"conv_{uuid4().hex[:20]}" + conversation_id = conversation_id or f"conv_{uuid4().hex[:20]}" merged.update( _system_values( assistant_id=assistant_id, diff --git a/backend/settings.py b/backend/settings.py index 668f1a2..8e197b9 100644 --- a/backend/settings.py +++ b/backend/settings.py @@ -45,6 +45,9 @@ REALTIME_TOKEN_SECRET = os.getenv("REALTIME_TOKEN_SECRET", AUTH_SECRET_KEY) REALTIME_CLIENT_SECRET_TTL_SECONDS = int( os.getenv("REALTIME_CLIENT_SECRET_TTL_SECONDS", "60") ) +RESPONSES_REQUEST_TIMEOUT_SECONDS = int( + os.getenv("RESPONSES_REQUEST_TIMEOUT_SECONDS", "120") +) # ---- WebRTC STUN / TURN ---- # Override STUN_URL in remote deployments to use the colocated coturn server diff --git a/backend/tests/test_brains.py b/backend/tests/test_brains.py index f72db1c..008d1f2 100644 --- a/backend/tests/test_brains.py +++ b/backend/tests/test_brains.py @@ -315,6 +315,35 @@ class DifyLLMServiceTests(unittest.IsolatedAsyncioTestCase): class PromptBrainTests(unittest.IsolatedAsyncioTestCase): + async def test_responses_state_skips_completed_prompt_startup(self): + cfg = AssistantConfig(type="prompt") + brain = build_brain(cfg) + brain.restore_text_state( + { + "type": "prompt", + "preflight_finished": True, + "opening_finished": True, + "entry_dispatched": True, + } + ) + + await brain.setup( + cfg, + BrainRuntime( + context=LLMContext(messages=[]), + llm=FakeLLM(), + queue_frame=noop_queue_frame, + set_system_prompt=lambda _prompt: None, + set_tools=lambda _tools: None, + call_end=FakeCallEnd(), + ), + ) + + self.assertTrue(brain._preflight_finished) + self.assertTrue(brain._opening_finished) + self.assertTrue(brain._entry_dispatched) + self.assertEqual(brain.text_state_snapshot()["type"], "prompt") + async def test_startup_only_tool_is_not_registered_with_llm(self): startup_tool = RuntimeTool( id="opening_message", diff --git a/backend/tests/test_openai_responses.py b/backend/tests/test_openai_responses.py new file mode 100644 index 0000000..d2daaad --- /dev/null +++ b/backend/tests/test_openai_responses.py @@ -0,0 +1,355 @@ +from __future__ import annotations + +import asyncio +import unittest +from datetime import UTC, datetime +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +from models import AssistantConfig +from pipecat.frames.frames import ( + LLMContextFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMTextFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from services.brains.base import BaseBrain, BrainSpec +from services.openai_responses.protocol import ResponsesAPIError, parse_request +from services.openai_responses.runner import ( + ResponsesClientToolPort, + ResponsesTextRunner, + _context_snapshot, +) +from services.openai_responses.state import _validate_tool_outputs, items_to_messages +from routes.openai_responses import _stream_response + + +class ResponsesProtocolTests(unittest.TestCase): + def test_first_turn_requires_model_and_normalizes_text(self): + request = parse_request( + { + "model": "assistant:asst_1", + "input": "你好", + "x_interactive_media": {"variables": {"customer": "王女士"}}, + } + ) + + self.assertEqual(request.model, "assistant:asst_1") + self.assertFalse(request.stream) + self.assertEqual(request.dynamic_variables, {"customer": "王女士"}) + self.assertEqual(request.input_items[0]["role"], "user") + self.assertEqual(request.input_items[0]["content"][0]["text"], "你好") + + def test_continuation_can_inherit_model(self): + request = parse_request( + { + "previous_response_id": "resp_parent", + "input": [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": {"ok": True}, + } + ], + "stream": True, + } + ) + + self.assertIsNone(request.model) + self.assertTrue(request.stream) + self.assertEqual(request.input_items[0]["output"], '{"ok": true}') + + def test_rejects_stateless_and_assistant_owned_options(self): + with self.assertRaisesRegex(ResponsesAPIError, "store must be true"): + parse_request( + {"model": "assistant:asst_1", "input": "hi", "store": False} + ) + with self.assertRaisesRegex(ResponsesAPIError, "store must be true"): + parse_request( + {"model": "assistant:asst_1", "input": "hi", "store": None} + ) + with self.assertRaisesRegex(ResponsesAPIError, "owned"): + parse_request( + { + "model": "assistant:asst_1", + "input": "hi", + "instructions": "override", + } + ) + + def test_rejects_unknown_top_level_parameters(self): + with self.assertRaises(ResponsesAPIError) as raised: + parse_request( + {"model": "assistant:asst_1", "input": "hi", "metadata": {}} + ) + + self.assertEqual(raised.exception.code, "unsupported_parameter") + self.assertEqual(raised.exception.param, "metadata") + + def test_rejects_images_and_unknown_items(self): + with self.assertRaises(ResponsesAPIError) as raised: + parse_request( + { + "model": "assistant:asst_1", + "input": [ + { + "type": "message", + "role": "user", + "content": [{"type": "input_image", "image_url": "x"}], + } + ], + } + ) + self.assertEqual(raised.exception.code, "invalid_request_error") + + +class ResponsesStateTests(unittest.TestCase): + def test_context_items_become_llm_messages(self): + messages = items_to_messages( + [ + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "查订单"}], + }, + { + "type": "function_call", + "call_id": "call_1", + "name": "lookup_order", + "arguments": '{"id":"A1"}', + }, + { + "type": "function_call_output", + "call_id": "call_1", + "output": '{"status":"ok"}', + }, + ] + ) + + self.assertEqual(messages[0], {"role": "user", "content": "查订单"}) + self.assertEqual(messages[1]["tool_calls"][0]["id"], "call_1") + self.assertEqual(messages[2]["role"], "tool") + + def test_tool_output_must_resolve_parent_pending_call(self): + parent = SimpleNamespace( + output_items=[{"type": "function_call", "call_id": "call_1"}], + input_items=[], + ) + _validate_tool_outputs( + [parent], + [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": "ok", + } + ], + ) + + with self.assertRaises(ResponsesAPIError) as raised: + _validate_tool_outputs( + [parent], + [ + { + "type": "function_call_output", + "call_id": "call_other", + "output": "ok", + } + ], + ) + self.assertEqual(raised.exception.code, "invalid_tool_output") + + def test_context_snapshot_aligns_provider_and_public_call_ids(self): + context = SimpleNamespace( + get_messages=lambda: [ + { + "role": "assistant", + "tool_calls": [ + { + "id": "provider_call", + "type": "function", + "function": {"name": "show_message", "arguments": "{}"}, + } + ], + } + ] + ) + snapshot = _context_snapshot( + context, + function_calls=[ + { + "call_id": "call_public", + "name": "show_message", + "arguments": "{}", + } + ], + ) + + self.assertEqual(snapshot[0]["tool_calls"][0]["id"], "call_public") + + def test_tool_output_cannot_be_replayed(self): + resolved = SimpleNamespace( + output_items=[{"type": "function_call", "call_id": "call_1"}], + input_items=[ + { + "type": "function_call_output", + "call_id": "call_1", + "output": "ok", + } + ], + ) + with self.assertRaises(ResponsesAPIError) as raised: + _validate_tool_outputs( + [resolved], + [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": "again", + } + ], + ) + self.assertEqual(raised.exception.code, "invalid_tool_output") + + +class ResponsesClientToolTests(unittest.IsolatedAsyncioTestCase): + async def test_client_tool_is_exposed_without_waiting_for_browser(self): + port = ResponsesClientToolPort() + task = asyncio.create_task( + port.call("show_message", {"message": "确认"}, timeout_seconds=3) + ) + await port.called.wait() + + self.assertEqual(port.calls[0]["type"], "function_call") + self.assertEqual(port.calls[0]["name"], "show_message") + self.assertIn('"message": "确认"', port.calls[0]["arguments"]) + + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + async def test_persisted_result_resumes_deterministic_client_tool(self): + port = ResponsesClientToolPort( + [ + { + "name": "show_message", + "result": {"status": "ok", "data": {"action": "confirmed"}}, + } + ] + ) + + result = await port.call( + "show_message", + {"message": "确认"}, + timeout_seconds=3, + ) + + self.assertEqual(result["data"]["action"], "confirmed") + self.assertFalse(port.called.is_set()) + + +class _FakeTextLLM(FrameProcessor): + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if not isinstance(frame, LLMContextFrame): + await self.push_frame(frame, direction) + return + await self.push_frame(LLMFullResponseStartFrame()) + await self.push_frame(LLMTextFrame("状态已恢复")) + await self.push_frame(LLMFullResponseEndFrame()) + + +class _FakeBrain(BaseBrain): + spec = BrainSpec( + type="dify", + supported_runtime_modes=frozenset({"pipeline"}), + owns_context=False, + ) + + def build_llm(self, _cfg, _context): + return _FakeTextLLM() + + +class ResponsesTextRunnerTests(unittest.IsolatedAsyncioTestCase): + async def test_transportless_runner_streams_and_finishes_text(self): + deltas: list[str] = [] + + async def on_delta(value: str) -> None: + deltas.append(value) + + with patch( + "services.openai_responses.runner.build_brain", + return_value=_FakeBrain(), + ): + result = await asyncio.wait_for( + ResponsesTextRunner().run( + AssistantConfig( + type="dify", + runtimeMode="pipeline", + greeting="", + ), + history_messages=[{"role": "user", "content": "上一轮"}], + current_messages=[{"role": "user", "content": "继续"}], + on_delta=on_delta, + ), + timeout=3, + ) + + self.assertEqual(result.text, "状态已恢复") + self.assertEqual(deltas, ["状态已恢复"]) + + +class ResponsesStreamingTests(unittest.IsolatedAsyncioTestCase): + async def test_stream_uses_responses_event_lifecycle(self): + now = datetime.now(UTC) + row = SimpleNamespace( + id="resp_1", + api_key_id="rtkey_1", + assistant_id="asst_1", + previous_response_id=None, + status="completed", + error=None, + output_items=[ + { + "id": "msg_1", + "type": "message", + "status": "completed", + "role": "assistant", + "content": [ + { + "type": "output_text", + "text": "完成", + "annotations": [], + } + ], + } + ], + created_at=now, + completed_at=now, + ) + prepared = SimpleNamespace(row=row) + + async def run_prepared(_prepared, *, on_delta, text_item_id): + self.assertEqual(text_item_id, "msg_1") + await on_delta("完") + await on_delta("成") + return row + + with patch( + "routes.openai_responses._run_prepared", + new=AsyncMock(side_effect=run_prepared), + ): + chunks = [ + chunk.decode("utf-8") + async for chunk in _stream_response(prepared, text_item_id="msg_1") + ] + + body = "".join(chunks) + self.assertIn("event: response.created", body) + self.assertIn("event: response.output_text.delta", body) + self.assertIn("event: response.output_text.done", body) + self.assertIn("event: response.completed", body) + + +if __name__ == "__main__": + unittest.main() diff --git a/docs-developer/docs.json b/docs-developer/docs.json index 18107c7..c9a823c 100644 --- a/docs-developer/docs.json +++ b/docs-developer/docs.json @@ -46,7 +46,7 @@ "groups": [ { "group": "开始与对接", - "pages": ["index", "integrations", "tool-resources", "realtime-protocol"] + "pages": ["index", "integrations", "tool-resources", "responses-api", "realtime-protocol"] } ] } diff --git a/docs-developer/index.mdx b/docs-developer/index.mdx index 3595105..fe720ae 100644 --- a/docs-developer/index.mdx +++ b/docs-developer/index.mdx @@ -40,8 +40,9 @@ make db-seed 保存后，从助手详情页打开语音预览即可测试。 -## 4. 对接 Realtime 协议 +## 4. 对接公开协议 +- [Responses 文本接口](/responses-api)：通过 HTTP 或 SSE 调用 Pipeline 助手，并使用 `previous_response_id` 延续多轮状态。 - [Realtime 兼容协议设计](/realtime-protocol)：了解后端提供的 OpenAI Realtime 兼容 WebRTC、WebSocket、鉴权与 Interactive Media 扩展协议。 diff --git a/docs-developer/responses-api.mdx b/docs-developer/responses-api.mdx new file mode 100644 index 0000000..49f6ebe --- /dev/null +++ b/docs-developer/responses-api.mdx @@ -0,0 +1,135 @@ +--- +title: Stateful Responses 文本接口 +description: 使用 OpenAI Responses 风格的 HTTP 与 SSE 接口调用平台助手。 +--- + +# Stateful Responses 文本接口 + +`POST /v1/responses` 为 Pipeline 助手提供文本调用。服务端保存每轮输入和输出，后续请求通过 `previous_response_id` 恢复上下文。语音、视频和长连接交互继续使用 `/v1/realtime`。 + +接口使用后台创建的长期 `sk-rt-...` API Key。短期 `ek-rt-...` Client Secret 不能调用文本接口。 + +## 首轮请求 + +```bash +curl -X POST "https://api.example.com/v1/responses" \ + -H "Authorization: Bearer $REALTIME_API_KEY" \ + -H "Content-Type: application/json" \ + -d '{ + "model": "assistant:asst_xxx", + "input": "请介绍一下这个产品" + }' +``` + +响应中的 `id` 是下一轮使用的状态指针： + +```json +{ + "id": "resp_xxx", + "object": "response", + "status": "completed", + "model": "assistant:asst_xxx", + "previous_response_id": null, + "output": [ + { + "id": "msg_xxx", + "type": "message", + "status": "completed", + "role": "assistant", + "content": [ + {"type": "output_text", "text": "……", "annotations": []} + ] + } + ] +} +``` + +## 继续多轮对话 + +后续请求可以省略 `model`。服务端从父响应恢复助手、动态变量、消息和工具上下文： + +```bash +curl -X POST "https://api.example.com/v1/responses" \ + -H "Authorization: Bearer $REALTIME_API_KEY" \ + -H "Content-Type: application/json" \ + -d '{ + "previous_response_id": "resp_xxx", + "input": "它适合哪些行业？" + }' +``` + +同一个父响应可以创建多个独立分支。Response 只能由创建它的 API Key 查询或继续： + +```bash +curl "https://api.example.com/v1/responses/resp_xxx" \ + -H "Authorization: Bearer $REALTIME_API_KEY" +``` + +## SSE 流式输出 + +设置 `stream: true` 后返回 `text/event-stream`： + +```json +{ + "model": "assistant:asst_xxx", + "input": "给我一个简短介绍", + "stream": true +} +``` + +主要事件依次为 `response.created`、`response.in_progress`、`response.output_text.delta`、`response.output_text.done` 和 `response.completed`。连接中断会取消运行，并把 Response 标记为 `cancelled`。 + +## 动态变量 + +首轮可以传入项目扩展字段，后续请求会继承并可增量覆盖： + +```json +{ + "model": "assistant:asst_xxx", + "input": "查询我的权益", + "x_interactive_media": { + "variables": { + "customer_name": "王女士", + "membership_level": "gold" + } + } +} +``` + +## Client Tool 续传 + +当助手需要调用客户端工具时，Response 的 `output` 会包含 `function_call`： + +```json +{ + "type": "function_call", + "call_id": "call_xxx", + "name": "show_message", + "arguments": "{\"message\":\"请确认\"}" +} +``` + +客户端执行后，通过父响应和对应的 `call_id` 继续： + +```json +{ + "previous_response_id": "resp_xxx", + "input": [ + { + "type": "function_call_output", + "call_id": "call_xxx", + "output": "{\"status\":\"ok\",\"data\":{\"action\":\"confirmed\"}}" + } + ] +} +``` + +如果父链存在多个尚未完成的 Client Tool，本轮必须一次提交全部对应结果。 + +## 首版边界 + +- 仅支持 `runtimeMode=pipeline` 的 Prompt、Workflow、Dify 和 FastGPT 助手。 +- 只支持文本 `message` 和 `function_call_output` 输入。 +- `store` 必须为 `true` 或省略。 +- 不支持 `conversation`、`background`、图片、文件以及调用方覆盖 instructions 或 tools。 +- 不提供 `/v1/chat/completions` 兼容接口。