⚙️ 模块3：性能优化与工程化处理

3.1 Token 消耗优化

优化策略	具体方法	预估节省
精简系统提示词	移除冗余描述、合并重复规则、用缩写替代长表述	15-30%
上下文压缩	用摘要替代原始对话，只保留关键信息	30-60%
分批处理	大文档分片处理，避免一次塞入所有内容	20-40%
缓存重复调用	相同问题返回缓存结果，不重复调用 API	视重复率而定
模型选择	简单任务用 Haiku（便宜），复杂任务用 Sonnet/Opus	50-80%

#!/usr/bin/env python3
"""TokenOptimizer — Token消耗监控与优化"""

import hashlib, json, time
from functools import lru_cache

class TokenMonitor:
    """Token 使用监控器"""

    def __init__(self):
        self.total_input_tokens = 0
        self.total_output_tokens = 0
        self.call_count = 0
        self.call_history: list[dict] = []

    def record(self, input_tokens: int, output_tokens: int, operation: str):
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens
        self.call_count += 1
        self.call_history.append({
            "op": operation, "in": input_tokens, "out": output_tokens,
            "time": time.time()
        })

    def report(self) -> str:
        total = self.total_input_tokens + self.total_output_tokens
        avg_in = self.total_input_tokens / max(self.call_count, 1)
        avg_out = self.total_output_tokens / max(self.call_count, 1)

        # 成本估算 (Claude Sonnet 参考价格)
        cost_estimate = (self.total_input_tokens / 1_000_000) * 3 + \
                        (self.total_output_tokens / 1_000_000) * 15

        return f"""
📊 Token 使用报告
{'='*40}
总调用次数: {self.call_count}
总输入Token: {self.total_input_tokens:,}
总输出Token: {self.total_output_tokens:,}
总Token消耗: {total:,}
平均输入/次: {avg_in:.0f}
平均输出/次: {avg_out:.0f}
预估费用: ${cost_estimate:.4f} USD
"""


class ResponseCache:
    """响应缓存 — 避免重复API调用"""

    def __init__(self, max_size: int = 100):
        self.cache: dict[str, dict] = {}
        self.max_size = max_size

    def _key(self, prompt: str, model: str) -> str:
        return hashlib.md5(f"{model}:{prompt}".encode()).hexdigest()

    def get(self, prompt: str, model: str) -> str | None:
        key = self._key(prompt, model)
        entry = self.cache.get(key)
        if entry and time.time() - entry["time"] < 3600:  # 1小时过期
            return entry["response"]
        return None

    def set(self, prompt: str, model: str, response: str):
        key = self._key(prompt, model)
        if len(self.cache) >= self.max_size:
            # 淘汰最旧的
            oldest = min(self.cache.items(), key=lambda x: x[1]["time"])
            del self.cache[oldest[0]]
        self.cache[key] = {"response": response, "time": time.time()}


def compress_context(conversation: list[dict], max_tokens: int = 4000) -> list[dict]:
    """
    上下文压缩：将长对话压缩到指定 token 限制内
    粗略估算: 1个中文字符 ≈ 0.5 token
    """
    compressed = []
    token_count = 0

    # 从最近的对话开始往前取
    for turn in reversed(conversation):
        turn_tokens = (len(turn.get("user", "")) + len(turn.get("assistant", ""))) * 0.5
        if token_count + turn_tokens > max_tokens:
            # 对更早的对话做摘要
            if not compressed:
                compressed.append({
                    "role": "user",
                    "content": f"[早期对话摘要: 共{len(conversation)}轮，已省略]"
                })
            break
        compressed.insert(0, turn)
        token_count += turn_tokens

    return compressed

3.2 代码模块化与解耦

# 推荐的项目目录结构
"""
ai_agent_project/
├── agent/
│   ├── __init__.py
│   ├── core.py           # Agent 核心运行循环
│   ├── memory.py         # 记忆系统（短期/长期/会话）
│   ├── tools.py          # 工具注册与执行
│   ├── planner.py        # 任务规划器
│   └── prompts.py        # Prompt 模板集中管理
├── services/
│   ├── __init__.py
│   ├── llm_service.py    # 大模型调用封装
│   ├── vector_store.py   # 向量数据库服务
│   └── cache_service.py  # 缓存服务
├── utils/
│   ├── __init__.py
│   ├── logger.py         # 日志系统
│   ├── config.py         # 配置管理
│   └── monitor.py        # 性能监控
├── tests/
│   ├── test_memory.py
│   ├── test_tools.py
│   └── test_agent.py
├── config.yaml           # 配置文件
├── requirements.txt
└── main.py               # 入口文件
"""

