后端服务与 Agent 集成模式

Agent 后端的技术栈全景

一个生产级 Agent 应用的后端不只是调用 LLM API，它需要解决：会话管理、记忆持久化、工具网关、流式响应、多租户隔离、安全认证等一整套工程问题。

┌─────────────────────────────────────────────────┐
│                   API Gateway                     │
│    认证(JWT/API Key) + 限流 + 路由 + CORS         │
├─────────────────────────────────────────────────┤
│                  Agent Service                    │
│  ┌──────────┐  ┌──────────┐  ┌───────────────┐  │
│  │ Orchestrator│  │ Memory   │  │ Tool Gateway  │  │
│  │ (编排引擎) │  │ (记忆层) │  │ (工具网关)    │  │
│  └──────────┘  └──────────┘  └───────────────┘  │
├─────────────────────────────────────────────────┤
│                 Infrastructure                    │
│  ┌──────────┐  ┌──────────┐  ┌───────────────┐  │
│  │ PostgreSQL│  │ Vector DB│  │ Redis (Cache)  │  │
│  └──────────┘  └──────────┘  └───────────────┘  │
└─────────────────────────────────────────────────┘

模式一：同步 Request-Response（最简单）

适合简单问答、文档总结等不需要流式输出的场景。

@app.post("/chat")
async def chat(request: ChatRequest):
    agent = AgentOrchestrator(session_id=request.session_id)
    result = await agent.run(request.message)
    return {"reply": result.content, "tools_used": result.tool_calls}

优点：实现简单、调试方便
缺点：长任务会超时、用户等待焦虑

模式二：SSE 流式响应（推荐）

适合需要实时反馈的场景，打字机效果 + 工具调用进度可见。

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    return StreamingResponse(
        agent.stream_run(request.message, request.session_id),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no"  # Nginx 禁用缓冲
        }
    )

# Agent 内部用 async generator
async def stream_run(self, message, session_id):
    yield f"data: {json.dumps({'type': 'start'})}\n\n"
    async for event in self.orchestrator.stream(message):
        if event.type == "llm_token":
            yield f"data: {json.dumps({'type': 'token', 'content': event.token})}\n\n"
        elif event.type == "tool_call":
            yield f"data: {json.dumps({'type': 'tool', 'name': event.tool_name})}\n\n"
        elif event.type == "done":
            yield f"data: {json.dumps({'type': 'done'})}\n\n"

事件类型设计

模式三：异步任务 + 回调（长任务）

适合代码生成、数据分析等耗时长的任务。提交任务后立即返回 task_id，通过轮询或 WebSocket 获取结果。

@app.post("/task/submit")
async def submit_task(request: TaskRequest):
    task_id = uuid4()
    background_tasks.add_task(run_agent_task, task_id, request)
    return {"task_id": task_id, "status": "pending"}

@app.get("/task/{task_id}/status")
async def task_status(task_id: str):
    return task_store.get(task_id)

会话管理

Agent 的会话管理与传统 Web 有本质区别——会话中不仅有对话历史，还有工具调用状态、上下文压缩、工作记忆。

class SessionManager:
    def __init__(self, redis: Redis, db: Database):
        self.redis = redis    # 热会话（最近 30min 活跃的）
        self.db = db          # 冷会话持久化

    async def get_or_create(self, user_id, session_id):
        # 1. 先从 Redis 取（热数据）
        session = await self.redis.get(f"session:{session_id}")
        if session:
            return Session.from_json(session)
        # 2. Redis 没有，从 DB 加载
        row = await self.db.fetch_one(
            "SELECT * FROM sessions WHERE id=$1 AND user_id=$2",
            session_id, user_id
        )
        # 3. 恢复到 Redis
        await self.cache_session(session)
        return session

数据库设计（核心表）

-- 用户表
CREATE TABLE users (
    id UUID PRIMARY KEY,
    api_key_hash TEXT NOT NULL,
    quota_limit INT DEFAULT 1000
);

-- 会话表
CREATE TABLE sessions (
    id UUID PRIMARY KEY,
    user_id UUID REFERENCES users(id),
    title TEXT,
    messages JSONB DEFAULT '[]',
    summary TEXT,
    token_count INT DEFAULT 0,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- 工具调用日志（可观测性）
CREATE TABLE tool_call_logs (
    id BIGSERIAL PRIMARY KEY,
    session_id UUID,
    tool_name TEXT,
    args JSONB,
    result JSONB,
    duration_ms INT,
    error TEXT,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_tool_logs_session ON tool_call_logs(session_id);

安全清单

• 所有 API 端点需要 JWT 或 API Key 认证
• 用户数据隔离：session_id/user_id 必须关联校验，防止越权访问
• 输入长度限制：message 长度上限（如 8000 字符），防止 Token 滥用
• 速率限制：每用户每分钟最多 N 次 LLM 调用
• 工具白名单：生产环境只注册安全审计过的工具
• Prompt 注入检测：用正则或分类器检测恶意 Prompt 模式

部署架构

• FastAPI + Uvicorn：最主流，异步支持好，SSE 原生支持
• Nginx 反向代理：SSL 终结、负载均衡、SSE 需要禁用缓冲
• Docker + HealthCheck：容器化部署，健康检查端点 /health
• Graceful Shutdown：收到 SIGTERM 后完成当前请求再退出

← 返回首页