7.4 小结和复习

🎯 小白理解：这章我们学了什么？
这一章我们学会了让程序"联网"——和外部世界交流：
之前：程序只能自己玩
           ↓
        学了这章
           ↓
现在：程序能调用 ChatGPT、获取天气、连接数据库...
1
2
3
4
5
三大核心技能：
技能作用你学到了什么
环境变量安全存密钥 .env 文件、os.getenv()
HTTP 请求和服务器通信 requests、httpx、错误处理
API 集成调用外部服务 OpenAI、Anthropic、智能路由
恭喜你！学完这章，你的 Agent 就能"联网"了！

技能	作用	你学到了什么
环境变量	安全存密钥	`.env` 文件、`os.getenv()`
HTTP 请求	和服务器通信	`requests`、`httpx`、错误处理
API 集成	调用外部服务	OpenAI、Anthropic、智能路由

本章核心概念

1. 环境变量管理

.env 文件：使用 python-dotenv 加载环境变量
配置类：使用 dataclass 封装配置
多环境：development/staging/production 配置分离
密钥安全：永远不提交 .env 到 Git

2. HTTP 请求

requests 库：同步 HTTP 请求
httpx 库：异步 HTTP 请求
错误处理：处理超时、连接错误、HTTP 错误
重试机制：指数退避策略

3. API 集成

统一接口：使用抽象基类统一不同 API
多提供商：OpenAI、Anthropic 等
智能路由：自动选择最佳 API
成本管理：追踪和控制使用成本

知识自测

选择题

使用 python-dotenv 时，如何覆盖已有的环境变量？
- A. load_dotenv(force=True)
- B. load_dotenv(override=True)
- C. load_dotenv(replace=True)
- D. 不能覆盖
requests 和 httpx 的主要区别是？
- A. httpx 支持 HTTP/2
- B. httpx 支持异步请求
- C. httpx 更快
- D. A 和 B 都对
API 重试的最佳实践是？
- A. 固定间隔重试
- B. 指数退避重试
- C. 随机间隔重试
- D. 立即重试

查看答案

B - 使用 override=True 参数覆盖已有环境变量
D - httpx 同时支持 HTTP/2 和异步请求
B - 指数退避是最佳实践，避免过度请求

高难度编程挑战

🎯 小白理解：这些挑战在做什么？
挑战 1：企业级 API 管理系统
想象你是一家公司的技术主管，公司有 100 个员工都要用 OpenAI API：
每个人需要自己的 API Key（多租户）
有些人只能用免费额度，有些人可以用更多（配额管理）
月底要知道谁用了多少钱（成本分摊）
有人快超额时要提醒他（告警系统）
挑战 2：智能缓存系统
用户经常问类似的问题：
"Python 是什么？"
"什么是 Python？"
"请介绍 Python"
每次都调用 API？太浪费了！缓存系统可以：
记住问过的问题和答案
识别"相似"的问题，复用答案
节省 90% 以上的 API 调用！

挑战 1：企业级 API 管理系统（难度：⭐⭐⭐⭐）

需求：构建一个企业级的 API 管理系统，支持以下功能：

多租户支持：不同用户/团队使用不同的 API Key
配额管理：每个用户有不同的调用限额和速率限制
使用追踪：记录每次调用的详细信息
成本分摊：计算每个用户的成本
告警系统：超过配额时发送告警
审计日志：记录所有 API 调用历史

代码框架：

python

from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import sqlite3
from abc import ABC, abstractmethod

class QuotaType(Enum):
    """配额类型"""
    DAILY = "daily"
    MONTHLY = "monthly"
    TOTAL = "total"

@dataclass
class User:
    """用户信息"""
    id: str
    name: str
    email: str
    tier: str  # free, pro, enterprise
    api_keys: List[str] = field(default_factory=list)
    active: bool = True

@dataclass
class Quota:
    """配额定义"""
    user_id: str
    quota_type: QuotaType
    max_calls: int
    max_tokens: int
    max_cost: float
    reset_at: datetime

@dataclass
class UsageRecord:
    """使用记录"""
    id: str
    user_id: str
    api_key: str
    provider: str
    model: str
    input_tokens: int
    output_tokens: int
    cost: float
    timestamp: datetime
    success: bool
    error: Optional[str] = None

