第六章: 生产级Agent系统

系统设计

架构图

用户请求 → API Gateway → Agent调度器
                          ↓
              [Agent 1] [Agent 2] [Agent 3]
                ↓         ↓         ↓
           工具层：数据库、API、搜索引擎
                          ↓
                      结果返回

性能优化

1. 缓存策略

from functools import lru_cache

@lru_cache(maxsize=1000)
def expensive_search(query: str):
    # 缓存搜索结果
    return search_api(query)

2. 异步执行

import asyncio

async def run_agent_async(query):
    result = await agent_executor.ainvoke({"input": query})
    return result

监控与日志

日志记录

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AgentWithLogging:
    def run(self, query):
        logger.info(f"开始执行：{query}")
        result = self.agent.run(query)
        logger.info(f"完成：{result}")
        return result

性能监控

import time

def monitor_agent(func):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        duration = time.time() - start
        print(f"耗时：{duration:.2f}秒")
        return result
    return wrapper

安全性

1. 输入验证

def sanitize_input(user_input: str) -> str:
    # 移除危险字符
    dangerous = ["<script>", "DROP TABLE", "rm -rf"]
    for d in dangerous:
        if d in user_input:
            raise ValueError("检测到危险输入")
    return user_input

2. 权限控制

ALLOWED_TOOLS = {
    "basic_user": [search_tool],
    "premium_user": [search_tool, database_tool],
    "admin": [search_tool, database_tool, system_tool]
}

def get_tools_for_user(user_role):
    return ALLOWED_TOOLS.get(user_role, [])

部署方案

Docker化

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["python", "agent_server.py"]

API服务

from fastapi import FastAPI

app = FastAPI()

@app.post("/agent")
async def run_agent(query: str):
    result = await agent_executor.ainvoke({"input": query})
    return {"result": result["output"]}

总结：从概念到生产，你已掌握AI Agent完整技术栈。继续探索，构建改变世界的智能应用！