1
@dataclass
2
class CostBreakdown:
3
    system_prompt: int  # 系统提示词（只算一次）
4
    conversation_history: int  # 对话历史
5
    tool_result: int  # 工具返回结果
6
    final_response: int  # 最终响应
7

8
# 估算成本
9
def estimate_cost(breakdown: CostBreakdown) -> float:
10
    return (
11
        breakdown.system_prompt * 3.0 +  # $3/1M tokens
12
        breakdown.conversation_history * 3.0 +
13
        breakdown.tool_result * 0.6 +  # $0.6/1M
14
        breakdown.final_response * 3.0
15
    ) / 1_000_000

1
简单问答:    1 次 LLM 调用 → 约 500 tokens → $0.001
2
工具调用:    3 次 LLM 调用 → 约 3000 tokens → $0.006
3
深度研究:    10 次 LLM 调用 → 约 15000 tokens → $0.03
4
多 Agent 协作: 20 次 LLM 调用 → 约 50000 tokens → $0.10

1
# 精简前
2
system_prompt_old = """
3
你是一个专业的研究助手。
4
你擅长信息检索、总结和分析。
5
你有以下工具可以使用：
6
1. search - 搜索网络
7
2. scrape - 抓取网页
8
3. calculator - 计算
9
请基于事实回答，不要编造信息。
10
"""
11

12
# 精简后
13
system_prompt_new = """
14
你是一个研究助手。使用 search/scrape/calculator。
15
基于事实回答。
16
"""

1
def compress_history(messages: list, max_tokens: int = 2000) -> list:
2
    """对话历史压缩"""
3
    if count_tokens(messages) <= max_tokens:
4
        return messages
5

6
    # 保留系统 + 最近 N 轮
7
    system = messages[0]  # 系统消息
8
    recent = messages[-5:]  # 最近5轮
9

10
    # 对旧消息摘要
11
    old_messages = messages[1:-5]
12
    summary = summarize_conversation(old_messages)
13

14
    return [system, summary] + recent

1
class PromptCompressor:
2
    """Prompt 自动压缩器"""
3

4
    async def compress(self, prompt: str, compression_ratio: float = 0.5) -> str:
5
        """将 Prompt 压缩到指定比例"""
6
        target_tokens = int(count_tokens(prompt) * compression_ratio)
7

8
        compressed = await llm.complete(f"""请将以下 Prompt 压缩到约 {target_tokens} tokens，
9
保持核心指令不变，去除冗余描述。
10

11
原始 Prompt:
12
{prompt}
13

14
压缩后 Prompt:""")
15

16
        return compressed.strip()
17

18
    async def compress_tool_descriptions(self, tools: list[dict]) -> list[dict]:
19
        """压缩工具描述"""
20
        compressed = []
21
        for tool in tools:
22
            short_desc = await self.compress(tool["description"], compression_ratio=0.4)
23
            compressed.append({
24
                "name": tool["name"],
25
                "description": short_desc,
26
                "parameters": self._shorten_params(tool["parameters"]),
27
            })
28
        return compressed
29

30
    def _shorten_params(self, params: dict) -> dict:
31
        """精简参数定义"""
32
        required = params.get("required", [])
33
        properties = {}
34
        for name, schema in params.get("properties", {}).items():
35
            if name in required:
36
                # 必填参数保留类型和简短描述
37
                properties[name] = {"type": schema.get("type", "string")}
38
            # 可选参数：根据需要保留或移除
39
        return {"type": "object", "required": required, "properties": properties}

1
class DynamicPromptSelector:
2
    """动态 Prompt 选择器"""
3

4
    def __init__(self):
5
        self.prompts = {
6
            "full": FULL_SYSTEM_PROMPT,      # 约 800 tokens
7
            "compact": COMPACT_SYSTEM_PROMPT, # 约 200 tokens
8
            "minimal": MINIMAL_SYSTEM_PROMPT, # 约 50 tokens
9
        }
