如何为AI应用构建REST API

Autonomous AI Database Wojtek Pluta 的头像为 AI 应用构建 REST API 时，要确保推理变慢、重试或接入真实数据时服务依然可靠。用一层轻薄的 FastAPI 校验并认证请求，创建持久化的预测任务，返回 202 Accepted 和状态查询 URL，把模型执行交给 worker 在 HTTP 请求之外处理。本文介绍如何让公开 API 与模型供应商解耦、在生产路径上用 Oracle AI Database 持久化任务状态、在需要时流式返回交互式响应，并落实租户级访问控制。内容基于 Oracle AI Developer Hub 中持续维护的 proof case，涵盖 SSE 推送、检索编排和企业数据访问。

示意性模式与持续维护的 proof case 本文中的代码用于说明 API 的边界。文中后面的持续维护 proof case 会展示这些边界在实际应用中的运行方式。生产环境请使用 Oracle AI Database 持久化、租户级访问范围、独立的 worker 或持久化队列、经过审批的身份体系、数据保留策略、监控以及分布式限流。SQL、VPD、流式和 BackgroundTasks 代码片段只是设计参考，不是可以直接复制部署的代码。

要点

围绕资源生命周期设计 REST API：提交任务、返回 202 Accepted、提供状态查询 URL。
调用模型之前，先用 Pydantic 校验有大小限制的 JSON，并从实现自动生成 OpenAPI 契约。
模型供应商的凭证只保留在服务端，把供应商特有的行为隔离在适配器之后。
持久化任务状态，每次读取都限定在已认证租户范围内，并支持幂等重试。
推理耗时不定或长时间运行时，使用 worker 和持久化队列；只有耗时可预测时才用同步响应。
把 FastAPI、Oracle AI Database、ORDS、OpenAI Responses API、流式和队列视为各自独立的架构边界，各司其职。
部署前测试认证、租户隔离、输入拒绝、限流、供应商故障、worker 状态流转以及公开的 OpenAPI 契约。

你将设计和验证什么？

你将设计一个小型 FastAPI 服务：接收预测请求，认证调用方，返回带 Location 响应头的 202 Accepted，在 HTTP 请求之外处理请求，并通过 GET /v1/predictions/{id} 暴露最终状态。同一个服务会自动生成 /docs 和 /openapi.json。用 Developer Hub 部分中三个持续维护的 proof case，验证对你的应用真正重要的那条边界：交互式流式输出、检索与编排，或感知身份的企业数据访问。文章余下部分给出设计清单：轻量路由、清晰的资源契约、供应商适配器、异步任务、流式输出、身份认证，以及受治理的持久化状态。proof case 则展示这些部分在持续维护的应用中如何协同工作。

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

使用清晰的资源契约，在调用模型前校验输入，把供应商凭证保留在服务端，有意识地选择同步或异步执行，持久化任务状态，把读取限定在已认证租户范围内，并公开 OpenAPI 与安全的错误信息。测试失败路径要和测试模型成功响应一样认真。推荐的实现通过五个边界落实这些实践：

**HTTP 边界：**FastAPI、Pydantic、OpenAPI、认证、限流和请求 ID。
任务边界：POST /v1/predictions 创建持久化资源并返回 202 Accepted。
**Worker 边界：**独立进程领取任务并调用模型适配器。
**数据边界：**Oracle AI Database 存储请求生命周期并施加租户范围。
**供应商边界：**模型集成可替换，且绝不能向客户端暴露供应商凭证。

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

让 REST API 保持轻薄：在 HTTP 边缘完成认证和校验，

创建或读取 job 资源，把模型调用、检索和持久状态交给专门的边界处理。路由里不该出现 provider 特有的逻辑、长时间运行的推理，也不该有数据层必须再次执行的授权规则。

把推理当作资源生命周期来对待，而不是一个长时间运行的 HTTP 函数。API 接收并记录工作，worker 执行推理，客户端通过稳定的状态 URL 读取生成的资源。在生产路径中，FastAPI 负责面向应用的边界，Oracle AI Database 负责持久化的 prediction job 状态。

