mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-30 11:54:50 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
844 lines
27 KiB
Python
844 lines
27 KiB
Python
"""
|
||
提供 OpenAI Models、Chat Completions 与 Responses 兼容接口。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import time
|
||
import uuid
|
||
from threading import Lock
|
||
from typing import TYPE_CHECKING, Any, AsyncIterator, List, Optional, Tuple, cast
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Request, Security
|
||
from fastapi.responses import JSONResponse, StreamingResponse
|
||
from fastapi.security import HTTPAuthorizationCredentials
|
||
|
||
from app.adapters.web.security.access import (
|
||
openai_bearer_scheme,
|
||
validate_api_credential_identity,
|
||
)
|
||
from app.agent.contracts import ReplyMode
|
||
from app.agent.loader import get_moviepilot_agent_type
|
||
from app.api.context import (
|
||
get_background_task_registry_compat,
|
||
resolve_background_task_registry,
|
||
)
|
||
from app.api.presentation.sse import build_sse_response, encode_data_event
|
||
from app.api.protocol import (
|
||
build_completion_payload,
|
||
build_prompt,
|
||
build_responses_input,
|
||
build_session_id,
|
||
)
|
||
from app.application.agent import get_running_agent_manager
|
||
from app.application.configuration import get_api_runtime_config_snapshot
|
||
from app.runtime.execution import run_in_threadpool
|
||
from app.runtime.tasks import TaskRegistry
|
||
from app.schemas.openai import OpenAIChatCompletionResponse as _SchemaOpenAIChatCompletionResponse
|
||
from app.schemas.openai import OpenAIChatCompletionsRequest as _SchemaOpenAIChatCompletionsRequest
|
||
from app.schemas.openai import OpenAIErrorDetail as _SchemaOpenAIErrorDetail
|
||
from app.schemas.openai import OpenAIErrorResponse as _SchemaOpenAIErrorResponse
|
||
from app.schemas.openai import OpenAIModelInfo as _SchemaOpenAIModelInfo
|
||
from app.schemas.openai import OpenAIModelListResponse as _SchemaOpenAIModelListResponse
|
||
from app.schemas.openai import OpenAIResponsesOutputMessage as _SchemaOpenAIResponsesOutputMessage
|
||
from app.schemas.openai import OpenAIResponsesOutputText as _SchemaOpenAIResponsesOutputText
|
||
from app.schemas.openai import OpenAIResponsesRequest as _SchemaOpenAIResponsesRequest
|
||
from app.schemas.openai import OpenAIResponsesResponse as _SchemaOpenAIResponsesResponse
|
||
from app.schemas.openai import OpenAIUsage as _SchemaOpenAIUsage
|
||
from app.schemas.types import NotificationChannel
|
||
|
||
if TYPE_CHECKING:
|
||
from app.agent.graph import AgentGraphCache
|
||
from app.agent.manager import AgentManager
|
||
from app.agent.orchestrator import MoviePilotAgent
|
||
|
||
OPENAI_ERROR_RESPONSES: dict[int | str, dict[str, object]] = {
|
||
400: {"model": _SchemaOpenAIErrorResponse, "description": "请求格式错误"},
|
||
401: {"model": _SchemaOpenAIErrorResponse, "description": "认证失败"},
|
||
422: {"model": _SchemaOpenAIErrorResponse, "description": "请求参数校验失败"},
|
||
500: {"model": _SchemaOpenAIErrorResponse, "description": "服务内部错误"},
|
||
503: {"model": _SchemaOpenAIErrorResponse, "description": "AI Agent 不可用"},
|
||
}
|
||
|
||
router = APIRouter(responses=OPENAI_ERROR_RESPONSES)
|
||
|
||
MODEL_ID = "moviepilot-agent"
|
||
SESSION_PREFIX = "openai:"
|
||
OpenAIStreamEvent = str | dict[str, str] | None
|
||
|
||
|
||
class _CollectingMoviePilotAgentMixin:
|
||
"""
|
||
捕获 Agent 最终输出，避免再通过消息渠道二次发送。
|
||
|
||
:ivar _graphs: 运行时 Agent 基类提供的会话图资源 owner
|
||
"""
|
||
|
||
_graphs: AgentGraphCache
|
||
|
||
def __init__(self, *args: Any, stream_mode: bool = False, **kwargs: Any) -> None:
|
||
"""
|
||
初始化协议 Agent 的消息收集与流式模式。
|
||
"""
|
||
super().__init__(*args, **kwargs)
|
||
self.collected_messages: List[str] = []
|
||
self.stream_mode = stream_mode
|
||
if stream_mode:
|
||
self.stream_handler = _get_openai_streaming_handler_type()()
|
||
|
||
def _should_stream(self) -> bool:
|
||
"""
|
||
返回当前请求是否启用流式输出。
|
||
"""
|
||
return self.stream_mode
|
||
|
||
def configure_protocol_request(
|
||
self,
|
||
*,
|
||
stream_mode: bool,
|
||
event_queue: Optional[asyncio.Queue[OpenAIStreamEvent]],
|
||
) -> None:
|
||
"""
|
||
切换请求级输出目标，并保持已编译工具引用的 handler identity。
|
||
"""
|
||
self.collected_messages = []
|
||
self.stream_mode = stream_mode
|
||
if isinstance(self.stream_handler, _OpenAIStreamingHandlerMixin):
|
||
self.stream_handler.bind_queue(event_queue if stream_mode else None)
|
||
return
|
||
if not stream_mode:
|
||
return
|
||
self.stream_handler = _get_openai_streaming_handler_type()()
|
||
self.stream_handler.bind_queue(event_queue)
|
||
# 已编译工具持有旧 handler；identity 变化时必须重建图和工具目录。
|
||
self._graphs.retire()
|
||
|
||
def release_protocol_request(
|
||
self,
|
||
event_queue: Optional[asyncio.Queue[OpenAIStreamEvent]],
|
||
) -> None:
|
||
"""
|
||
释放已结束请求的输出队列，不影响同会话已重绑的新请求。
|
||
"""
|
||
if isinstance(self.stream_handler, _OpenAIStreamingHandlerMixin):
|
||
self.stream_handler.unbind_queue(event_queue)
|
||
|
||
async def send_agent_message(self, message: str, title: str = "") -> None:
|
||
"""
|
||
收集 Agent 消息并按请求模式输出。
|
||
"""
|
||
text = (message or "").strip()
|
||
if title and text:
|
||
text = f"{title}\n{text}"
|
||
elif title:
|
||
text = title.strip()
|
||
if text:
|
||
self.collected_messages.append(text)
|
||
if self.stream_mode:
|
||
self.stream_handler.emit(text)
|
||
|
||
|
||
class _OpenAIStreamingHandlerMixin:
|
||
"""
|
||
将 Agent 流式输出转发到 OpenAI SSE 队列，不向站内消息系统落消息。
|
||
"""
|
||
|
||
_lock: Lock
|
||
_buffer: str
|
||
|
||
def __init__(self) -> None:
|
||
"""
|
||
初始化流式事件队列绑定。
|
||
"""
|
||
super().__init__()
|
||
self._event_queue: Optional[asyncio.Queue[OpenAIStreamEvent]] = None
|
||
|
||
def bind_queue(self, queue: Optional[asyncio.Queue[OpenAIStreamEvent]]) -> None:
|
||
"""
|
||
绑定当前协议请求的输出队列。
|
||
"""
|
||
self._event_queue = queue
|
||
|
||
def unbind_queue(self, queue: Optional[asyncio.Queue[OpenAIStreamEvent]]) -> None:
|
||
"""
|
||
仅当仍指向该请求时解除绑定，避免清掉已排队的新请求。
|
||
"""
|
||
if self._event_queue is queue:
|
||
self._event_queue = None
|
||
|
||
def emit(self, token: str) -> str:
|
||
"""
|
||
发送已整理的流式文本到请求事件队列。
|
||
"""
|
||
emit_base = getattr(super(), "emit", None)
|
||
emitted = str(emit_base(token)) if callable(emit_base) else token
|
||
if emitted and self._event_queue is not None:
|
||
self._event_queue.put_nowait(emitted)
|
||
return emitted
|
||
|
||
def emit_tool_message(self, message: str) -> str:
|
||
"""
|
||
将工具提示转发到 OpenAI SSE 队列，同时保留流式缓冲内容。
|
||
"""
|
||
# 沿用动态流式缓冲基类的方法调用，将返回值统一转换为文本。
|
||
emit_base = getattr(super(), "emit_tool_message", None)
|
||
emitted = str(emit_base(message)) if callable(emit_base) else message
|
||
if emitted and self._event_queue is not None:
|
||
self._event_queue.put_nowait(emitted)
|
||
return emitted
|
||
|
||
def flush_pending_tool_summary(self) -> str:
|
||
"""
|
||
结算待发送的工具摘要并写入事件队列。
|
||
"""
|
||
flush_base = getattr(super(), "flush_pending_tool_summary", None)
|
||
emitted = str(flush_base()) if callable(flush_base) else ""
|
||
if emitted and self._event_queue is not None:
|
||
self._event_queue.put_nowait(emitted)
|
||
return str(emitted)
|
||
|
||
async def start_streaming(
|
||
self,
|
||
channel: Optional[str] = None,
|
||
source: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
username: Optional[str] = None,
|
||
original_message_id: Optional[str] = None,
|
||
original_chat_id: Optional[str] = None,
|
||
title: str = "",
|
||
) -> None:
|
||
"""
|
||
绑定当前请求的渠道信息并启用流式输出。
|
||
"""
|
||
self._channel = channel
|
||
self._source = source
|
||
self._user_id = user_id
|
||
self._username = username
|
||
self._original_message_id = original_message_id
|
||
self._original_chat_id = original_chat_id
|
||
self._title = title
|
||
self._streaming_enabled = True
|
||
self._sent_text = ""
|
||
self._message_response = None
|
||
self._msg_start_offset = 0
|
||
self._max_message_length = 0
|
||
|
||
async def stop_streaming(self) -> Tuple[bool, str]:
|
||
"""
|
||
结束流式输出并返回最终文本。
|
||
"""
|
||
if not self._streaming_enabled:
|
||
return False, ""
|
||
self._streaming_enabled = False
|
||
with self._lock:
|
||
final_text = self._buffer
|
||
self._buffer = ""
|
||
self._sent_text = ""
|
||
self._message_response = None
|
||
self._msg_start_offset = 0
|
||
return True, final_text
|
||
|
||
|
||
def _get_openai_streaming_handler_type() -> type:
|
||
"""
|
||
首次兼容协议调用时才解析完整流式处理器。
|
||
"""
|
||
global _OPENAI_STREAMING_HANDLER_TYPE
|
||
if _OPENAI_STREAMING_HANDLER_TYPE is not None:
|
||
return _OPENAI_STREAMING_HANDLER_TYPE
|
||
with _OPENAI_STREAMING_HANDLER_TYPE_LOCK:
|
||
if _OPENAI_STREAMING_HANDLER_TYPE is None:
|
||
from app.agent.callback import StreamingHandler
|
||
|
||
_OPENAI_STREAMING_HANDLER_TYPE = type(
|
||
"_RuntimeOpenAIStreamingHandler",
|
||
(_OpenAIStreamingHandlerMixin, StreamingHandler),
|
||
{"__module__": __name__},
|
||
)
|
||
return _OPENAI_STREAMING_HANDLER_TYPE
|
||
|
||
|
||
_OPENAI_STREAMING_HANDLER_TYPE_LOCK = Lock()
|
||
_OPENAI_STREAMING_HANDLER_TYPE: Optional[type] = None
|
||
|
||
|
||
def _build_collecting_agent_type(agent_base_type: type) -> type:
|
||
"""
|
||
为 OpenAI 与 Anthropic 兼容协议组合唯一的运行时类型。
|
||
"""
|
||
return type(
|
||
"_RuntimeCollectingMoviePilotAgent",
|
||
(_CollectingMoviePilotAgentMixin, agent_base_type),
|
||
{"__module__": __name__},
|
||
)
|
||
|
||
|
||
_COLLECTING_AGENT_TYPE_LOCK = Lock()
|
||
_COLLECTING_AGENT_TYPE: Optional[type] = None
|
||
|
||
|
||
def _get_collecting_agent_type() -> type:
|
||
"""
|
||
在首个真实兼容协议请求边界 single-flight 解析 Agent 类型。
|
||
"""
|
||
global _COLLECTING_AGENT_TYPE
|
||
if _COLLECTING_AGENT_TYPE is not None:
|
||
return _COLLECTING_AGENT_TYPE
|
||
with _COLLECTING_AGENT_TYPE_LOCK:
|
||
if _COLLECTING_AGENT_TYPE is None:
|
||
_COLLECTING_AGENT_TYPE = _build_collecting_agent_type(
|
||
get_moviepilot_agent_type()
|
||
)
|
||
return _COLLECTING_AGENT_TYPE
|
||
|
||
|
||
def _sse_payload(data: dict[str, object]) -> str:
|
||
"""
|
||
保留旧测试入口并委托独立的 OpenAI SSE 协议转换器。
|
||
"""
|
||
return encode_data_event(data)
|
||
|
||
|
||
async def _stream_response(
|
||
manager: AgentManager,
|
||
session_id: str,
|
||
user_id: str,
|
||
username: str,
|
||
prompt: str,
|
||
images: List[str],
|
||
cleanup_session: bool,
|
||
task_registry: TaskRegistry | None = None,
|
||
) -> AsyncIterator[str]:
|
||
"""
|
||
把受管 Agent 事件转换为 OpenAI 流式响应。
|
||
"""
|
||
event_queue: asyncio.Queue[OpenAIStreamEvent] = asyncio.Queue()
|
||
|
||
created = int(time.time())
|
||
completion_id = f"chatcmpl-{uuid.uuid4().hex}"
|
||
finished = False
|
||
|
||
async def _run_agent() -> None:
|
||
"""
|
||
运行受管 Agent 并把完成或异常结果送入响应队列。
|
||
"""
|
||
try:
|
||
await _run_managed_agent(
|
||
manager=manager,
|
||
session_id=session_id,
|
||
user_id=user_id,
|
||
username=username,
|
||
source="openai",
|
||
prompt=prompt,
|
||
images=images,
|
||
stream_mode=True,
|
||
event_queue=event_queue,
|
||
)
|
||
except asyncio.CancelledError:
|
||
await event_queue.put({"error": "MoviePilot AI agent is unavailable."})
|
||
except Exception as exc:
|
||
await event_queue.put({"error": str(exc)})
|
||
finally:
|
||
await event_queue.put(None)
|
||
|
||
task = resolve_background_task_registry(task_registry).create(
|
||
_run_agent(),
|
||
owner="api.openai.stream",
|
||
)
|
||
|
||
try:
|
||
yield _sse_payload(
|
||
{
|
||
"id": completion_id,
|
||
"object": "chat.completion.chunk",
|
||
"created": created,
|
||
"model": MODEL_ID,
|
||
"choices": [
|
||
{
|
||
"index": 0,
|
||
"delta": {"role": "assistant"},
|
||
"finish_reason": None,
|
||
}
|
||
],
|
||
}
|
||
)
|
||
|
||
while True:
|
||
item = await event_queue.get()
|
||
if item is None:
|
||
break
|
||
if isinstance(item, dict) and item.get("error"):
|
||
yield _sse_payload(
|
||
{
|
||
"error": {
|
||
"message": str(item["error"]),
|
||
"type": "server_error",
|
||
"code": "agent_execution_failed",
|
||
}
|
||
}
|
||
)
|
||
yield "data: [DONE]\n\n"
|
||
return
|
||
text = str(item or "")
|
||
if not text:
|
||
continue
|
||
yield _sse_payload(
|
||
{
|
||
"id": completion_id,
|
||
"object": "chat.completion.chunk",
|
||
"created": created,
|
||
"model": MODEL_ID,
|
||
"choices": [
|
||
{
|
||
"index": 0,
|
||
"delta": {"content": text},
|
||
"finish_reason": None,
|
||
}
|
||
],
|
||
}
|
||
)
|
||
|
||
finished = True
|
||
yield _sse_payload(
|
||
{
|
||
"id": completion_id,
|
||
"object": "chat.completion.chunk",
|
||
"created": created,
|
||
"model": MODEL_ID,
|
||
"choices": [
|
||
{
|
||
"index": 0,
|
||
"delta": {},
|
||
"finish_reason": "stop",
|
||
}
|
||
],
|
||
}
|
||
)
|
||
yield "data: [DONE]\n\n"
|
||
finally:
|
||
if cleanup_session:
|
||
await manager.clear_session(session_id=session_id, user_id=user_id)
|
||
elif not task.done():
|
||
await manager.stop_current_task(session_id)
|
||
if not task.done():
|
||
task.cancel()
|
||
try:
|
||
await task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
elif finished:
|
||
await task
|
||
|
||
|
||
def _is_manager_unavailable(error: BaseException) -> bool:
|
||
"""
|
||
识别管理器请求接纳检查的稳定错误，不导入完整编排模块。
|
||
"""
|
||
return getattr(error, "code", None) == "agent_manager_unavailable"
|
||
|
||
|
||
def _is_manager_queue_full(error: BaseException) -> bool:
|
||
"""
|
||
识别 Agent 会话排队已满，供兼容 API 返回可重试状态。
|
||
"""
|
||
return getattr(error, "code", None) == "agent_manager_queue_full"
|
||
|
||
|
||
def _manager_execution_error(error: BaseException) -> JSONResponse:
|
||
"""
|
||
把 AgentManager 稳定错误映射为 OpenAI 兼容错误响应。
|
||
"""
|
||
if _is_manager_unavailable(error):
|
||
return _error_response(
|
||
"MoviePilot AI agent is unavailable.",
|
||
503,
|
||
error_type="server_error",
|
||
code="ai_agent_unavailable",
|
||
)
|
||
if _is_manager_queue_full(error):
|
||
return _error_response(
|
||
str(error),
|
||
429,
|
||
error_type="rate_limit_error",
|
||
code="ai_agent_queue_full",
|
||
)
|
||
return _error_response(
|
||
str(error),
|
||
500,
|
||
error_type="server_error",
|
||
code="agent_execution_failed",
|
||
)
|
||
|
||
|
||
async def _run_managed_agent(
|
||
*,
|
||
manager: AgentManager,
|
||
session_id: str,
|
||
user_id: str,
|
||
username: str,
|
||
source: str,
|
||
prompt: str,
|
||
images: List[str],
|
||
stream_mode: bool,
|
||
event_queue: Optional[asyncio.Queue[OpenAIStreamEvent]] = None,
|
||
) -> tuple[str, List[str]]:
|
||
"""
|
||
通过 AgentManager 执行协议请求，并在 worker 内配置请求级输出。
|
||
"""
|
||
agent_holder: dict[str, _CollectingMoviePilotAgentMixin] = {}
|
||
|
||
def configure_agent(agent: MoviePilotAgent) -> None:
|
||
"""
|
||
为请求绑定协议模式和流式事件队列。
|
||
"""
|
||
# 会话管理器使用上方动态工厂创建或复用同一组合类型。
|
||
collecting_agent = cast(_CollectingMoviePilotAgentMixin, agent)
|
||
collecting_agent.configure_protocol_request(
|
||
stream_mode=stream_mode,
|
||
event_queue=event_queue,
|
||
)
|
||
agent_holder["agent"] = collecting_agent
|
||
|
||
try:
|
||
result = await manager.process_message(
|
||
session_id=session_id,
|
||
user_id=user_id,
|
||
message=prompt,
|
||
images=images,
|
||
files=None,
|
||
channel=NotificationChannel.Web.value,
|
||
source=source,
|
||
username=username,
|
||
reply_mode=ReplyMode.CAPTURE_ONLY,
|
||
allow_message_tools=True,
|
||
agent_factory=_get_collecting_agent_type(),
|
||
agent_setup=configure_agent,
|
||
wait_for_completion=True,
|
||
)
|
||
agent = agent_holder.get("agent")
|
||
return result, list(agent.collected_messages if agent else [])
|
||
finally:
|
||
agent = agent_holder.get("agent")
|
||
if agent is not None:
|
||
agent.release_protocol_request(event_queue)
|
||
|
||
|
||
def _error_response(
|
||
message: str,
|
||
status_code: int,
|
||
error_type: str = "invalid_request_error",
|
||
code: Optional[str] = None,
|
||
) -> JSONResponse:
|
||
"""
|
||
构造 OpenAI 协议格式的错误响应。
|
||
"""
|
||
return JSONResponse(
|
||
status_code=status_code,
|
||
content=_SchemaOpenAIErrorResponse(
|
||
error=_SchemaOpenAIErrorDetail(
|
||
message=message,
|
||
type=error_type,
|
||
code=code,
|
||
)
|
||
).model_dump(),
|
||
headers={"WWW-Authenticate": "Bearer"},
|
||
)
|
||
|
||
|
||
async def _check_auth(
|
||
credentials: Optional[HTTPAuthorizationCredentials],
|
||
) -> Optional[JSONResponse]:
|
||
"""
|
||
OpenAI 兼容接口以 API_TOKEN 认证受信客户端，认证通过即按管理员级 Agent 集成处理。
|
||
"""
|
||
if not credentials or credentials.scheme.lower() != "bearer":
|
||
return _error_response(
|
||
"Invalid bearer token.",
|
||
401,
|
||
error_type="authentication_error",
|
||
code="invalid_api_key",
|
||
)
|
||
if credentials.credentials != get_api_runtime_config_snapshot().api_token:
|
||
return _error_response(
|
||
"Invalid bearer token.",
|
||
401,
|
||
error_type="authentication_error",
|
||
code="invalid_api_key",
|
||
)
|
||
try:
|
||
await run_in_threadpool(validate_api_credential_identity)
|
||
except HTTPException as error:
|
||
authentication_error = error.status_code == 401
|
||
return _error_response(
|
||
(
|
||
"Invalid bearer token."
|
||
if authentication_error
|
||
else "Authentication service unavailable."
|
||
),
|
||
error.status_code,
|
||
error_type="authentication_error" if authentication_error else "api_error",
|
||
code="invalid_api_key" if authentication_error else None,
|
||
)
|
||
return None
|
||
|
||
|
||
@router.get(
|
||
"/models",
|
||
summary="OpenAI 兼容模型列表",
|
||
response_model=_SchemaOpenAIModelListResponse,
|
||
)
|
||
async def list_models(
|
||
credentials: Optional[HTTPAuthorizationCredentials] = Security(
|
||
openai_bearer_scheme
|
||
),
|
||
) -> _SchemaOpenAIModelListResponse | JSONResponse:
|
||
"""
|
||
校验凭据并返回当前可用的兼容模型。
|
||
"""
|
||
auth_error = await _check_auth(credentials)
|
||
if auth_error:
|
||
return auth_error
|
||
now = int(time.time())
|
||
return _SchemaOpenAIModelListResponse(
|
||
data=[_SchemaOpenAIModelInfo(id=MODEL_ID, created=now)]
|
||
)
|
||
|
||
|
||
async def _chat_completions_impl(
|
||
payload: _SchemaOpenAIChatCompletionsRequest,
|
||
request: Request,
|
||
credentials: Optional[HTTPAuthorizationCredentials] = Security(
|
||
openai_bearer_scheme
|
||
),
|
||
task_registry: TaskRegistry | None = None,
|
||
) -> JSONResponse | StreamingResponse:
|
||
"""
|
||
校验并执行 OpenAI Chat Completions 请求。
|
||
"""
|
||
auth_error = await _check_auth(credentials)
|
||
if auth_error:
|
||
return auth_error
|
||
|
||
if not get_api_runtime_config_snapshot().ai_agent_enable:
|
||
return _error_response(
|
||
"MoviePilot AI agent is disabled.",
|
||
503,
|
||
error_type="server_error",
|
||
code="ai_agent_disabled",
|
||
)
|
||
manager = get_running_agent_manager()
|
||
if manager is None:
|
||
return _error_response(
|
||
"MoviePilot AI agent is unavailable.",
|
||
503,
|
||
error_type="server_error",
|
||
code="ai_agent_unavailable",
|
||
)
|
||
|
||
if not payload.messages:
|
||
return _error_response(
|
||
"`messages` must be a non-empty array.",
|
||
400,
|
||
code="invalid_messages",
|
||
)
|
||
|
||
session_key = (
|
||
str(payload.user or "").strip()
|
||
or str(request.headers.get("x-session-id") or "").strip()
|
||
or str(uuid.uuid4())
|
||
)
|
||
use_server_session = bool(
|
||
str(payload.user or "").strip()
|
||
or str(request.headers.get("x-session-id") or "").strip()
|
||
)
|
||
|
||
try:
|
||
prompt, images = build_prompt(
|
||
payload.messages, use_server_session=use_server_session
|
||
)
|
||
except ValueError as exc:
|
||
return _error_response(str(exc), 400, code="invalid_messages")
|
||
|
||
session_id = build_session_id(session_key, SESSION_PREFIX)
|
||
username = str(payload.user or "openai-client")
|
||
if payload.stream:
|
||
return build_sse_response(
|
||
_stream_response(
|
||
manager=manager,
|
||
session_id=session_id,
|
||
user_id=session_key,
|
||
username=username,
|
||
prompt=prompt,
|
||
images=images,
|
||
cleanup_session=not use_server_session,
|
||
task_registry=task_registry,
|
||
),
|
||
)
|
||
|
||
collected_messages: list[str] = []
|
||
try:
|
||
result, collected_messages = await _run_managed_agent(
|
||
manager=manager,
|
||
session_id=session_id,
|
||
user_id=session_key,
|
||
username=username,
|
||
source="openai",
|
||
prompt=prompt,
|
||
images=images,
|
||
stream_mode=False,
|
||
)
|
||
except Exception as exc:
|
||
return _manager_execution_error(exc)
|
||
finally:
|
||
if not use_server_session:
|
||
await manager.clear_session(session_id=session_id, user_id=session_key)
|
||
|
||
content = "\n\n".join(
|
||
message.strip()
|
||
for message in collected_messages
|
||
if message and message.strip()
|
||
).strip()
|
||
if not content and result:
|
||
content = str(result).strip()
|
||
if not content:
|
||
content = "未获得有效回复。"
|
||
|
||
return JSONResponse(content=build_completion_payload(content, MODEL_ID))
|
||
|
||
|
||
async def _responses_impl(
|
||
payload: _SchemaOpenAIResponsesRequest,
|
||
credentials: Optional[HTTPAuthorizationCredentials] = Security(
|
||
openai_bearer_scheme
|
||
),
|
||
) -> _SchemaOpenAIResponsesResponse | JSONResponse:
|
||
"""
|
||
校验并执行 OpenAI Responses 请求。
|
||
"""
|
||
auth_error = await _check_auth(credentials)
|
||
if auth_error:
|
||
return auth_error
|
||
|
||
if not get_api_runtime_config_snapshot().ai_agent_enable:
|
||
return _error_response(
|
||
"MoviePilot AI agent is disabled.",
|
||
503,
|
||
error_type="server_error",
|
||
code="ai_agent_disabled",
|
||
)
|
||
manager = get_running_agent_manager()
|
||
if manager is None:
|
||
return _error_response(
|
||
"MoviePilot AI agent is unavailable.",
|
||
503,
|
||
error_type="server_error",
|
||
code="ai_agent_unavailable",
|
||
)
|
||
|
||
if payload.stream:
|
||
return _error_response(
|
||
"Streaming is not supported for /responses yet.",
|
||
400,
|
||
code="unsupported_stream",
|
||
)
|
||
|
||
normalized_messages = build_responses_input(
|
||
payload.input, instructions=payload.instructions
|
||
)
|
||
if not normalized_messages:
|
||
return _error_response(
|
||
"`input` must include at least one usable message.",
|
||
400,
|
||
code="invalid_input",
|
||
)
|
||
|
||
try:
|
||
prompt, images = build_prompt(
|
||
normalized_messages, use_server_session=bool(payload.user)
|
||
)
|
||
except ValueError as exc:
|
||
return _error_response(str(exc), 400, code="invalid_input")
|
||
|
||
session_key = str(payload.user or uuid.uuid4())
|
||
session_id = build_session_id(session_key, SESSION_PREFIX)
|
||
collected_messages: list[str] = []
|
||
try:
|
||
result, collected_messages = await _run_managed_agent(
|
||
manager=manager,
|
||
session_id=session_id,
|
||
user_id=session_key,
|
||
username=str(payload.user or "openai-client"),
|
||
source="openai.responses",
|
||
prompt=prompt,
|
||
images=images,
|
||
stream_mode=False,
|
||
)
|
||
except Exception as exc:
|
||
return _manager_execution_error(exc)
|
||
finally:
|
||
if not payload.user:
|
||
await manager.clear_session(session_id=session_id, user_id=session_key)
|
||
|
||
content = "\n\n".join(
|
||
message.strip()
|
||
for message in collected_messages
|
||
if message and message.strip()
|
||
).strip()
|
||
if not content and result:
|
||
content = str(result).strip()
|
||
if not content:
|
||
content = "未获得有效回复。"
|
||
|
||
created_at = int(time.time())
|
||
response_id = f"resp_{uuid.uuid4().hex}"
|
||
output_message = _SchemaOpenAIResponsesOutputMessage(
|
||
id=f"msg_{uuid.uuid4().hex}",
|
||
content=[_SchemaOpenAIResponsesOutputText(text=content)],
|
||
)
|
||
return _SchemaOpenAIResponsesResponse(
|
||
id=response_id,
|
||
created_at=created_at,
|
||
model=MODEL_ID,
|
||
output=[output_message],
|
||
usage=_SchemaOpenAIUsage(),
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/chat/completions",
|
||
summary="OpenAI 兼容聊天补全",
|
||
response_model=_SchemaOpenAIChatCompletionResponse,
|
||
responses={200: {"description": "OpenAI chat completion 或 SSE 数据流", "content": {"text/event-stream": {"schema": {"type": "string"}}}}},
|
||
)
|
||
async def chat_completions(
|
||
payload: _SchemaOpenAIChatCompletionsRequest,
|
||
request: Request,
|
||
credentials: Optional[HTTPAuthorizationCredentials] = Security(openai_bearer_scheme),
|
||
task_registry: TaskRegistry = Depends(get_background_task_registry_compat),
|
||
) -> JSONResponse | StreamingResponse:
|
||
"""
|
||
OpenAI Chat Completions 兼容公开入口。
|
||
"""
|
||
return await _chat_completions_impl(
|
||
payload,
|
||
request,
|
||
credentials,
|
||
task_registry=task_registry,
|
||
)
|
||
|
||
|
||
@router.post(
|
||
"/responses",
|
||
summary="OpenAI 兼容响应生成",
|
||
response_model=_SchemaOpenAIResponsesResponse,
|
||
)
|
||
async def responses(
|
||
payload: _SchemaOpenAIResponsesRequest,
|
||
credentials: Optional[HTTPAuthorizationCredentials] = Security(openai_bearer_scheme),
|
||
) -> _SchemaOpenAIResponsesResponse | JSONResponse:
|
||
"""
|
||
OpenAI Responses 兼容公开入口。
|
||
"""
|
||
return await _responses_impl(payload, credentials)
|