阶段 7：工程化、评估与安全

📖 本阶段目标

掌握 Agent 的工程化最佳实践
学习 API 服务化和部署
理解评估指标和测试方法
掌握安全和隐私保护措施

🏗️ 工程化架构

生产级 Agent 架构

┌─────────────────────────────────────────────┐
│              API Gateway                     │
│         (认证、限流、路由)                   │
└──────────────────┬──────────────────────────┘
                   │
┌──────────────────┴──────────────────────────┐
│           Agent Service Layer                │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│  │ Agent 1  │  │ Agent 2  │  │ Agent N  │  │
│  └──────────┘  └──────────┘  └──────────┘  │
└──────────────────┬──────────────────────────┘
                   │
┌──────────────────┴──────────────────────────┐
│         Infrastructure Layer                 │
│  ┌──────┐  ┌──────┐  ┌────────┐  ┌──────┐ │
│  │Cache │  │ DB   │  │Vector  │  │Queue │ │
│  │Redis │  │Postgr│  │Store   │  │RabbitM│ │
│  └──────┘  └──────┘  └────────┘  └──────┘ │
└─────────────────────────────────────────────┘

🌐 API 服务化

1. FastAPI 实现

# examples/stage7/agent_api.py
from fastapi import FastAPI, HTTPException, Depends, Header
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, List
import uvicorn
from datetime import datetime
import os

app = FastAPI(
    title="AI Agent API",
    description="生产级 AI Agent 服务",
    version="1.0.0"
)

# CORS 配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境应限制来源
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 请求/响应模型
class ChatRequest(BaseModel):
    message: str
    session_id: Optional[str] = None
    user_id: Optional[str] = None

class ChatResponse(BaseModel):
    response: str
    session_id: str
    timestamp: str

class AgentStatus(BaseModel):
    status: str
    version: str
    uptime: str

# 认证依赖
async def verify_api_key(x_api_key: str = Header(...)):
    """验证 API Key"""
    valid_keys = os.getenv("VALID_API_KEYS", "").split(",")
    if x_api_key not in valid_keys:
        raise HTTPException(status_code=401, detail="Invalid API Key")
    return x_api_key

# 简单的 Agent（实际应该更复杂）
class SimpleAgent:
    def __init__(self):
        self.sessions = {}

    def chat(self, message: str, session_id: str) -> str:
        # 这里应该是真实的 Agent 逻辑
        if session_id not in self.sessions:
            self.sessions[session_id] = []

        self.sessions[session_id].append({
            "role": "user",
            "content": message
        })

        # 模拟响应
        response = f"收到消息: {message}"

        self.sessions[session_id].append({
            "role": "assistant",
            "content": response
        })

        return response

agent = SimpleAgent()

# API 端点
@app.get("/")
async def root():
    """根路径"""
    return {"message": "AI Agent API is running"}

@app.get("/health")
async def health_check():
    """健康检查"""
    return {"status": "healthy", "timestamp": datetime.now().isoformat()}

@app.post("/chat", response_model=ChatResponse)
async def chat(
    request: ChatRequest,
    api_key: str = Depends(verify_api_key)
):
    """对话端点"""
    try:
        session_id = request.session_id or f"session_{datetime.now().timestamp()}"

        response = agent.chat(request.message, session_id)

        return ChatResponse(
            response=response,
            session_id=session_id,
            timestamp=datetime.now().isoformat()
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/status", response_model=AgentStatus)
async def get_status():
    """获取 Agent 状态"""
    return AgentStatus(
        status="running",
        version="1.0.0",
        uptime="24h"
    )

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

2. 日志和监控

# examples/stage7/logging_config.py
import logging
from logging.handlers import RotatingFileHandler
import json
from datetime import datetime

class StructuredLogger:
    """结构化日志"""

    def __init__(self, name: str):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.INFO)

        # 文件处理器
        file_handler = RotatingFileHandler(
            'agent.log',
            maxBytes=10*1024*1024,  # 10MB
            backupCount=5
        )

        # 控制台处理器
        console_handler = logging.StreamHandler()

        # 格式化器
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )

        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)

        self.logger.addHandler(file_handler)
        self.logger.addHandler(console_handler)

    def log_request(self, user_id: str, message: str):
        """记录请求"""
        log_data = {
            "timestamp": datetime.now().isoformat(),
            "type": "request",
            "user_id": user_id,
            "message": message
        }
        self.logger.info(json.dumps(log_data, ensure_ascii=False))

    def log_response(self, user_id: str, response: str, latency: float):
        """记录响应"""
        log_data = {
            "timestamp": datetime.now().isoformat(),
            "type": "response",
            "user_id": user_id,
            "response": response,
            "latency_ms": latency
        }
        self.logger.info(json.dumps(log_data, ensure_ascii=False))

    def log_error(self, error: Exception, context: dict):
        """记录错误"""
        log_data = {
            "timestamp": datetime.now().isoformat(),
            "type": "error",
            "error": str(error),
            "error_type": type(error).__name__,
            "context": context
        }
        self.logger.error(json.dumps(log_data, ensure_ascii=False))

