Python AI Agent实战：从零构建高可用智能代理系统

1次阅读

共计 1720 个字符，预计需要花费 5 分钟才能阅读完成。

在实时交互场景中，AI Agent 面临几个典型挑战：

对话状态保持 ：传统的无状态 HTTP 请求难以处理多轮对话，每次交互都需要重建上下文，导致响应延迟增加。
多轮决策一致性 ：当用户意图需要分步骤完成时（如订票流程），同步阻塞式处理会导致超时风险。

我们对比了同步 Flask 和异步 FastAPI 的实现（压测工具：locust）：

架构类型	QPS (并发 100)	平均延迟
同步	128	780ms
异步	421	230ms

采用三层设计：

API 网关层 ：FastAPI 处理 HTTP/WebSocket
状态管理层 ：Redis 存储会话数据
算法执行层 ：策略模式实现模块替换

from fastapi import BackgroundTasks

async def async_inference(task_id: str):
    # 模拟耗时操作
    await asyncio.sleep(0.1)
    return {"result": f"task_{task_id}_completed"}

@app.post("/tasks")
async def create_task(bg: BackgroundTasks):
    task_id = str(uuid4())
    bg.add_task(async_inference, task_id)
    return {"task_id": task_id}

import redis.asyncio as redis

r = redis.from_url("redis://localhost")

async def log_interaction(session_id: str, event: dict):
    await r.xadd(f"session:{session_id}", event)

from abc import ABC, abstractmethod
from functools import lru_cache

class BaseAgent(ABC):
    @lru_cache(maxsize=1000)
    def cached_llm_call(self, prompt: str) -> str:
        # 实际调用 LLM 的代码
        return self.llm(prompt)

    @abstractmethod
    async def execute(self, input_data: dict) -> dict:
        pass

队列类型	平均延迟	峰值内存占用
Redis	8ms	120MB
RabbitMQ	15ms	210MB
Kafka	4ms	350MB

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def sanitize_prompt(text: str) -> str:
    # 防御 Prompt 注入的基础实现
    return re.sub(r"[^\w\s,.?!]", "", text)

from prometheus_client import Counter

REQUEST_COUNT = Counter('agent_requests', 'API call count')

@app.middleware("http")
async def monitor_requests(request, call_next):
    REQUEST_COUNT.inc()
    return await call_next(request)

# docker-compose.yml 示例
services:
  agent:
    cpus: 2
    mem_limit: 1G
    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 1G