如何为 AI 应用构建 REST API

构建一个在推理缓慢、重试或连接真实数据时仍能保持可靠的 REST API。使用轻薄的 FastAPI 层来验证和认证请求，创建持久化的预测任务，返回 202 Accepted 和状态 URL，让 Worker 在 HTTP 请求之外处理模型执行。

本指南展示如何保持公共 API provider-agnostic（不绑定特定提供商），在生产路径中使用 Oracle AI Database 持久化任务状态，在需要时流式传输交互式响应，并执行租户感知的访问控制。内容基于 Oracle AI Developer Hub 中关于 SSE 传输、检索编排和企业数据访问的维护型 proof cases。

Illustrative patterns 与 maintained proof cases

本指南中的代码用于解释 API 边界。文章后面的维护型 proof cases 展示这些边界在实际应用中的表现。生产环境中，请使用 Oracle AI Database 持久化、租户作用域的访问、独立的 Worker 或持久化队列、批准的 identity、retention、监控和分布式限流。SQL、VPD、流式传输和 BackgroundTasks 代码片段是设计指导，而非可直接复制部署的代码。

围绕资源生命周期构建 REST API：提交工作，返回 202 Accepted，暴露状态 URL。

在调用模型之前使用 Pydantic 验证有界的 JSON，并从实现中自动生成 OpenAPI 契约。

将模型提供商凭证保留在服务端，并将提供商特定的行为隔离在适配器之后。

持久化任务状态，将每次读取限定在已认证的租户范围内，并支持幂等重试。

对于可变或长时间运行的推理，使用 Worker 和持久化队列；仅在时间可预测时使用同步响应。

将 FastAPI、Oracle AI Database、ORDS、OpenAI Responses API、流式传输和队列视为具有不同职责的独立架构边界。

在部署前测试认证、租户隔离、输入拒绝、限流、提供商故障、Worker 转换和公共 OpenAPI 契约。

你将设计和验证什么？

你将设计一个小型的 FastAPI 服务，它接受预测请求、认证调用者、返回 202 Accepted（带 Location 头）、在 HTTP 请求之外处理请求，并通过 GET /v1/predictions/{id} 暴露最终状态。同一服务自动生成 /docs 和 /openapi.json。

使用 Developer Hub 部分中的三个维护型 proof cases 来验证对你的应用而言最重要的边界：交互式流式传输、检索和编排，或身份感知的企业数据访问。

文章其余部分为你提供设计检查清单：轻量路由、清晰的资源契约、提供商适配器、异步工作、流式传输、身份认证和受治理的持久化状态。Proof cases 展示这些组件如何在维护型应用中协同工作。

为 AI 应用构建 REST API 的最佳实践是什么？

使用清晰的资源契约，在模型调用前验证输入，将提供商凭证保留在服务端，审慎选择同步或异步执行，持久化任务状态，将读取限定在已认证租户范围内，暴露 OpenAPI 和安全的错误信息。在部署前，像测试成功响应一样仔细地测试失败路径。

推荐的实现通过五个边界来应用这些实践：

• HTTP 边界：FastAPI、Pydantic、OpenAPI、认证、限流和请求 ID。
• 任务边界：POST /v1/predictions 创建一个持久化资源并返回 202 Accepted。
• Worker 边界：独立进程认领工作并调用模型适配器。
• 数据边界：Oracle AI Database 存储请求生命周期并应用租户作用域。
• 提供商边界：模型集成是可替换的，不得将提供商凭证暴露给客户端。

如何为 AI 应用设计一个轻薄的 REST API 层？

保持 REST API 轻薄：在 HTTP 边缘进行认证和验证，创建或读取任务资源，将模型调用、检索和持久化状态委托给专用边界。路由不应包含提供商特定逻辑、长时间运行的推理，或数据层必须再次强制执行的授权规则。

将推理视为资源生命周期，而非单一的长时间运行的 HTTP 函数。API 接受并记录工作，Worker 执行推理，客户端通过稳定的状态 URL 读取结果资源。在生产路径中，FastAPI 拥有面向应用的边界，Oracle AI Database 拥有持久的预测任务状态。

Chatbot REST API 应该在 FastAPI 中暴露哪些端点？

Chatbot REST API 应该将对话资源、消息、预测任务和可选流分开。这使得聊天历史可寻址，使缓慢的推理可观测，并让授权应用到每个资源上，而不是将整个生命周期隐藏在单个 /chat 请求背后。