FastAPI 中的聊天机器人 REST API 应该暴露哪些端点？

聊天机器人 REST API 应该把对话资源、消息、prediction job 和可选的流分开。这样聊天历史可寻址，缓慢的推理变得可观测，授权也能作用于每个资源，而不是把整个生命周期藏在一个 /chat 请求后面。

对于聊天机器人风格的 REST API，暴露的端点应干净地映射到同一套资源生命周期：

POST /v1/conversations/{conversation_id}/messages ——创建消息并调度底层的 prediction job。
GET /v1/conversations/{conversation_id} ——返回对话元数据和当前状态。
GET /v1/conversations/{conversation_id}/messages ——返回已授权的消息历史。
GET /v1/predictions/{id} ——轮询持久化 job，直到模型响应就绪。
GET /v1/conversations/{conversation_id}/messages/{message_id}/stream ——可选地用 Server-Sent Events（SSE）流式输出增量结果。
DELETE /v1/conversations/{conversation_id} ——按保留策略删除对话数据。

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

当模型推理可能超出请求超时、客户端需要可靠的状态信息，或者 AI 应用必须在受治理的业务数据旁保留请求上下文时，这个模式很有用。

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

Oracle AI Database 让生产路径有一个地方，可以把持久化的 prediction job 状态与受治理的应用数据放在一起。它可以按租户划分工作范围，支持可靠的状态读取和幂等记录，并把数据访问控制施加在 API 所暴露的那些行附近。

验证案例：把 API 模式套用到真实的 AI 开发者任务上

REST API 模式只有在连接到开发者真正要交付的工作时才有用。以下三个 Oracle AI Developer Hub 资源不是本教程的替代品；每一个都是设计中某个边界的可用验证案例。

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

当你需要从 POST /v1/chat 原型走向一个交互式应用，在流式输出进度的同时把 agent harness 留在浏览器之外时，使用 Total Recall。它是一个 FastAPI 后端加浏览器应用，通过 SSE 流式传输，然后把周边关注点——上下文、检索、记忆、工具和 trace——作为独立的层暴露出来。

这对 API 设计证明了什么： 流式传输是 HTTP 边界上的一种交付机制；它并不能免除对稳定资源模型、服务端 provider 凭据或持久状态的需求。当产品需求是“现在就展示有用的进度”而不是“稍后返回一个阻塞式响应”时，用这个验证案例。

现在就跑起来： 准备 Python 3.11+、一个可访问且带有其所需 embedding 模型的 Oracle AI Database，以及一个已获批的模型 provider 凭据；然后按照 appbook 的设置操作并运行 ./run.sh。它记录的成功的信号是：本地浏览器应用可用，FastAPI 服务可用，并且有数据库、harness、reranker 回退或可用性、已配置模型 provider 的状态指示。用实时 trace 和流式聊天来检查你的路由是否在不让 harness 进入客户端的前提下交付了增量输出。

任务：让检索支撑的 AI API 基于正确的证据作答

当 API 需要做的不仅仅是调用模型时，用 From RAG to Agents workshop。这个 workshop 从数据加载和检索开始，一路构建出一个研究论文助手，涵盖 RAG、agent 工具、多 agent 编排和持久会话记忆。 这对 API 设计意味着什么： 检索和编排应当放在一个与供应商无关的服务边界之后。让 POST /v1/predictions 或 POST /v1/chat 继续负责校验、身份、状态和响应结构；把检索、工具调用和模型执行的组装交给 worker 或 agent runtime。当开发者的任务是“把有依据的检索变成应用功能，同时不把内部流水线暴露为公开 API”时，用这个验证案例。 现在就跑起来： 安装 Docker、Python 和 Jupyter；启动 workshop 的 Oracle AI Database 容器，安装依赖，打开 workshop/notebook_student.ipynb。学习路径从数据加载和检索开始，依次经过 RAG、工具、编排和会话记忆。要验证的不只是模型能给出回答——而是你能查清是哪个检索和编排边界产出了这个回答。

任务：在不绕过身份和访问策略的前提下开放企业数据

