AI Agents · 课时

通过 API 提供智能体服务

使用 FastAPI 封装智能体，加入身份验证、请求验证，并通过 SSE 或 WebSockets 进行流式传输。

通过 API 提供智能体服务 是 CoddyKit 上的免费 AI Agents 课时。这是第 1 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

本课时的部分内容尚未翻译，以英文显示。

从笔记本到服务

在笔记本中运行正常的智能体只是起点。要投入使用，请将其封装成其他服务可以调用的 HTTP API。

FastAPI Skeleton

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class AgentRequest(BaseModel):
    query: str
    user_id: str

class AgentResponse(BaseModel):
    answer: str
    trace_id: str

@app.post('/agents/qa', response_model=AgentResponse)
def qa(req: AgentRequest):
    trace_id = run_agent(req.query, req.user_id)
    return AgentResponse(answer=result.answer, trace_id=trace_id)

身份验证

每个智能体端点都必须进行身份验证。请根据调用方类型使用 API 密钥或 OAuth：

from fastapi import Header, HTTPException

def require_api_key(api_key: str = Header(alias='X-API-Key')):
    if not is_valid(api_key):
        raise HTTPException(401, 'Invalid API key')
    return api_key

@app.post('/agents/qa')
def qa(req: AgentRequest, key=Depends(require_api_key)):
    ...

流式响应

对于聊天界面，请使用 SSE 流式传输词元：

from fastapi.responses import StreamingResponse

@app.post('/agents/qa/stream')
def qa_stream(req: AgentRequest):
    def gen():
        for chunk in stream_agent(req.query):
            yield f'data: {chunk}\n\n'
        yield 'data: [DONE]\n\n'
    return StreamingResponse(gen(), media_type='text/event-stream')

请求验证

Pydantic 强制执行类型和基本验证。请根据需要添加自定义验证：

from pydantic import Field

class AgentRequest(BaseModel):
    query: str = Field(min_length=1, max_length=10_000)
    user_id: str
    options: dict = {}

错误处理

将异常映射为简洁的 HTTP 响应：

from fastapi import HTTPException

@app.exception_handler(BudgetExceeded)
def budget_handler(req, exc):
    return JSONResponse({'error': 'budget-exceeded'}, status_code=429)

@app.exception_handler(Exception)
def catch_all(req, exc):
    log.exception('Unhandled')
    return JSONResponse({'error': 'internal', 'trace_id': get_trace_id()}, status_code=500)

并发

FastAPI 默认使用异步模式。对于 CPU 密集型智能体（很少见）或会阻塞的 SDK，请使用异步客户端：

from openai import AsyncOpenAI
client = AsyncOpenAI()

async def run_agent_async(query):
    return await client.chat.completions.create(...)

Health and Readiness Endpoints

@app.get('/healthz')
def healthz():
    return {'status': 'ok'}

@app.get('/readyz')
def readyz():
    if not vector_db.is_healthy():
        raise HTTPException(503)
    return {'status': 'ready'}

Containerise

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ['uvicorn', 'main:app', '--host', '0.0.0.0', '--port', '8080']

使用多个工作进程运行

用于生产环境的 Uvicorn + Gunicorn：

gunicorn -k uvicorn.workers.UvicornWorker -w 4 main:app

反向代理

请在前面部署 Nginx 或 Caddy，以处理 TLS、gzip 和 IP 允许列表。

Logging Structured Output

import structlog
log = structlog.get_logger()

@app.middleware('http')
async def log_requests(request, call_next):
    response = await call_next(request)
    log.info('http', method=request.method, path=request.url.path, status=response.status_code)
    return response

CORS for Browser Clients

from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=['https://app.example.com'],
    allow_methods=['*'], allow_headers=['*']
)

生产环境检查项

对于任何生产环境中的代理接口，哪一项是必需的？

回顾

FastAPI + Pydantic + 异步客户端 + 流式传输 + 身份验证 + 结构化日志 + 健康探针。这就是生产环境的最低配置。

常见问题解答

「通过 API 提供智能体服务」课时是免费的吗？

是的 — 「通过 API 提供智能体服务」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「通过 API 提供智能体服务」这节课中我会学到什么？

使用 FastAPI 封装智能体，加入身份验证、请求验证，并通过 SSE 或 WebSockets 进行流式传输。你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 1 节课，共 4 节。

「通过 API 提供智能体服务」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Agents