对于 Chatbot 风格的 REST API，暴露映射到相同资源生命周期的端点：

POST /v1/conversations/{conversation_id}/messages — 创建一条消息并调度底层预测任务。
GET /v1/conversations/{conversation_id} — 返回对话元数据和当前状态。
GET /v1/conversations/{conversation_id}/messages — 返回授权后的消息历史。
GET /v1/predictions/{id} — 轮询持久化任务直到模型响应就绪。
GET /v1/conversations/{conversation_id}/messages/{message_id}/stream — 可选地通过 Server-Sent Events (SSE) 流式传输增量输出。
DELETE /v1/conversations/{conversation_id} — 根据保留策略删除对话数据。

面向客户端的流程保持一致：FastAPI 认证并创建工作，Worker 在 Oracle AI Database 中更新持久化任务状态，客户端通过状态 URL 或授权后的流读取响应。

这个模式在以下场景中很有用：模型推理可能超出请求超时时间；客户端需要可靠的状态信息；或 AI 应用必须在受治理的业务数据旁边保留请求上下文。

为什么选择 Oracle AI Database 作为生产路径？

Oracle AI Database 为生产路径提供了一处统一管理持久化预测任务状态的位置。它可以将工作限定在租户范围内，支持可靠的状态读取和幂等记录，并在 API 暴露的数据行附近应用数据访问控制。

Proof cases：将 API 模式应用到实际的 AI 开发任务

REST API 模式只有在连接到开发者实际需要交付的工作时才有价值。以下三个 Oracle AI Developer Hub 资产不是本教程的替代品；每个都是设计中某个边界的可用 proof case。

任务：交付交互式 AI 体验，而不将路由变成 Agent

当你需要从 POST /v1/chat 原型迁移到交互式应用时，使用 Total Recall。它通过 SSE 流式传输进度，同时将 Agent 框架保留在浏览器之外。它是一个 FastAPI 后端和浏览器应用，通过 SSE 流式传输，然后将上下文、检索、memory、工具和 trace 等周围关注点作为独立层暴露。

这证明了什么 API 设计：流式传输是 HTTP 边界处的传输机制；它不能消除对稳定资源模型、服务端提供商凭证或持久化状态的需求。当产品需求是"现在就显示有用的进度"而非"稍后返回一个阻塞响应"时，使用这个 proof case。

立即运行：准备 Python 3.11+、一个可访问的 Oracle AI Database（及其所需的 embedding 模型）和一个已批准的模型提供商凭证；然后按照 appbook 设置并运行 ./run.sh。其有记录的成功信号是一个本地浏览器应用，FastAPI 服务可用，以及数据库、框架、重排序器回退或可用性以及已配置模型提供商的状态指示器。使用实时 trace 和流式聊天来检查你的路由是否在不将框架移入客户端的情况下交付增量输出。

任务：让检索增强的 AI API 从正确的证据中作答

当 API 必须做的不仅仅是调用模型时，使用 From RAG to Agents 工作坊。该工作坊从数据加载和检索，到 RAG、Agent 工具、多 Agent 编排和持久化会话 memory，构建一个研究论文助手。

这证明了什么 API 设计：检索和编排属于提供商无关的服务边界之后。让 POST /v1/predictions 或 POST /v1/chat 负责验证、身份、状态和响应结构；让 Worker 或 Agent 运行时组装检索、工具调用和模型执行。当开发者任务是"将基于检索的 grounding 转化为应用特性，同时不将内部管道暴露为公共 API"时，使用这个 proof case。

立即运行：安装 Docker、Python 和 Jupyter；启动 workshop 的 Oracle AI Database 容器，安装依赖，然后打开 workshop/notebook_student.ipynb。学习路径从数据加载和检索开始，经过 RAG、工具、编排和会话记忆。证明不仅是模型能回答——而是你可以检查是哪个检索和编排边界产生了回答。

工作目标：在不绕过身份和访问策略的前提下暴露企业数据

当 API 服务于应该看到不同数据的人时，使用 Enterprise Data Agent workshop。它将笔记本与正在运行的应用配对，演示身份感知的行级和列级策略、检索、记忆、工具以及针对同一数据库的实时聊天界面。