10

11
    def select(self, query: str, context: dict) -> str:
12
        complexity = self._estimate_complexity(query, context)
13

14
        if complexity < 0.3:
15
            return self.prompts["minimal"]
16
        elif complexity < 0.7:
17
            return self.prompts["compact"]
18
        else:
19
            return self.prompts["full"]
20

21
    def _estimate_complexity(self, query: str, context: dict) -> float:
22
        """评估查询复杂度 (0-1)"""
23
        score = 0.0
24

25
        # 长查询更复杂
26
        if len(query) > 200:
27
            score += 0.2
28

29
        # 需要多步骤的任务更复杂
30
        if any(kw in query for kw in ["分析", "比较", "总结", "研究"]):
31
            score += 0.3
32

33
        # 需要工具调用的任务更复杂
34
        if context.get("tools_available"):
35
            score += 0.2
36

37
        # 历史对话长更复杂
38
        if context.get("history_length", 0) > 10:
39
            score += 0.2
40

41
        return min(score, 1.0)

1
from cachetools import cached
2
import hashlib
3

4
@cached(cache={}, key=lambda query: hashlib.md5(query.encode()).hexdigest())
5
async def cached_search(query: str) -> str:
6
    """搜索结果缓存"""
7
    return await search(query)

1
from redis import Redis
2

3
redis_client = Redis()
4

5
def cache_embedding(text: str, embedding: list[float]):
6
    key = f"embedding:{hash(text)}"
7
    redis_client.setex(key, ttl=86400, value=json.dumps(embedding))
8

9
def get_cached_embedding(text: str) -> Optional[list[float]]:
10
    key = f"embedding:{hash(text)}"
11
    return redis_client.get(key)

1
import numpy as np
2

3
class SemanticCache:
4
    """语义缓存：基于向量相似度的缓存"""
5

6
    def __init__(
7
        self,
8
        embedding_model,
9
        similarity_threshold: float = 0.92,
10
        ttl_seconds: int = 3600,
11
    ):
12
        self.embedding_model = embedding_model
13
        self.similarity_threshold = similarity_threshold
14
        self.ttl = ttl_seconds
15
        self.cache_store = {}  # 生产环境用 Redis + 向量数据库
16

17
    async def get(self, query: str) -> tuple[str | None, float]:
18
        """查询缓存，返回 (缓存结果, 相似度)"""
19
        query_embedding = await self.embedding_model.embed(query)
20

21
        best_match = None
22
        best_similarity = 0.0
23

24
        for key, entry in self.cache_store.items():
25
            # 检查 TTL
26
            if self._is_expired(entry):
27
                continue
28

29
            similarity = cosine_similarity(query_embedding, entry["embedding"])
30
            if similarity > best_similarity:
31
                best_similarity = similarity
32
                best_match = entry["result"]
33

34
        if best_similarity >= self.similarity_threshold:
35
            return best_match, best_similarity
36
        return None, best_similarity
37

38
    async def set(self, query: str, result: str):
39
        """存入缓存"""
40
        embedding = await self.embedding_model.embed(query)
41
        self.cache_store[hashlib.md5(query.encode()).hexdigest()] = {
42
            "query": query,
43
            "result": result,
44
            "embedding": embedding,
45
            "timestamp": time.time(),
46
        }
47

48
    def _is_expired(self, entry: dict) -> bool:
49
        return time.time() - entry["timestamp"] > self.ttl
50

51
def cosine_similarity(a: list[float], b: list[float]) -> float:
52
    a_np = np.array(a)
53
    b_np = np.array(b)
54
    return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))

1
class MultiLevelCache:
2
    """多级缓存"""
3

4
    def __init__(self):
5
        self.l1 = LRUCache(maxsize=100)      # 内存缓存：最快，容量小
6
        self.l2 = RedisCache(ttl=3600)        # Redis：精确匹配