class QuotaManager:
    """配额管理器"""
    
    def __init__(self, db_path: str = "api_usage.db"):
        self.db_path = db_path
        self._init_database()
    
    def _init_database(self):
        """初始化数据库"""
        # TODO: 创建表结构
        # - users 表
        # - quotas 表
        # - usage_records 表
        pass
    
    def create_user(self, user: User) -> str:
        """创建用户"""
        # TODO: 实现用户创建逻辑
        pass
    
    def generate_api_key(self, user_id: str) -> str:
        """为用户生成 API Key"""
        # TODO: 生成并存储 API Key
        pass
    
    def set_quota(self, quota: Quota):
        """设置用户配额"""
        # TODO: 实现配额设置逻辑
        pass
    
    def check_quota(self, user_id: str, tokens: int, cost: float) -> bool:
        """检查是否超过配额"""
        # TODO: 实现配额检查逻辑
        pass
    
    def record_usage(self, record: UsageRecord):
        """记录使用情况"""
        # TODO: 实现使用记录逻辑
        pass
    
    def get_usage_stats(
        self,
        user_id: str,
        start_date: Optional[datetime] = None,
        end_date: Optional[datetime] = None
    ) -> Dict[str, Any]:
        """获取使用统计"""
        # TODO: 实现统计逻辑
        pass

class AlertSystem:
    """告警系统"""
    
    def __init__(self, quota_manager: QuotaManager):
        self.quota_manager = quota_manager
        self.alert_thresholds = [0.8, 0.9, 1.0]  # 80%, 90%, 100%
    
    def check_and_alert(self, user_id: str):
        """检查并发送告警"""
        # TODO: 实现告警逻辑
        pass
    
    def send_alert(self, user: User, message: str):
        """发送告警通知"""
        # TODO: 实现通知发送（邮件/短信/Webhook）
        pass

class EnterpriseAPIManager:
    """企业 API 管理器"""
    
    def __init__(self):
        self.quota_manager = QuotaManager()
        self.alert_system = AlertSystem(self.quota_manager)
        self.llm_manager = MultiLLMManager()  # 从上一节导入
    
    def authenticate(self, api_key: str) -> Optional[User]:
        """验证 API Key"""
        # TODO: 实现认证逻辑
        pass
    
    async def chat_completion(
        self,
        api_key: str,
        messages: List[Message],
        provider: APIProvider = APIProvider.OPENAI,
        **kwargs
    ) -> CompletionResponse:
        """带配额控制的聊天补全"""
        # TODO: 实现以下逻辑
        # 1. 验证 API Key
        # 2. 检查配额
        # 3. 调用 LLM API
        # 4. 记录使用情况
        # 5. 检查告警
        pass
    
    def get_billing_report(
        self,
        user_id: str,
        start_date: datetime,
        end_date: datetime
    ) -> Dict[str, Any]:
        """生成账单报告"""
        # TODO: 实现账单生成逻辑
        pass

# 测试代码
async def test_enterprise_system():
    manager = EnterpriseAPIManager()
    
    # 创建用户
    user = User(
        id="user-001",
        name="测试用户",
        email="test@example.com",
        tier="pro"
    )
    user_id = manager.quota_manager.create_user(user)
    
    # 生成 API Key
    api_key = manager.quota_manager.generate_api_key(user_id)
    
    # 设置配额
    quota = Quota(
        user_id=user_id,
        quota_type=QuotaType.DAILY,
        max_calls=100,
        max_tokens=100000,
        max_cost=10.0,
        reset_at=datetime.now() + timedelta(days=1)
    )
    manager.quota_manager.set_quota(quota)
    
    # 测试 API 调用
    messages = [Message(role="user", content="Hello")]
    
    try:
        response = await manager.chat_completion(
            api_key=api_key,
            messages=messages,
            provider=APIProvider.OPENAI
        )
        print(f"响应: {response.content}")
    except Exception as e:
        print(f"错误: {e}")
    
    # 获取使用统计
    stats = manager.quota_manager.get_usage_stats(user_id)
    print(f"\n使用统计: {stats}")

# asyncio.run(test_enterprise_system())

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199

评分标准：

多租户和 API Key 管理（20分）
配额检查和限制（20分）
使用追踪和存储（20分）
成本计算和账单（20分）
告警系统（20分）

挑战 2：智能 API 缓存系统（难度：⭐⭐⭐⭐⭐）

需求：设计一个智能缓存系统，减少 API 调用和成本。

核心功能：

语义缓存：相似问题使用缓存结果
TTL 管理：不同类型的查询有不同的过期时间
LRU 淘汰：内存不足时淘汰最少使用的缓存
缓存预热：预加载热门查询
统计报告：缓存命中率、节省成本等

代码框架：

python

from typing import Dict, Any, Optional, List, Tuple
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import OrderedDict
import hashlib
import json

@dataclass
class CacheEntry:
    """缓存条目"""
    key: str
    value: Any
    created_at: datetime
    accessed_at: datetime
    access_count: int
    ttl: int  # 秒
    tokens_saved: int
    cost_saved: float
    
    @property
    def is_expired(self) -> bool:
        """是否过期"""
        return datetime.now() > self.created_at + timedelta(seconds=self.ttl)