当 API 服务的用户应当看到不同的数据时，用 Enterprise Data Agent workshop。它把一个 notebook 和一个运行中的应用配对，演示针对同一数据库的身份感知行级和列级策略、检索、记忆、工具，以及一个实时聊天界面。 这对 API 设计意味着什么： 在路由上做认证是必要的，但还不够。在 FastAPI 中推导出调用方身份，通过经过认可的模式把这个身份带入数据库会话，并让检索、会话读取和工具访问都应用同一套范围。当开发者的任务是“给企业数据加上 AI 能力，同时不制造第二条不受治理的访问路径”时，用这个验证案例。 现在就跑起来： 用 workshop 的 GitHub Codespaces 路径，配置摩擦最小；或者按它的本地 Docker、Python 和 Node.js 配置启动 notebook 和配套应用。要验证的是数据访问随身份而变：同一个应用和 agent 工作流应当受当前身份和数据策略约束，而不是信任聊天请求里传入的一个无范围限定的标识符。

证据边界： 这些链接固定到 Developer Hub 的某个具体 commit，因此其代码和配置说明始终可复查。它们展示的是一个可运行应用中的 API 边界：Total Recall 展示 FastAPI 和 SSE；From RAG to Agents 展示检索和编排；Enterprise Data Agent 展示身份感知访问。它们不对你部署的延迟、吞吐、可用性或安全性做基准测试。发布前，请针对你自己的模型、数据、身份提供商和流量去测量这些指标。

如何让公开 API 独立于 AI 供应商？

在应用边界定义一套与供应商无关的请求、结果和错误契约。把每个模型 SDK 放在一个适配器之后，由适配器把内部请求翻译成对供应商的调用，再把结果映射回你 API 的稳定 schema。客户端不应收到供应商的响应对象、供应商特有的错误载荷或凭据。 Copied to ClipboardError: Could not Copy

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 契约、任务 schema、授权和测试都能保持稳定。

开发期间 FastAPI `/docs` 和 `/openapi.json` 有什么帮助？

FastAPI 会根据请求和响应模型生成一个交互式 /docs 页面和一份机器可读的 /openapi.json 契约。用它们来试发带认证的开发请求、检查校验规则、与前端或平台团队共享精确契约、生成客户端，以及添加契约测试。它们改善的是开发者反馈；它们不能替代认证、授权、速率限制或网关控制。

部署前，确认 /openapi.json 仅暴露有意公开的字段和响应。决定 /docs 是否应在开发环境之外继续可用，并按部署的运维策略加以保护。

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

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

AI 响应未必能在可预测的 HTTP 截止时间内就绪。作业资源让 API 能快速确认收到有效工作，避免请求线程阻塞在推理上，同时给客户端一个稳定的轮询资源。

使用以下资源：

端点	用途
`POST /v1/predictions`	校验提示词并提交预测作业，返回 `202 Accepted`。
`GET /v1/predictions/{id}`	读取作业的排队、运行、成功或失败状态。
`GET /healthz`	供平台健康检查确认 API 进程可用。

不要把 /generateText 这类 URL 当作主要接口。预测是一个有生命周期的资源，API 应当让这个生命周期可见。

Copied to ClipboardError: Could not Copy

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

进程内后台任务对本地原型够用。生产环境要单独跑一个 worker 进程，并使用持久化队列或同等的平台服务，这样已接收的工作才能在 Web 进程重启后存活。

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

当客户端能从增量响应内容中获益时，用 Server-Sent Events。流式传输只是输出方式的选择，不能替代持久化的作业状态：即使用户界面收到的是部分内容，也要持久化权威结果并提供状态资源。

Copied to ClipboardError: Could not Copy

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 契约。

下面两段代码展示请求模型和提交边界。把它们当作可适配到你的身份提供商、作业存储和 worker 运行时的模式；维护中的验证用例展示了周边应用的行为。

第 1 步：定义有边界的请求 schema

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

Copied to ClipboardError: Could not Copy

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)]

这些限制是应用层面的决策，不是通用默认值。应根据模型上下文窗口、滥用策略和预期的 payload 形态来设定。