7
        self.l3 = VectorDBCache(threshold=0.9) # 向量库：语义匹配
8

9
    async def get(self, query: str) -> str | None:
10
        # L1: 内存精确匹配
11
        result = self.l1.get(query)
12
        if result:
13
            return result
14

15
        # L2: Redis 精确匹配
16
        result = await self.l2.get(query)
17
        if result:
18
            self.l1.set(query, result)  # 回填 L1
19
            return result
20

21
        # L3: 向量库语义匹配
22
        result, similarity = await self.l3.search(query)
23
        if result and similarity > self.l3.threshold:
24
            self.l1.set(query, result)  # 回填
25
            await self.l2.set(query, result)
26
            return result
27

28
        return None
29

30
    async def set(self, query: str, result: str):
31
        """写入所有缓存层"""
32
        self.l1.set(query, result)
33
        await self.l2.set(query, result)
34
        await self.l3.add(query, result)

1
async def route(query: str) -> str:
2
    """根据查询复杂度选择处理路径"""
3
    complexity = estimate_complexity(query)
4

5
    if complexity == "simple":
6
        return await simple_rag(query)  # 便宜快速
7
    else:
8
        return await deep_research(query)  # 昂贵但深入

1
async def cheap_first(query: str) -> str:
2
    """先用小模型尝试，失败再调用大模型"""
3
    try:
4
        # 先用便宜模型
5
        result = await llm.mini.complete(query)
6
        if is_satisfactory(result):
7
            return result
8
    except QualityCheckFailed:
9
        pass
10

11
    # 回退到大模型
12
    return await llm.pro.complete(query)

1
from enum import Enum
2

3
class ModelTier(Enum):
4
    CHEAP = "cheap"       # GPT-4o-mini, Haiku, Flash
5
    STANDARD = "standard"  # GPT-4o, Sonnet
6
    PREMIUM = "premium"    # Claude Opus, o1
7

8
class ModelRouter:
9
    """智能模型路由器"""
10

11
    TIER_MODELS = {
12
        ModelTier.CHEAP: ["gpt-4o-mini", "claude-haiku-3.5", "gemini-2.0-flash"],
13
        ModelTier.STANDARD: ["gpt-4o", "claude-sonnet-4", "deepseek-chat"],
14
        ModelTier.PREMIUM: ["claude-opus-4", "o1", "deepseek-reasoner"],
15
    }
16

17
    def __init__(self):
18
        self.routing_rules = self._load_rules()
19
        self.fallback_enabled = True
20

21
    async def route(self, query: str, context: dict | None = None) -> tuple[str, ModelTier]:
22
        """为查询选择最合适的模型"""
23
        complexity = await self._classify_complexity(query, context)
24

25
        if complexity.score < 0.3:
26
            tier = ModelTier.CHEAP
27
        elif complexity.score < 0.7:
28
            tier = ModelTier.STANDARD
29
        else:
30
            tier = ModelTier.PREMIUM
31

32
        # 从对应层级中选择模型
33
        model = self._select_from_tier(tier, complexity)
34
        return model, tier
35

36
    async def _classify_complexity(self, query: str, context: dict | None) -> dict:
37
        """分类查询复杂度"""
38
        score = 0.0
39
        signals = []
40

41
        # 信号 1: 查询长度
42
        if len(query) > 500:
43
            score += 0.2
44
            signals.append("长查询")
45
        elif len(query) < 50:
46
            score -= 0.1
47
            signals.append("短查询")
48

49
        # 信号 2: 是否需要推理
50
        reasoning_keywords = ["分析", "比较", "为什么", "如何", "evaluate", "compare", "why"]
51
        if any(kw in query.lower() for kw in reasoning_keywords):
52
            score += 0.25
53
            signals.append("需要推理")
54

55
        # 信号 3: 是否需要工具调用
56
        tool_keywords = ["搜索", "查询", "计算", "查找", "search", "calculate"]