这对 API 设计的证明意义在于：路由级别的身份验证是必要的，但并不足够。在 FastAPI 中派生调用者身份，通过批准的模式将其传递到数据库会话，并使检索、对话读取和工具访问应用相同的作用域。当开发者的工作是"为企业数据添加 AI 能力而不创建第二条不受治理的访问路径"时，使用此证明案例。

立即运行：使用 workshop 的 GitHub Codespaces 路径以获得最低摩擦的设置，或按照其本地 Docker、Python 和 Node.js 设置来启动笔记本和配套应用。需要寻找的证明是身份相关的数据访问：相同的应用和智能体工作流应该受到活动身份和数据策略的约束，而不是信任聊天请求中提供的无作用域标识符。

证据边界：这些链接固定在特定的 Developer Hub 提交上，以便其代码和设置说明保持可审查性。它们演示了工作应用中的 API 边界：Total Recall 演示了 FastAPI 和 SSE；从 RAG 到智能体演示了检索和编排；Enterprise Data Agent 演示了身份感知访问。它们不对你的部署的延迟、吞吐量、可用性或安全性进行基准测试。在发布前根据你的模型、数据、身份提供者和流量进行衡量。

如何让公共 API 独立于 AI 提供商？

在应用边界定义一个与提供商无关的请求、结果和错误契约。将每个模型 SDK 放在适配器后面，该适配器将内部请求转换为提供商调用，并将结果映射回你的 API 的稳定模式。客户端不应接收提供商响应对象、提供商特定的错误负载或凭据。

from typing import Protocol

class ModelResult(BaseModel):
    text: str
    provider_request_id: str | None = None

class ModelAdapter(Protocol):
    def generate(self, *, prompt: str, temperature: float) -> ModelResult: ...

通过服务端配置选择适配器。这样，当提供商或模型更改时，HTTP 契约、作业模式、授权和测试可以保持稳定。

FastAPI 的 /docs 和 /openapi.json 在开发过程中如何提供帮助？

FastAPI 从请求和响应模型生成交互式 /docs 页面和机器可读的 /openapi.json 契约。使用它们来尝试经过身份验证的开发请求、检查验证规则、与前端或平台团队分享精确契约、生成客户端以及添加契约测试。它们改进了开发者反馈；它们不能替代身份验证、授权、速率限制或网关控制。

在部署前，确认 /openapi.json 仅暴露预期的字段和响应。决定 /docs 是否应在开发环境之外保持可用，并根据部署的运营策略对其进行保护。

AI 推理应该是同步还是异步的？

仅当完成时间可预测且符合 API 契约时，才使用同步响应。当推理可能需要不可预测的时间、需要重试或必须独立于 HTTP 请求进行审计时，使用异步作业资源。

AI 响应可能在可预测的 HTTP 截止时间内未准备就绪。作业资源让 API 快速确认有效工作，防止请求线程等待推理，并给客户端一个稳定的资源来轮询。

不要将诸如 /generateText 这样的 URL 作为主要接口。预测是一个有生命周期的资源；API 应该让这个生命周期可见。

client → FastAPI → prediction_jobs in Oracle AI Database
                  ↓
             durable worker → approved model adapter
                  ↓
client ← GET /v1/predictions/{id}

进程中后台任务对本地原型很有用。对于生产环境，运行单独的工作进程并使用持久队列或等效的平台服务，以便接受的工作能够在 Web 进程重启后存活。

如何在 FastAPI 中为聊天添加流式响应？

当客户端受益于增量响应内容时，使用 Server-Sent Events。流式传输是一种输出传递选择，而不是持久作业状态的替代方案：持久化权威结果并公开状态资源，即使用户界面接收到部分内容。

import asyncio
import json

from fastapi import Request
from fastapi.responses import StreamingResponse

@app.get("/v1/conversations/{conversation_id}/messages/{message_id}/stream")
async def stream_message(request: Request, conversation_id: str, message_id: str):
    async def events():
        try:
            async for chunk in worker.stream(message_id):
                if await request.is_disconnected():
                    return
                yield "event: message\\ndata: " + json.dumps({"delta": chunk}) + "\\n\\n"
            yield "event: done\\ndata: {}\\n\\n"
        except asyncio.CancelledError:
            return
        except Exception:
            yield 'event: error\\ndata: {"code":"stream_failed"}\\n\\n'