# 使用示例
logger = StructuredLogger("agent_service")

def process_request(user_id: str, message: str):
    """处理请求示例"""
    import time

    logger.log_request(user_id, message)

    start_time = time.time()
    try:
        # 处理逻辑
        response = f"处理: {message}"

        latency = (time.time() - start_time) * 1000
        logger.log_response(user_id, response, latency)

        return response
    except Exception as e:
        logger.log_error(e, {"user_id": user_id, "message": message})
        raise

3. 性能优化

# examples/stage7/performance.py
from functools import lru_cache
import asyncio
from typing import List
import time

class PerformanceOptimizer:
    """性能优化工具"""

    @staticmethod
    @lru_cache(maxsize=1000)
    def cached_embedding(text: str):
        """缓存 Embedding 结果"""
        # 实际应该调用 Embedding API
        time.sleep(0.1)  # 模拟 API 调用
        return f"embedding_{hash(text)}"

    @staticmethod
    async def batch_process(items: List[str], batch_size: int = 10):
        """批量处理"""
        results = []

        for i in range(0, len(items), batch_size):
            batch = items[i:i + batch_size]

            # 并行处理批次
            batch_results = await asyncio.gather(*[
                asyncio.create_task(process_item(item))
                for item in batch
            ])

            results.extend(batch_results)

        return results

    @staticmethod
    def rate_limiter(max_calls: int, time_window: int):
        """速率限制装饰器"""
        from collections import deque
        from time import time

        calls = deque()

        def decorator(func):
            def wrapper(*args, **kwargs):
                now = time()

                # 清理过期的调用记录
                while calls and calls[0] < now - time_window:
                    calls.popleft()

                # 检查是否超过限制
                if len(calls) >= max_calls:
                    raise Exception("Rate limit exceeded")

                calls.append(now)
                return func(*args, **kwargs)

            return wrapper
        return decorator

async def process_item(item: str):
    """处理单个项目"""
    await asyncio.sleep(0.1)  # 模拟处理
    return f"processed_{item}"

# 使用示例
@PerformanceOptimizer.rate_limiter(max_calls=10, time_window=60)
def api_call():
    print("API called")

if __name__ == "__main__":
    # 测试缓存
    print(PerformanceOptimizer.cached_embedding("test"))
    print(PerformanceOptimizer.cached_embedding("test"))  # 从缓存读取

    # 测试批量处理
    items = [f"item_{i}" for i in range(25)]
    results = asyncio.run(PerformanceOptimizer.batch_process(items))
    print(f"处理了 {len(results)} 个项目")

🔒 安全与隐私

1. 输入验证和清理

# examples/stage7/security.py
import re
from typing import Optional

class SecurityValidator:
    """安全验证器"""

    @staticmethod
    def validate_input(user_input: str, max_length: int = 1000) -> tuple[bool, Optional[str]]:
        """验证用户输入"""
        # 1. 长度检查
        if len(user_input) > max_length:
            return False, f"输入超过最大长度 {max_length}"

        # 2. 检查恶意模式
        dangerous_patterns = [
            r"<script",  # XSS
            r"javascript:",
            r"onerror=",
            r"onclick=",
            r"';.*--",  # SQL 注入
            r"union\s+select",
        ]

        for pattern in dangerous_patterns:
            if re.search(pattern, user_input, re.IGNORECASE):
                return False, f"检测到潜在恶意内容"

        return True, None

    @staticmethod
    def sanitize_output(output: str) -> str:
        """清理输出"""
        # 移除潜在的敏感信息
        patterns = [
            (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]'),
            (r'\b\d{3}-\d{2}-\d{4}\b', '[SSN]'),
            (r'\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b', '[CARD]'),
        ]

        sanitized = output
        for pattern, replacement in patterns:
            sanitized = re.sub(pattern, replacement, sanitized)

        return sanitized

    @staticmethod
    def check_prompt_injection(user_input: str) -> bool:
        """检查提示词注入攻击"""
        injection_patterns = [
            r"ignore\s+(previous|above)\s+instructions",
            r"you\s+are\s+(now|a)\s+",
            r"system\s*:",
            r"</?\s*system\s*>",
        ]

        for pattern in injection_patterns:
            if re.search(pattern, user_input, re.IGNORECASE):
                return True

        return False