57
        if any(kw in query.lower() for kw in tool_keywords):
58
            score += 0.15
59
            signals.append("需要工具")
60

61
        # 信号 4: 是否涉及代码
62
        code_indicators = ["代码", "函数", "bug", "code", "function", "debug"]
63
        if any(kw in query.lower() for kw in code_indicators):
64
            score += 0.2
65
            signals.append("涉及代码")
66

67
        # 信号 5: 历史对话轮次
68
        if context and context.get("turn_count", 0) > 5:
69
            score += 0.1
70
            signals.append("多轮对话")
71

72
        return {
73
            "score": max(0, min(1, score)),
74
            "signals": signals,
75
        }
76

77
    def _select_from_tier(self, tier: ModelTier, complexity: dict) -> str:
78
        """从指定层级选择具体模型"""
79
        models = self.TIER_MODELS[tier]
80
        # 简单的轮询策略，生产环境可加入更多考量
81
        return models[0]
82

83

84
class CascadingRouter:
85
    """级联回退路由器：先用便宜模型，不行再升级"""
86

87
    def __init__(self, router: ModelRouter):
88
        self.router = router
89
        self.quality_threshold = 0.7
90

91
    async def route_with_fallback(self, query: str) -> str:
92
        """带回退的路由"""
93
        # 先尝试便宜模型
94
        model, tier = self.router.route(query, context=None)
95

96
        if tier == ModelTier.CHEAP:
97
            result = await self._call_model(model, query)
98
            quality = await self._assess_quality(query, result)
99

100
            if quality >= self.quality_threshold:
101
                return result
102

103
            # 便宜模型不行，升级到标准模型
104
            model = self.router.TIER_MODELS[ModelTier.STANDARD][0]
105

106
        result = await self._call_model(model, query)
107

108
        # 如果标准模型也不行（不太常见），再升级
109
        quality = await self._assess_quality(query, result)
110
        if quality < self.quality_threshold and model not in self.router.TIER_MODELS[ModelTier.PREMIUM]:
111
            premium_model = self.router.TIER_MODELS[ModelTier.PREMIUM][0]
112
            result = await self._call_model(premium_model, query)
113

114
        return result
115

116
    async def _call_model(self, model: str, query: str) -> str:
117
        return await llm.complete(query, model=model)
118

119
    async def _assess_quality(self, query: str, response: str) -> float:
120
        """快速质量评估（用便宜模型）"""
121
        assessment = await llm.complete(
122
            f"评估以下回答是否充分回答了问题。只输出 0-1 的分数。\n\n"
123
            f"问题: {query}\n回答: {response[:500]}\n\n分数:",
124
            model="gpt-4o-mini",  # 用便宜模型做质量检查
125
        )
126
        try:
127
            return float(assessment.strip())
128
        except ValueError:
129
            return 0.5  # 解析失败，默认中等质量

1
import asyncio
2
from collections import defaultdict
3

4
class BatchProcessor:
5
    """LLM 批量请求处理器"""
6

7
    def __init__(self, max_batch_size: int = 10, max_wait_seconds: float = 0.5):
8
        self.max_batch_size = max_batch_size
9
        self.max_wait = max_wait_seconds
10
        self.queue: list[tuple[str, asyncio.Future]] = []
11
        self.lock = asyncio.Lock()
12

13
    async def submit(self, prompt: str) -> str:
14
        """提交单个请求，等待批量处理"""
15
        loop = asyncio.get_event_loop()
16
        future = loop.create_future()
17

18
        async with self.lock:
19
            self.queue.append((prompt, future))
20

21
            # 队列满了就立即处理
22
            if len(self.queue) >= self.max_batch_size:
23
                await self._flush()
24

25
        return await future
26

27
    async def _flush(self):
28
        """批量发送"""
29
        batch = self.queue[:self.max_batch_size]
30
        self.queue = self.queue[self.max_batch_size:]
31

32
        prompts = [p for p, _ in batch]