return StreamingResponse(events(), media_type="text/event-stream")

将事件类型和负载结构定义为公共契约的一部分。将每个块序列化为 JSON，而不是将原始模型输出内插到 SSE 帧中；然后检查客户端断开连接、处理取消、将提供商错误保留在服务端，并为后续检索最终化持久结果。

如何使用 FastAPI 构建 AI REST API？

定义有边界的 Pydantic 请求和响应模型，拒绝未知字段，使用 FastAPI 依赖项进行身份验证，返回明确的状态码，并让 FastAPI 从实现中生成 OpenAPI 契约。

接下来的两个代码片段展示了请求模型和提交边界。将它们视为可适应你的身份提供者、作业存储和工作运行时的模式；维护的证明案例展示了周围的应用行为。

第 1 步：定义一个有边界的请求模式

使用 Pydantic 在请求到达昂贵的模型调用之前拒绝未知字段并限定输入。

class PredictionRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")

prompt: Annotated[str, Field(min_length=5, max_length=8_000)]
    temperature: Annotated[float, Field(default=0.2, ge=0.0, le=1.0)]

这些限制是应用决策，不是通用默认值。从模型上下文窗口、你的滥用策略和预期负载结构中设置它们。

第 2 步：提交工作并返回 202 Accepted

端点在调度工作之前创建一个预测作业。它返回一个 Location 头和一个 JSON 状态 URL。

from fastapi import BackgroundTasks, Depends, Header, Request, status
from fastapi.responses import JSONResponse

@app.post("/v1/predictions")
async def create_prediction(
    payload: PredictionRequest,
    background_tasks: BackgroundTasks,
    request: Request,
    identity: VerifiedIdentity = Depends(require_identity),
    idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"),
):
    job_id = str(uuid4())
    status_url = f"/v1/predictions/{job_id}"
    repository.create(
        job_id=job_id,
        tenant_id=identity.tenant_id,
        client_id=identity.user_id,
        prompt=payload.prompt,
        temperature=payload.temperature,
        request_id=request.state.request_id,
        idempotency_key=idempotency_key,
    )
    background_tasks.add_task(worker.process_once)
    return JSONResponse(
        status_code=status.HTTP_202_ACCEPTED,
        headers={"Location": status_url},
        content={"id": job_id, "status": "queued", "status_url": status_url},
    )

当 API 已接受工作但尚未完成时，使用 202 Accepted。不要选择固定的时间阈值（如两秒）；当结果无法在请求契约内可靠地完成时，使用作业资源。

此 BackgroundTasks 示例是说明性的，而不是生产运行模式。对于繁重的、需要重试的或必须能够承受进程重启的工作，使用单独的工作进程和持久队列；FastAPI 在其后台任务指导中做出了同样的区分。

如何在生产环境中用 Oracle AI Database 持久化 AI 任务？

将任务请求、租户作用域、生命周期状态、安全结果或错误码、请求关联 ID 以及幂等指纹存储在 Oracle AI Database 中。保持事务简短，并通过连接池使用绑定变量。

第 3 步：在 Oracle AI Database 中持久化预测状态

该服务将请求、客户端作用域、状态、结果、安全错误码和请求 ID 记录到一张 prediction_jobs 表中。这使 API 状态与可能影响模型请求或控制谁可以获取结果的应用程序数据保持紧密关联。

CREATE TABLE prediction_jobs (
    id             VARCHAR2(36) PRIMARY KEY,
    tenant_id      VARCHAR2(255) NOT NULL,
    client_id      VARCHAR2(255) NOT NULL,
    status         VARCHAR2(16) NOT NULL,
    prompt         CLOB NOT NULL,
    temperature    NUMBER(3,2) NOT NULL,
    generated_text CLOB,
    error_code     VARCHAR2(128),
    request_id     VARCHAR2(64) NOT NULL,
    idempotency_key VARCHAR2(255),
    request_fingerprint VARCHAR2(64),
    created_at     TIMESTAMP WITH TIME ZONE NOT NULL,
    updated_at     TIMESTAMP WITH TIME ZONE NOT NULL,
    CONSTRAINT prediction_jobs_idempotency_uq
      UNIQUE (tenant_id, idempotency_key)
);

数据库调用使用绑定变量。生产实现应使用 oracledb.create_pool，然后仅在创建、认领或完成任务的短事务期间获取连接。