第 2 步：提交任务并返回 `202 Accepted`

端点先创建预测任务，再调度执行。它返回一个 Location 头和 JSON 状态 URL。

Copied to ClipboardError: Could not Copy

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。不要选两秒这样的固定时间阈值；只要结果无法在请求契约内可靠完成，就用 job 资源。

这个 BackgroundTasks 示例只是示意，不是生产运行时的模式。对于繁重、需要重试或必须跨进程重启存活的任务，要用独立的 worker 和持久化队列；FastAPI 在后台任务指南中做了同样的区分。

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

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

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

服务把请求、客户端范围、状态、结果、安全错误码和请求 ID 记录到 prediction_jobs 表中。这样 API 状态就贴近了应用数据，而这些数据可能决定模型请求的形态，或决定谁能取回结果。

Copied to ClipboardError: Could not Copy

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}

如果推理过程可变、成本较高或需要审计，就把直接调用 SDK 换成上文所述的 202 预测任务模式。在把这个路由用于本地原型之外的场景之前，先加上安全章节里介绍的带认证的 Depends 模式。不要返回提供方的原始响应对象，也不要把提供方的凭据暴露给客户端。

如何把 REST API 接到 OpenAI Responses API？

把模型提供方藏在服务端适配器后面。从密钥管理服务或环境变量中加载 OPENAI_API_KEY，显式设置 OPENAI_MODEL，在 worker 中调用 Responses API，记录提供方的请求 ID 以便排查问题，并且只返回你自己 API 的稳定响应契约。

应该如何安全地存储和加载 OpenAI API key？

把 OpenAI API key 存放在经过批准的密钥管理服务中，在运行时注入服务端或 worker。环境变量适合作为进程级的传递方式，但密钥值应当来自受管理的密钥存储，而不是源代码、Docker 镜像、notebook、浏览器 JavaScript 或已提交的 .env 文件。

每个环境和工作负载身份使用不同的密钥。
限制哪些人和哪些组件可以读取该密钥。
缺少必需密钥时让启动失败，而不是从客户端请求中接收密钥。
不要在日志、trace、异常、健康检查响应或配置导出中打印密钥。
按照组织的凭据策略轮换密钥，并在怀疑泄露后立即轮换。

在服务端加载密钥，直接传给 provider 客户端：

Copied to ClipboardError: Could not Copy

import os
from openai import OpenAI

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

这里展示的是 provider 边界的示意写法，并非可直接下载使用的实现或配置方案。当前的 OpenAI API 参考文档将 Responses 描述为直接发起模型请求的 API，要求使用服务端 bearer 凭证，并建议在生产环境使用前记录请求 ID、检查速率限制。

不要把 provider 密钥放进浏览器 JavaScript，也不要从请求体接收密钥。应返回一个稳定的内部错误，而不是 provider 的原始错误。API 自身的调用方认证与模型 provider 的认证是两个独立的信任边界。

如何对 AI REST API 做认证和安全防护？

在进入昂贵操作之前先完成认证，从已验证的凭证中解析租户身份，对每一次任务读取和更新都施加授权检查，限制请求频率，约束请求体大小，支持幂等重试，对 provider 错误做脱敏，并为 prompt 和生成结果定义保留策略。

如何存储会话并实施用户归属校验？

每条会话都带上由服务端推导出的 tenant_id 和 owner_id 存储，再把会话标识带入它的消息和预测任务。每一次读取、更新、流式传输和删除操作都必须带上已认证的归属范围；知道一个会话 ID 并不等于获得授权。

Copied to ClipboardError: Could not Copy

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 中用经过审批的策略（例如 VPD）强制执行租户谓词，并测试一个用户无法读取、流式获取、更新或删除另一个用户的会话。

对于聊天机器人风格的端点，存储会话和消息时带上所有者或租户标识，并在每一个返回上下文、历史或生成内容的端点上强制校验这一归属关系。同时还要考虑 prompt 注入、不安全的生成输出、非预期的工具调用、敏感数据泄露和成本耗尽。用严格的 schema 校验、工具白名单、最小权限数据访问、超时、token 预算、脱敏，以及对有重大影响的操作引入人工审批来缓解这些风险。