33
        futures = [f for _, f in batch]
34

35
        try:
36
            # 使用批量 API（如 OpenAI Batch API）
37
            results = await batch_llm_call(prompts)
38

39
            for future, result in zip(futures, results):
40
                if not future.done():
41
                    future.set_result(result)
42
        except Exception as e:
43
            for future in futures:
44
                if not future.done():
45
                    future.set_exception(e)
46

47
    async def start(self):
48
        """启动定时刷新"""
49
        while True:
50
            await asyncio.sleep(self.max_wait)
51
            async with self.lock:
52
                if self.queue:
53
                    await self._flush()

1
import json
2
import time
3

4
class OpenAIBatchAPI:
5
    """OpenAI Batch API 封装"""
6

7
    def __init__(self, client):
8
        self.client = client
9

10
    async def submit_batch(self, tasks: list[dict]) -> str:
11
        """提交批量任务"""
12
        # 创建批量输入文件
13
        batch_lines = []
14
        for i, task in enumerate(tasks):
15
            batch_lines.append(json.dumps({
16
                "custom_id": f"task-{i}",
17
                "method": "POST",
18
                "url": "/v1/chat/completions",
19
                "body": {
20
                    "model": task.get("model", "gpt-4o-mini"),
21
                    "messages": task["messages"],
22
                    "max_tokens": task.get("max_tokens", 1000),
23
                }
24
            }))
25

26
        # 上传文件
27
        batch_file = self.client.files.create(
28
            file="\n".join(batch_lines),
29
            purpose="batch",
30
        )
31

32
        # 创建批量任务
33
        batch = self.client.batches.create(
34
            input_file_id=batch_file.id,
35
            endpoint="/v1/chat/completions",
36
            completion_window="24h",
37
        )
38

39
        return batch.id
40

41
    async def wait_and_get_results(self, batch_id: str) -> list[dict]:
42
        """等待批量任务完成并获取结果"""
43
        while True:
44
            batch = self.client.batches.retrieve(batch_id)
45

46
            if batch.status == "completed":
47
                # 下载结果文件
48
                results = self.client.files.content(batch.output_file_id)
49
                return [json.loads(line) for line in results.text.strip().split("\n")]
50
            elif batch.status == "failed":
51
                raise Exception(f"Batch failed: {batch.errors}")
52
            elif batch.status in ("expired", "cancelled"):
53
                raise Exception(f"Batch {batch.status}")
54

55
            time.sleep(30)  # 每 30 秒检查一次

1
class ParallelToolExecutor:
2
    """并行工具执行器"""
3

4
    async def execute_tools(self, tool_calls: list[dict]) -> list[dict]:
5
        """并行执行独立的工具调用"""
6
        # 分析依赖关系
7
        independent, dependent = self._analyze_dependencies(tool_calls)
8

9
        # 并行执行独立调用
10
        results = {}
11
        if independent:
12
            tasks = [
13
                self._execute_single(tc)
14
                for tc in independent
15
            ]
16
            parallel_results = await asyncio.gather(*tasks, return_exceptions=True)
17
            for tc, result in zip(independent, parallel_results):
18
                results[tc["id"]] = result
19

20
        # 串行执行有依赖的调用
21
        for tc in dependent:
22
            # 注入之前的结果作为上下文
23
            tc_with_context = self._inject_context(tc, results)
24
            result = await self._execute_single(tc_with_context)
25
            results[tc["id"]] = result
26

27
        return results
28

29
    def _analyze_dependencies(self, tool_calls: list[dict]) -> tuple[list, list]:
30
        """分析工具调用的依赖关系"""
31
        independent = []
32
        dependent = []
33

34
        for tc in tool_calls:
35
            # 简单启发式：如果参数引用了其他调用的结果，则依赖
36
            params_str = json.dumps(tc.get("params", {}))
37
            if any(f"${{result.{other_id}}}" in params_str for other_id in range(len(tool_calls))):