# 使用示例
def safe_agent_call(user_input: str):
    """安全的 Agent 调用"""
    validator = SecurityValidator()

    # 1. 验证输入
    is_valid, error_msg = validator.validate_input(user_input)
    if not is_valid:
        return f"输入验证失败: {error_msg}"

    # 2. 检查提示词注入
    if validator.check_prompt_injection(user_input):
        return "检测到潜在的提示词注入攻击"

    # 3. 调用 Agent（这里模拟）
    response = f"处理结果: {user_input}"

    # 4. 清理输出
    sanitized_response = validator.sanitize_output(response)

    return sanitized_response

if __name__ == "__main__":
    # 测试
    print(safe_agent_call("正常的输入"))
    print(safe_agent_call("ignore previous instructions"))
    print(safe_agent_call("我的邮箱是 test@example.com"))

2. 访问控制

# examples/stage7/access_control.py
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import hashlib
import secrets

class AccessControl:
    """访问控制系统"""

    def __init__(self):
        self.users = {}  # user_id -> user_info
        self.api_keys = {}  # api_key -> user_id
        self.rate_limits = {}  # user_id -> [timestamps]

    def create_user(self, user_id: str, role: str = "user") -> str:
        """创建用户并生成 API Key"""
        api_key = self._generate_api_key()

        self.users[user_id] = {
            "role": role,
            "created_at": datetime.now(),
            "rate_limit": 100 if role == "user" else 1000
        }

        self.api_keys[api_key] = user_id

        return api_key

    def _generate_api_key(self) -> str:
        """生成 API Key"""
        random_bytes = secrets.token_bytes(32)
        return hashlib.sha256(random_bytes).hexdigest()

    def verify_api_key(self, api_key: str) -> Optional[str]:
        """验证 API Key"""
        return self.api_keys.get(api_key)

    def check_rate_limit(self, user_id: str) -> bool:
        """检查速率限制"""
        if user_id not in self.users:
            return False

        user_info = self.users[user_id]
        max_requests = user_info["rate_limit"]

        # 获取最近1小时的请求
        now = datetime.now()
        if user_id not in self.rate_limits:
            self.rate_limits[user_id] = []

        # 清理过期记录
        self.rate_limits[user_id] = [
            ts for ts in self.rate_limits[user_id]
            if now - ts < timedelta(hours=1)
        ]

        # 检查是否超限
        if len(self.rate_limits[user_id]) >= max_requests:
            return False

        # 记录本次请求
        self.rate_limits[user_id].append(now)
        return True

    def has_permission(self, user_id: str, resource: str, action: str) -> bool:
        """检查权限"""
        if user_id not in self.users:
            return False

        role = self.users[user_id]["role"]

        # 简单的基于角色的访问控制
        permissions = {
            "admin": ["read", "write", "delete"],
            "user": ["read", "write"],
            "guest": ["read"]
        }

        return action in permissions.get(role, [])

# 使用示例
if __name__ == "__main__":
    ac = AccessControl()

    # 创建用户
    api_key_user = ac.create_user("user_001", role="user")
    api_key_admin = ac.create_user("admin_001", role="admin")

    print(f"User API Key: {api_key_user}")
    print(f"Admin API Key: {api_key_admin}")

    # 验证 API Key
    user_id = ac.verify_api_key(api_key_user)
    print(f"\nVerified user: {user_id}")

    # 检查权限
    print(f"User can write: {ac.has_permission('user_001', 'document', 'write')}")
    print(f"User can delete: {ac.has_permission('user_001', 'document', 'delete')}")
    print(f"Admin can delete: {ac.has_permission('admin_001', 'document', 'delete')}")

    # 检查速率限制
    print(f"\nRate limit ok: {ac.check_rate_limit('user_001')}")

📊 评估与测试

1. 单元测试

# examples/stage7/test_agent.py
import unittest
from unittest.mock import Mock, patch