如何使用 FastAPI 添加用户身份验证 `Depends`？

在创建推理任务或调用提供商之前，先注入一个经过验证的身份对象。从校验通过的 JWT、OAuth 会话或等效凭证中推导出 tenant_id 和 user_id；不要把它们当作请求体中的权威字段来接受。

Copied to ClipboardError: Could not Copy

from fastapi import Depends, HTTPException, status
from pydantic import BaseModel

class VerifiedIdentity(BaseModel):
 tenant_id: str
 user_id: str

async def require_identity() -> VerifiedIdentity:
 identity = await verify_request_credentials()
 if identity is None:
 raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
 return identity

@app.post("/v1/chat")
async def chat(request: ChatRequest,
 identity: VerifiedIdentity = Depends(require_identity)):
 # Scope reads and writes with identity.tenant_id and identity.user_id.
 return await create_scoped_job(request, identity)

这个依赖项仅作示例。请将 verify_request_credentials() 替换为该部署所批准的认证机制，并在数据库策略中强制实施同一身份。

身份验证与授权： 要求调用方经过身份验证，并将每个任务限定到某个租户或用户。不透明的预测 ID 不等于授权。
速率限制： 进程本地的防护只适用于演示。在多实例部署中，应通过 API 网关、基于 Redis 的服务或等效的平台控制来实施分布式限流。
输入控制： 限制提示长度，拒绝未知的 JSON 字段，并将媒体上传与文本请求分开校验。
幂等性： 对于可能在网络故障后重试 POST 的客户端，添加幂等键。
可观测性： 返回请求 ID，记录任务 ID 和租户安全的关联数据，并测量队列延迟、推理耗时、错误和放弃情况。
保留策略： 明确提示和生成响应的过期或必须删除的时间；默认不要无限期保留它们。

Oracle AI Database 模式：租户范围与请求关联

API 必须先对调用方进行身份验证，然后才能打开数据库会话。以下是简化后的 Oracle 伪代码，基于应用程序上下文和 Oracle Virtual Private Database (VPD) 以及 DBMS_APPLICATION_INFO。它不是可直接复制粘贴的生产策略：请配置受信任的上下文设置包，并针对目标部署审查 VPD 的许可和安全要求。实现 VPD 的 DBMS_RLS 仅在 Oracle AI Database 企业版中提供；采用此模式前，请确认版本和部署控制措施。

已复制到剪贴板错误：无法复制

-- The API derives these values from authenticated identity; do not accept a
-- tenant ID or client identity as an untrusted request-body field.
BEGIN
 trusted_ai_api_context.set_identity(
 tenant_id => :authenticated_tenant_id,
 client_id => :authenticated_client_id
 );

 -- Safe correlation only: do not place prompts, tokens, or secrets here.
 DBMS_APPLICATION_INFO.SET_MODULE(
 module_name => 'ai_predictions_api',
 action_name => 'create_prediction'
 );
 DBMS_APPLICATION_INFO.SET_CLIENT_INFO(:request_id);
END;
/

-- A VPD policy function can add this predicate whenever PREDICTION_JOBS is
-- queried or changed. The application query stays simple; the database
-- enforces the current tenant scope.
CREATE OR REPLACE FUNCTION prediction_tenant_predicate(
 object_schema VARCHAR2,
 object_name VARCHAR2
) RETURN VARCHAR2 AS
BEGIN
 RETURN q'[tenant_id = SYS_CONTEXT('AI_API_CTX', 'TENANT_ID')]';
END;
/

BEGIN
 DBMS_RLS.ADD_POLICY(
 object_schema => 'APP_SCHEMA', -- replace with the schema that owns PREDICTION_JOBS
 object_name => 'PREDICTION_JOBS',
 policy_name => 'PREDICTION_TENANT_SCOPE',
 policy_function => 'PREDICTION_TENANT_PREDICATE',
 statement_types => 'SELECT,INSERT,UPDATE,DELETE',
 update_check => TRUE
 );
END;
/

