تقديم الوكلاء خلف 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)

المصادقة

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)):
    ...

الاستجابات المتدفقة

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

التحقق من الطلبات

from pydantic import Field

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

معالجة الأخطاء

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)

التزامن

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

شغّله باستخدام عدة عمال

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

الوكيل العكسي

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=['*']
)

عنصر في قائمة التحقق للإنتاج

مراجعة