class TestAgent(unittest.TestCase):
    """Agent 单元测试"""

    def setUp(self):
        """测试前准备"""
        self.agent = SimpleAgent()  # 假设的 Agent 类

    def test_basic_response(self):
        """测试基本响应"""
        response = self.agent.chat("Hello")
        self.assertIsNotNone(response)
        self.assertIsInstance(response, str)

    def test_tool_calling(self):
        """测试工具调用"""
        with patch('agent.get_weather') as mock_weather:
            mock_weather.return_value = "晴天，25°C"

            response = self.agent.chat("北京天气怎么样？")

            mock_weather.assert_called_once()
            self.assertIn("25", response)

    def test_memory(self):
        """测试记忆功能"""
        self.agent.chat("我叫张三")
        response = self.agent.chat("我叫什么名字？")

        self.assertIn("张三", response)

    def test_error_handling(self):
        """测试错误处理"""
        with self.assertRaises(ValueError):
            self.agent.chat("")  # 空输入应该抛出异常

class SimpleAgent:
    """简单的 Agent 实现（用于测试）"""
    def __init__(self):
        self.memory = []

    def chat(self, message: str) -> str:
        if not message:
            raise ValueError("Message cannot be empty")

        self.memory.append(message)
        return f"Response to: {message}"

if __name__ == "__main__":
    unittest.main()

2. 性能测试

# examples/stage7/performance_test.py
import time
import statistics
from typing import List

class PerformanceTest:
    """性能测试"""

    @staticmethod
    def measure_latency(func, num_runs: int = 100) -> dict:
        """测量延迟"""
        latencies = []

        for _ in range(num_runs):
            start = time.time()
            func()
            end = time.time()
            latencies.append((end - start) * 1000)  # 转换为毫秒

        return {
            "mean": statistics.mean(latencies),
            "median": statistics.median(latencies),
            "p95": sorted(latencies)[int(num_runs * 0.95)],
            "p99": sorted(latencies)[int(num_runs * 0.99)],
            "min": min(latencies),
            "max": max(latencies)
        }

    @staticmethod
    def load_test(func, duration_seconds: int = 60, target_rps: int = 10):
        """负载测试"""
        import asyncio

        async def run_requests():
            start_time = time.time()
            request_count = 0
            errors = 0

            while time.time() - start_time < duration_seconds:
                try:
                    await asyncio.create_task(asyncio.to_thread(func))
                    request_count += 1
                except Exception as e:
                    errors += 1

                # 控制速率
                await asyncio.sleep(1 / target_rps)

            return {
                "total_requests": request_count,
                "errors": errors,
                "success_rate": (request_count - errors) / request_count * 100,
                "actual_rps": request_count / duration_seconds
            }

        return asyncio.run(run_requests())

# 使用示例
def sample_agent_call():
    """示例 Agent 调用"""
    time.sleep(0.1)  # 模拟处理时间
    return "response"

if __name__ == "__main__":
    # 延迟测试
    print("延迟测试:")
    latency_stats = PerformanceTest.measure_latency(sample_agent_call, num_runs=100)
    for key, value in latency_stats.items():
        print(f"  {key}: {value:.2f} ms")

    # 负载测试
    print("\n负载测试 (10 秒):")
    load_stats = PerformanceTest.load_test(sample_agent_call, duration_seconds=10, target_rps=5)
    for key, value in load_stats.items():
        print(f"  {key}: {value}")

🚀 部署

Docker 化

# Dockerfile
FROM python:3.10-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["uvicorn", "agent_api:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'

services:
  agent-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - DATABASE_URL=postgresql://user:pass@db:5432/agentdb
    depends_on:
      - db
      - redis

  db:
    image: postgres:15
    environment:
      - POSTGRES_DB=agentdb
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7
    ports:
      - "6379:6379"

volumes:
  postgres_data:

✅ 阶段完成检查清单

[ ] 实现了 RESTful API 服务
[ ] 配置了日志和监控系统
[ ] 实现了安全验证和访问控制
[ ] 编写了单元测试和性能测试
[ ] 了解了部署和容器化
[ ] 掌握了性能优化技巧
[ ] 理解了安全最佳实践

🎉 恭喜完成学习路线！

你已经掌握了从基础到工程化的完整 AI Agent 开发技能。接下来：

持续实践：构建真实项目，解决实际问题
深入专研：选择感兴趣的方向深入学习
社区参与：分享经验，帮助他人
保持更新：关注最新技术和最佳实践

参考学习资源获取更多材料和社区链接。

💡 最后的建议：技术在变，但解决问题的思维方式是永恒的。祝你在 AI Agent 开发的道路上越走越远！🚀