38
                dependent.append(tc)
39
            else:
40
                independent.append(tc)
41

42
        return independent, dependent

1
# 成本监控指标
2
agent_cost_total{category="research"} 125.50
3
agent_cost_per_request 0.15
4
agent_token_usage{purpose="tool_result"} 45%
5
agent_cache_hit_rate 0.72

1
@dataclass
2
class CostAlert:
3
    threshold_usd: float = 1.0
4
    window_minutes: int = 60
5

6
async def check_cost_alert(cost_breakdown: CostBreakdown):
7
    if cost_breakdown.total_usd > alert.threshold_usd:
8
        await send_alert(
9
            f"Agent 成本超限: ${cost_breakdown.total_usd}"
10
        )

1
class CostOptimizationDashboard:
2
    """成本优化仪表板"""
3

4
    def generate_report(self, period: str = "daily") -> dict:
5
        return {
6
            "period": period,
7
            "total_cost": self._get_total_cost(period),
8
            "cost_by_agent": self._get_cost_by_agent(period),
9
            "cost_by_model": self._get_cost_by_model(period),
10
            "savings": {
11
                "cache_savings": self._get_cache_savings(period),
12
                "routing_savings": self._get_routing_savings(period),
13
                "compression_savings": self._get_compression_savings(period),
14
            },
15
            "recommendations": self._generate_recommendations(),
16
        }
17

18
    def _generate_recommendations(self) -> list[str]:
19
        """生成成本优化建议"""
20
        recs = []
21

22
        # 检查是否有简单任务使用了昂贵模型
23
        expensive_for_simple = self._check_model_task_mismatch()
24
        if expensive_for_simple:
25
            recs.append(
26
                f"发现 {len(expensive_for_simple)} 个简单任务使用了昂贵模型，"
27
                f"建议启用智能路由，预计节省 ${expensive_for_simple['potential_savings']:.2f}/月"
28
            )
29

30
        # 检查缓存命中率
31
        cache_hit_rate = self._get_cache_hit_rate()
32
        if cache_hit_rate < 0.3:
33
            recs.append(
34
                f"缓存命中率仅 {cache_hit_rate:.1%}，建议优化缓存策略"
35
            )
36

37
        # 检查 Prompt 冗余
38
        avg_prompt_tokens = self._get_avg_prompt_tokens()
39
        if avg_prompt_tokens > 2000:
40
            recs.append(
41
                f"平均 Prompt 长度 {avg_prompt_tokens} tokens，建议启用 Prompt 压缩"
42
            )
43

44
        return recs

1
日均请求: 100 次
2
平均每次: 3 次 LLM 调用, 约 2000 tokens
3
模型: GPT-4o-mini ($0.15/$0.60 per 1M tokens)
4

5
月度估算:
6
- Token 消耗: 100 * 3 * 2000 * 30 = 18M tokens/月
7
- 输入成本: ~$2.70/月
8
- 输出成本: ~$1.08/月
9
- 总计: ~$5/月

1
日均请求: 10,000 次
2
优化前: GPT-4o 全量, 约 5000 tokens/次
3
优化后: 70% GPT-4o-mini + 30% GPT-4o, 约 3000 tokens/次
4

5
月度估算（优化前）:
6
- 约 $4,500/月
7

8
月度估算（优化后）:
9
- 约 $1,200/月
10
- 节省: 73%

1
日均请求: 1,000,000 次
2
策略: 智能路由 + 语义缓存 + 批量处理 + 本地模型兜底
3

4
优化措施效果:
5
- 语义缓存: 命中率 40%, 节省 40%
6
- 智能路由: 60% 走便宜模型, 节省 45%
7
- Prompt 压缩: 减少 30% Token
8
- 批量 API: 减少 50% API 调用成本
9
- 本地模型: 20% 流量走本地, 成本接近 0
10

11
综合优化: 约 85% 成本节省