class SemanticCache:
    """语义缓存系统"""
    
    def __init__(
        self,
        max_size: int = 1000,
        default_ttl: int = 3600,
        similarity_threshold: float = 0.85
    ):
        self.max_size = max_size
        self.default_ttl = default_ttl
        self.similarity_threshold = similarity_threshold
        
        self.cache: OrderedDict[str, CacheEntry] = OrderedDict()
        self.embeddings: Dict[str, List[float]] = {}
        
        # 统计信息
        self.hits = 0
        self.misses = 0
        self.total_tokens_saved = 0
        self.total_cost_saved = 0.0
    
    def _generate_key(self, messages: List[Message]) -> str:
        """生成缓存键"""
        # TODO: 实现键生成逻辑
        pass
    
    def _compute_embedding(self, text: str) -> List[float]:
        """计算文本嵌入（用于语义相似度）"""
        # TODO: 实现嵌入计算（可以调用 OpenAI Embedding API）
        pass
    
    def _cosine_similarity(self, vec1: List[float], vec2: List[float]) -> float:
        """计算余弦相似度"""
        # TODO: 实现相似度计算
        pass
    
    def _find_similar(self, key: str, embedding: List[float]) -> Optional[str]:
        """查找语义相似的缓存"""
        # TODO: 实现相似度搜索
        pass
    
    def _evict_lru(self):
        """淘汰最少使用的缓存"""
        # TODO: 实现 LRU 淘汰策略
        pass
    
    def get(
        self,
        messages: List[Message],
        use_semantic: bool = True
    ) -> Optional[CompletionResponse]:
        """获取缓存"""
        # TODO: 实现缓存查询逻辑
        # 1. 生成缓存键
        # 2. 精确匹配查找
        # 3. 语义相似查找（如果启用）
        # 4. 更新统计信息
        pass
    
    def set(
        self,
        messages: List[Message],
        response: CompletionResponse,
        ttl: Optional[int] = None
    ):
        """设置缓存"""
        # TODO: 实现缓存存储逻辑
        # 1. 生成键和嵌入
        # 2. 检查容量，必要时淘汰
        # 3. 存储缓存
        pass
    
    def warm_up(self, popular_queries: List[Tuple[List[Message], CompletionResponse]]):
        """缓存预热"""
        # TODO: 实现预热逻辑
        pass
    
    def get_stats(self) -> Dict[str, Any]:
        """获取统计信息"""
        total = self.hits + self.misses
        hit_rate = self.hits / total if total > 0 else 0.0
        
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": hit_rate,
            "cache_size": len(self.cache),
            "tokens_saved": self.total_tokens_saved,
            "cost_saved": self.total_cost_saved
        }
    
    def clear_expired(self):
        """清理过期缓存"""
        # TODO: 实现过期清理逻辑
        pass

class CachedAPIManager:
    """带缓存的 API 管理器"""
    
    def __init__(self, llm_manager: MultiLLMManager):
        self.llm_manager = llm_manager
        self.cache = SemanticCache()
    
    async def chat_completion(
        self,
        messages: List[Message],
        provider: APIProvider = APIProvider.OPENAI,
        use_cache: bool = True,
        **kwargs
    ) -> CompletionResponse:
        """带缓存的聊天补全"""
        # TODO: 实现缓存逻辑
        # 1. 尝试从缓存获取
        # 2. 缓存未命中，调用 API
        # 3. 将结果存入缓存
        # 4. 返回结果
        pass

# 测试代码
async def test_semantic_cache():
    manager = MultiLLMManager()
    cached_manager = CachedAPIManager(manager)
    
    # 测试相似查询
    queries = [
        "什么是 Python?",
        "Python 是什么?",
        "请解释 Python",
        "什么是 JavaScript?",  # 不相似
    ]
    
    for query in queries:
        messages = [Message(role="user", content=query)]
        response = await cached_manager.chat_completion(messages)
        print(f"\nQ: {query}")
        print(f"缓存命中: {response.from_cache if hasattr(response, 'from_cache') else False}")
    
    # 打印统计
    stats = cached_manager.cache.get_stats()
    print(f"\n=== 缓存统计 ===")
    print(f"命中率: {stats['hit_rate']:.2%}")
    print(f"节省 Token: {stats['tokens_saved']}")
    print(f"节省成本: ${stats['cost_saved']:.4f}")

# asyncio.run(test_semantic_cache())

评分标准：

基础缓存实现（15分）
语义相似度匹配（25分）
LRU 淘汰策略（20分）
TTL 管理和过期清理（15分）
统计和成本节省（25分）

学习资源

进阶主题

API 网关设计
服务网格（Service Mesh）
分布式追踪（OpenTelemetry）
GraphQL vs REST

下一章：Module 8 - 综合项目

7.4 小结和复习 ​

本章核心概念 ​

1. 环境变量管理 ​

2. HTTP 请求 ​

3. API 集成 ​

知识自测 ​

选择题 ​

高难度编程挑战 ​

挑战 1：企业级 API 管理系统（难度：⭐⭐⭐⭐） ​

挑战 2：智能 API 缓存系统（难度：⭐⭐⭐⭐⭐） ​

学习资源 ​

推荐阅读 ​

进阶主题 ​