第 4 步：由 Worker 处理模型推理

Worker 认领一个排队的任务，调用模型适配器，并记录结果或安全错误码。

def process_once(self) -> bool:
    job = self._repository.claim_next()
    if not job:
        return False
    try:
        result = self._model.generate(prompt=job.prompt, temperature=job.temperature)
        self._repository.mark_succeeded(job.id, result)
    except Exception:
        self._repository.mark_failed(job.id, "model_inference_failed")
    return True

不要将提供商的原始错误暴露在公共响应中。只记录经过脱敏处理的关联诊断信息，然后使用 request_id 在内部调查故障。

如何用 OpenAI SDK 构建一个最小的 FastAPI 聊天端点？

对于一个最小化的同步原型，验证消息、从服务器调用 OpenAI SDK，然后返回你自己稳定的 JSON 形状。保持这个路由精简：它演示的是请求边界，而不是一个持久的生产任务系统。

import os
from fastapi import FastAPI
from openai import AsyncOpenAI
from pydantic import BaseModel, Field

app = FastAPI()
client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])

class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=8_000)

@app.post("/v1/chat")
async def create_chat(payload: ChatRequest):
    response = await client.responses.create(
        model=os.environ["OPENAI_MODEL"],
        input=payload.message,
    )
    return {"output_text": response.output_text}

对于可变的、昂贵的或需要审计的推理，用前文描述的 202 预测任务模式替换直接的 SDK 调用。在将此路由用于本地原型之外的环境之前，先在安全章节中添加经过 authenticated Depends 模式的验证。不要返回提供商的原始响应对象，也不要向客户端暴露提供商凭证。

如何将 REST API 连接到 OpenAI Responses API？

将模型提供商放在服务器端适配器后面。从密钥管理器或环境变量加载 OPENAI_API_KEY，显式设置 OPENAI_MODEL，从 Worker 调用 Responses API，捕获提供商的请求 ID 用于诊断，并仅返回你自己的 API 的稳定响应契约。

如何安全地存储和加载 OpenAI API 密钥？

将 OpenAI API 密钥存储在经过批准的密钥管理器中，并在运行时将其注入到服务器或 Worker 中。环境变量适合作为进程级传递机制，但值应来源于托管的密钥存储，而不是源代码、Docker 镜像、Jupyter notebook、浏览器 JavaScript 或提交的 .env 文件。

为每个环境和工作负载身份使用不同的密钥。

限制谁和什么可以读取该密钥。

当所需密钥缺失时使启动失败，而不是接受客户端请求中的密钥。

不要在日志、追踪、异常、健康检查响应或配置转储中打印密钥。

根据组织的凭证策略以及在疑似泄露后轮换密钥。

在服务器端加载值并直接传递给提供商客户端：

import os
from openai import OpenAI

api_key = os.environ["OPENAI_API_KEY"]
openai_client = OpenAI(api_key=api_key)

这只是说明提供商边界的一个示例，并非可下载的实现或配置配方。当前的 OpenAI API 参考将 Responses 描述为直接的模型请求 API，需要服务器端持有者凭证，并建议在生产前记录请求 ID 和审查速率限制。

不要将提供商密钥放在浏览器 JavaScript 中，也不要接受请求体中的密钥。返回稳定的内部错误而不是原始提供商错误。API 的调用方认证和模型提供商的认证是两个独立的信任边界。

如何对 AI REST API 进行身份验证和安全防护？

在昂贵工作之前进行身份验证，从经过验证的凭证中解析租户身份，对每个任务的读取和更新应用授权，执行请求限制，约束负载大小，支持幂等重试，脱敏提供商错误，并定义提示词和生成内容的保留策略。

如何存储对话并执行用户所有权检查？

将每个对话与服务器推导的 tenant_id 和 owner_id 一起存储，然后将对话标识符传递到其消息和预测任务中。每个读取、更新、流转和删除操作都必须包含经过身份验证的所有权作用域；知道对话 ID 并不等于授权。

SELECT conversation_id, status, updated_at
FROM ai_conversations
WHERE conversation_id = :conversation_id
  AND tenant_id = :authenticated_tenant_id
  AND owner_id = :authenticated_user_id;

加载消息历史或返回生成内容时应用相同的作用域。作为深度防御，在 Oracle AI Database 中通过批准的策略强制执行租户谓词。