-- Idempotency belongs in durable state, not only in the API process.
-- The PREDICTION_JOBS table definition above already enforces it with
-- UNIQUE (tenant_id, idempotency_key); do not add the constraint twice.

用 DBMS_APPLICATION_INFO 把安全请求 ID 与数据库操作关联起来；它的 CLIENT_INFO 字段不是存放密钥或敏感提示词数据的地方。对于使用 VPD 的系统，策略函数挂在受保护的表上，Oracle 会把谓词应用到已配置的语句类型上。把整个 SQL 部分当作示意性的架构指导，并针对目标版本、schema 设计、会话池行为和身份模型做一次评审。

什么时候该用 FastAPI、ORDS、流式传输还是队列？

面向应用的校验和模型编排用 FastAPI，受治理的数据库自有资源用 ORDS，增量式交互输出用 SSE，生产环境的 worker 投递用持久化队列。这些选择可以共存，但它们解决的是不同边界上的问题。

需求	最合适的边界	示例或资源
短时、可预测的推理	同步 REST 响应	带输入上限和超时的直接路由
长时间运行的推理	`202` 作业资源加 worker	上文描述的作业资源与 worker 模式
逐 token 交互	SSE 或其他流式协议	Total Recall appbook，作为进阶路径
数据库自有的状态资源	ORDS	Oracle REST Data Services 文档
检索增强的应用	独立的检索 API	From RAG to Agents 工作坊

当数据库自有的资源需要一套受治理的 REST 接口时，ORDS 很有用。它不必取代 FastAPI 作为模型编排的公共 API 边界。在这种模式里，FastAPI 负责请求校验和模型生命周期；当现有的数据库侧应用确实需要时，ORDS 可以为作业状态数据提供一个范围很窄的只读集成。

如何测试和部署 AI REST API？

部署前要测试公共契约、认证、租户隔离、幂等性、限流、worker 状态流转、provider 故障以及 OpenAPI 输出。生产环境中把 worker 单独运行，并把 provider 和数据库凭据配置在源码管理之外。

先看现有的 Total Recall appbook，了解 FastAPI/SSE 的边界处理，再看 From RAG to Agents workshop，掌握基于数据库的编排模式。
用有界的 Pydantic schema 和带认证的身份依赖实现资源契约。
在暴露 provider 调用之前，先加上持久化任务状态、租户隔离、幂等性和 worker。
通过经过审核的 repository 实现，接入 Oracle AI Database 持久化和连接池。
部署前跑完契约、认证、租户隔离、流式、worker 和 provider adapter 测试。

OpenAPI 契约应位于 /openapi.json。用它生成客户端 SDK、跑契约测试，并在集成期间核对请求/响应 schema。开发阶段可以用 FastAPI 的 /docs 页面查看端点和已验证的模型，但这两个端点都不是安全控制：授权、租户隔离和限流仍必须在 API 逻辑和网关策略中强制执行。

AI REST API 排障

401 unauthorized： 确认 API 从经过审核的凭证中获取身份，且客户端提供了所需的凭证。
422 Unprocessable Entity： 检查 prompt 长度、temperature 范围、JSON 字段名和 Content-Type: application/json 请求头。
409 idempotency_conflict： 只有在逻辑请求体完全相同时才复用 Idempotency-Key；请求内容变了就生成新 key。
轮询时出现 404 not_found： 状态查询要限定在创建该任务的同一认证租户范围内。
重启后任务消失： 把已接受的任务写入持久化状态，通过独立的 worker 或队列处理。
/docs 能加载但请求失败： OpenAPI 描述的是契约，不提供认证或配置。检查请求凭证和 JSON 错误码。

生产环境检查清单

FAQ

如何为 AI 应用构建 REST API？

用 FastAPI 校验和认证请求，持久化推理任务，返回 202 Accepted 和状态 URL，由 worker 调用模型 provider。任务状态存入 Oracle AI Database 这类持久化数据库，读取时限定在认证租户范围内，并返回稳定的 JSON 状态和错误响应。

AI REST API 的最佳实践是什么？