3.3 日志监控与容错机制

#!/usr/bin/env python3
"""生产级日志、监控与容错系统"""

import logging
import traceback
from datetime import datetime
from pathlib import Path
from typing import Callable

# --- 日志系统 ---
def setup_logger(name: str = "agent", log_dir: str = "./logs") -> logging.Logger:
    """配置生产级日志系统"""
    Path(log_dir).mkdir(parents=True, exist_ok=True)

    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)

    # 文件处理器（详细日志）
    file_handler = logging.FileHandler(
        f"{log_dir}/{name}_{datetime.now():%Y%m%d}.log",
        encoding="utf-8"
    )
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s [%(levelname)s] %(name)s:%(funcName)s:%(lineno)d - %(message)s'
    ))

    # 控制台处理器（简要日志）
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(logging.Formatter(
        '%(asctime)s [%(levelname)s] %(message)s', '%H:%M:%S'
    ))

    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    return logger


# --- 容错装饰器 ---
def retry_on_failure(max_retries: int = 3, delay: float = 1.0,
                     backoff: float = 2.0, exceptions: tuple = (Exception,)):
    """
    自动重试装饰器
    用法: @retry_on_failure(max_retries=3, delay=1.0)
    """
    import time
    def decorator(func: Callable) -> Callable:
        def wrapper(*args, **kwargs):
            last_exception = None
            current_delay = delay

            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_retries:
                        logging.warning(
                            f"{func.__name__} 失败 (尝试 {attempt+1}/{max_retries+1}): {e}"
                        )
                        time.sleep(current_delay)
                        current_delay *= backoff
                    else:
                        logging.error(
                            f"{func.__name__} 最终失败: {e}\n{traceback.format_exc()}"
                        )

            raise last_exception
        return wrapper
    return decorator


def safe_execute(func: Callable, *args, default=None, **kwargs):
    """
    安全执行函数，失败返回默认值
    用法: result = safe_execute(risky_function, arg1, default="备用值")
    """
    try:
        return func(*args, **kwargs)
    except Exception as e:
        logging.error(f"safe_execute 捕获异常: {e}")
        return default


# --- 健康检查 ---
class HealthChecker:
    """Agent 健康检查器"""

    def __init__(self, agent):
        self.agent = agent
        self.checks: dict[str, Callable] = {}

    def register_check(self, name: str, check_fn: Callable):
        self.checks[name] = check_fn

    def run_all(self) -> dict:
        results = {}
        for name, check_fn in self.checks.items():
            try:
                ok, detail = check_fn()
                results[name] = {"healthy": ok, "detail": detail}
            except Exception as e:
                results[name] = {"healthy": False, "detail": str(e)}
        return results

    def is_healthy(self) -> bool:
        return all(r["healthy"] for r in self.run_all().values())


# 使用示例
@retry_on_failure(max_retries=2, delay=0.5, exceptions=(Exception,))
def call_llm_with_retry(prompt: str, max_tokens: int = 500) -> str:
    """带自动重试的 LLM 调用"""
    # ... 实际的 API 调用
    pass

3.4 本地轻量化部署与简易云端部署

本地部署方案

# 1. 使用 Flask 将 Agent 包装为本地 API 服务
# pip install flask

from flask import Flask, request, jsonify
app = Flask(__name__)
agent = None  # 全局 Agent 实例

@app.route('/ask', methods=['POST'])
def ask():
    data = request.get_json()
    question = data.get('question', '')
    result = agent.ask(question)
    return jsonify({"answer": result, "timestamp": datetime.now().isoformat()})

@app.route('/health')
def health():
    return jsonify({"status": "ok"})

if __name__ == '__main__':
    agent = YourAgent()  # 初始化你的Agent
    app.run(host='0.0.0.0', port=5000, debug=False)
    # 访问: http://localhost:5000/ask

简易云端部署方案（Docker）

# Dockerfile
FROM python:3.12-slim

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

COPY . .
EXPOSE 5000

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

# 构建 & 运行
# docker build -t my-agent .
# docker run -p 5000:5000 --env-file .env my-agent