核心实践包括：有边界的输入校验、服务端完成提供商认证、限流、持久化的异步任务、租户感知的授权、幂等性、结构化错误、请求关联、留存控制，以及针对失败路径的测试——而不只是测试推理成功的情况。

如何为基于 OpenAI 的 REST API 做认证？

在服务边界认证 API 调用方，然后在服务端或 worker 中单独向模型提供商认证。把提供商密钥放在密钥管理器或受保护的环境变量里，显式指定模型，记录一个安全的提供商请求 ID，不要把密钥或提供商的原始错误返回给客户端。

AI 应用的 API 该用 FastAPI 还是 ORDS？

面向应用的校验、认证、编排和模型生命周期管理用 FastAPI。当某个由数据库拥有的资源需要受管控的 REST 接口时，用 ORDS。两者可以共存；ORDS 不必取代 FastAPI 作为对外的模型编排边界。

AI REST API 应该同步返回结果吗？

对于可预测的短操作可以。如果某个操作在请求窗口内可能无法可靠完成，或者你需要持久化的重试、审计和状态查询，就用任务资源和 202 Accepted。

FastAPI 足以支撑生产级 AI 应用吗？

FastAPI 是一个好用的 HTTP 框架，但它只是其中一层。生产环境还需要持久化存储、worker 或队列、认证、限流、可观测性和留存控制。

针对 AI REST API，我应该跑哪个仍在维护的验证案例？

FastAPI 和 SSE 跑 Total Recall，检索和编排跑 From RAG to Agents workshop，身份感知且受管控的数据访问跑 Enterprise Data Agent workshop。选那个与你需要优先验证的边界相匹配的案例。

Oracle AI Database 对 AI API 有什么帮助？

它可以把任务状态和应用数据持久化在同一个受管控的数据平台里。应用 API 仍然负责校验和编排请求；Oracle AI Database 在其背后提供持久化状态和数据访问层。

下一步

选择一个仍在维护的示例项目，然后套用本指南中的 API 契约：

要点

你将设计和验证什么？

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

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

FastAPI 中的聊天机器人 REST API 应该暴露哪些端点？

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

验证案例：把 API 模式套用到真实的 AI 开发者任务上

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

任务：让检索支撑的 AI API 基于正确的证据作答

任务：在不绕过身份和访问策略的前提下开放企业数据

如何让公开 API 独立于 AI 供应商？

开发期间 FastAPI /docs 和 /openapi.json 有什么帮助？

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

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

如何用 FastAPI 构建 AI REST API？

第 1 步：定义有边界的请求 schema

第 2 步：提交任务并返回 202 Accepted

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

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

第 4 步：让 worker 处理模型推理

如何用 OpenAI SDK 搭建一个最小化的 FastAPI 聊天接口？

如何把 REST API 接到 OpenAI Responses API？

应该如何安全地存储和加载 OpenAI API key？

如何对 AI REST API 做认证和安全防护？

如何存储会话并实施用户归属校验？

如何使用 FastAPI 添加用户身份验证 Depends？

Oracle AI Database 模式：租户范围与请求关联

什么时候该用 FastAPI、ORDS、流式传输还是队列？

如何测试和部署 AI REST API？

AI REST API 排障

生产环境检查清单

FAQ

如何为 AI 应用构建 REST API？

AI REST API 的最佳实践是什么？

如何为基于 OpenAI 的 REST API 做认证？

AI 应用的 API 该用 FastAPI 还是 ORDS？

AI REST API 应该同步返回结果吗？

FastAPI 足以支撑生产级 AI 应用吗？

针对 AI REST API，我应该跑哪个仍在维护的验证案例？

Oracle AI Database 对 AI API 有什么帮助？

下一步

文档

作者

Wojtek Pluta

技术产品营销总监

Portable Oracle Select AI Agent Teams：从开发到生产

用 Oracle AI Database 构建可控的 Cursor MCP 工作流

开发期间 FastAPI `/docs` 和 `/openapi.json` 有什么帮助？

第 2 步：提交任务并返回 `202 Accepted`

如何使用 FastAPI 添加用户身份验证 `Depends`？