作为一名经历过多次在线教育平台后端重构的技术负责人，我今天分享一套在生产环境验证超过2年的自适应学习系统架构。这套方案利用大语言模型实现动态知识点掌握度评估，实测单学生评估延迟<800ms，月度Token成本降低67%，支持QPS 500+的并发场景。
一、系统架构总览
自适应学习系统的核心挑战在于：如何让AI真正理解学生的知识薄弱点，而不是简单打分。我设计的架构包含四层：
- 知识图谱层：Neo4j图数据库存储知识点依赖关系，支持复杂度O(1)的关联查询
- LLM评估引擎：基于HolySheep AI API的GPT-4.1实现评估逻辑
- 学生画像服务：Redis缓存+ClickHouse持久化，支持毫秒级画像查询
- 自适应调度器：基于掌握度梯度自动推荐下一学习路径
┌─────────────────────────────────────────────────────────────────┐
│ Adaptive Learning Arch │
├─────────────────────────────────────────────────────────────────┤
│ ┌──────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ Client │───▶│ API Gateway │───▶│ Assessment Engine │ │
│ │ (App) │ │ (Rate Limit)│ │ (LLM Evaluation) │ │
│ └──────────┘ └──────────────┘ └──────────────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────┐ ┌──────────────┐ │
│ │ Redis │◀───────────────────────│ Knowledge │ │
│ │ (Cache) │ │ Graph │ │
│ └──────────┘ │ (Neo4j) │ │
│ │ └──────────────┘ │
│ ▼ │
│ ┌──────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ClickHouse│◀───│ Student │───▶│ HolySheep API │ │
│ │ (OLAP) │ │ Profile Svc │ │ (LLM Backbone) │ │
│ └──────────┘ └──────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘二、核心评估流程实现
评估引擎是整个系统的核心。我采用"问题-响应-分析-建议"的四阶段流水线，每个学生答题后触发完整评估链路。HolySheep API的国内直连<50ms延迟让整个评估周期控制在800ms以内。
2.1 评估服务主入口
import asyncio
import httpx
from typing import Optional, List, Dict
from dataclasses import dataclass
from datetime import datetime
import json
HolySheep API Configuration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # 替换为你的Key
@dataclass
class AssessmentRequest:
student_id: str
question_id: str
question_text: str
student_response: str
knowledge_points: List[str] # 关联的知识点列表
@dataclass
class AssessmentResult:
mastery_level: float # 0.0 - 1.0
misconceptions: List[str] # 识别出的错误概念
suggested_activities: List[str] # 推荐学习活动
confidence: float # 评估置信度
tokens_used: int
processing_time_ms: float
class AdaptiveAssessmentEngine:
"""自适应评估引擎 - 生产级实现"""
def __init__(self, api_key: str, base_url: str = HOLYSHEEP_BASE_URL):
self.client = httpx.AsyncClient(
base_url=base_url,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
timeout=30.0
)
self.cost_tracker = CostTracker()
async def evaluate(
self,
request: AssessmentRequest,
model: str = "gpt-4.1" # 使用GPT-4.1进行评估
) -> AssessmentResult:
"""执行完整评估流程"""
start_time = asyncio.get_event_loop().time()
# 阶段1: 构建评估Prompt
evaluation_prompt = self._build_evaluation_prompt(request)
# 阶段2: 调用LLM进行评估
response = await self._call_llm(evaluation_prompt, model)
# 阶段3: 解析结果
result = self._parse_evaluation_response(response)
# 阶段4: 计算成本
processing_time = (asyncio.get_event_loop().time() - start_time) * 1000
self.cost_tracker.record(
model=model,
input_tokens=response.usage.input_tokens,
output_tokens=response.usage.output_tokens,
latency_ms=processing_time
)
return AssessmentResult(
mastery_level=result["mastery_level"],
misconceptions=result["misconceptions"],
suggested_activities=result["suggested_activities"],
confidence=result["confidence"],
tokens_used=response.usage.total_tokens,
processing_time_ms=processing_time
)
def _build_evaluation_prompt(self, request: AssessmentRequest) -> str:
"""构建结构化评估Prompt"""
knowledge_points_str = ", ".join(request.knowledge_points)
return f"""你是一位资深教育专家，负责评估学生对特定知识点的掌握程度。
题目信息
题目ID: {request.question_id}
题目内容: {request.question_text}
学生回答: {request.student_response}
涉及知识点: {knowledge_points_str}
评估要求
请从以下维度进行评估（JSON格式输出）:
{{
"mastery_level": 0.0-1.0的掌握度分数,
"misconceptions": ["错误概念1", "错误概念2"],
"suggested_activities": ["推荐活动1", "推荐活动2"],
"confidence": 0.0-1.0的评估置信度,
"reasoning": "评估理由（50字内）"
}}
注意：
- 如果回答完全正确，mastery_level应≥0.9
- 如果回答错误且原因不明，confidence应降低
- suggested_activities必须针对学生的具体薄弱点"""
async def _call_llm(
self,
prompt: str,
model: str,
temperature: float = 0.3 # 评估用低温保证稳定性
) -> dict:
"""调用HolySheep API"""
payload = {
"model": model,
"messages": [
{"role": "user", "content": prompt}
],
"temperature": temperature,
"max_tokens": 500
}
async with self.client.stream("POST", "/chat/completions", json=payload) as resp:
if resp.status_code != 200:
raise LLMAPIError(f"API调用失败: {resp.status_code}")
return await resp.json()
成本追踪器
class CostTracker:
def __init__(self):
self.records: List[dict] = []
self._daily_cost = 0.0
def record(self, model: str, input_tokens: int, output_tokens: int, latency_ms: float):
# GPT-4.1 pricing: $8/MTok output, $2/MTok input
prices = {
"gpt-4.1": {"input": 2.0, "output": 8.0},
"gpt-4.1-mini": {"input": 0.4, "output": 1.6},
"deepseek-v3.2": {"input": 0.12, "output": 0.42}
}
price = prices.get(model, prices["gpt-4.1"])
cost = (input_tokens / 1_000_000 * price["input"] +
output_tokens / 1_000_000 * price["output"])
self.records.append({
"timestamp": datetime.now().isoformat(),
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"latency_ms": latency_ms,
"cost_usd": cost
})
self._daily_cost += cost
def get_daily_cost(self) -> float:
return self._daily_cost2.2 知识点掌握度追踪
单一题目评估不够精准，我设计了基于贝叶斯更新的掌握度追踪算法。每次答题后结合先验概率和新证据，动态更新学生对每个知识点的掌握度。
import math
from typing import Dict
from collections import defaultdict
class KnowledgeMasteryTracker:
"""
基于贝叶斯更新的知识点掌握度追踪器
使用Beta分布建模掌握/未掌握的概率
"""
def __init__(self):
# Beta分布参数: {知识点ID: (alpha, beta)}
self.beta_params: Dict[str, tuple] = defaultdict(lambda: (1, 1))
# 历史评估缓存
self.assessment_cache: Dict[str, list] = defaultdict(list)
def update_mastery(
self,
knowledge_point_id: str,
is_correct: bool,
question_difficulty: float = 0.5, # 0.0-1.0
confidence: float = 1.0 # LLM评估置信度
) -> float:
"""
贝叶斯更新掌握度
返回更新后的后验概率
"""
alpha, beta = self.beta_params[knowledge_point_id]
# 难度调整因子：越难的题目答对/答错信息量更大
difficulty_weight = 1.0 + question_difficulty
if is_correct:
# 答对：增强掌握假设
alpha += 2.0 * difficulty_weight * confidence
else:
# 答错：增强未掌握假设
beta += 1.5 * difficulty_weight * (2 - confidence)
# 限制参数范围避免数值爆炸
alpha = min(alpha, 100)
beta = min(beta, 100)
self.beta_params[knowledge_point_id] = (alpha, beta)
# 计算后验均值作为掌握度
mastery = alpha / (alpha + beta)
# 记录历史
self.assessment_cache[knowledge_point_id].append({
"timestamp": datetime.now().isoformat(),
"is_correct": is_correct,
"mastery": mastery,
"confidence": confidence
})
return mastery
def get_mastery(self, knowledge_point_id: str) -> float:
"""获取当前掌握度"""
alpha, beta = self.beta_params[knowledge_point_id]
return alpha / (alpha + beta)
def get_mastery_uncertainty(self, knowledge_point_id: str) -> float:
"""获取掌握度的不确定性（标准差）"""
alpha, beta = self.beta_params[knowledge_point_id]
mean = alpha / (alpha + beta)
variance = (alpha * beta) / ((alpha + beta) ** 2 * (alpha + beta + 1))
return math.sqrt(variance)
def get_weak_points(self, threshold: float = 0.6, min_samples: int = 3) -> List[Dict]:
"""获取薄弱知识点列表"""
weak_points = []
for kp_id in self.beta_params:
samples = len(self.assessment_cache[kp_id])
if samples >= min_samples:
mastery = self.get_mastery(kp_id)
uncertainty = self.get_mastery_uncertainty(kp_id)
if mastery < threshold:
weak_points.append({
"knowledge_point_id": kp_id,
"mastery": round(mastery, 3),
"uncertainty": round(uncertainty, 3),
"sample_count": samples,
"priority": (threshold - mastery) / uncertainty if uncertainty > 0 else 999
})
# 按优先级排序
weak_points.sort(key=lambda x: x["priority"], reverse=True)
return weak_points[:10]
错误类型识别
class MisconceptionDetector:
"""基于LLM的错误模式识别"""
SYSTEM_PROMPT = """你是一个教育诊断专家，负责分析学生常见错误模式。
常见错误类型包括：
- 概念混淆（混淆相似概念）
- 步骤遗漏（解题步骤不完整）
- 计算失误（数值计算错误）
- 原理误解（对原理理解错误）
- 审题错误（理解题意有误）
请识别学生回答中的错误模式并给出具体建议。"""
async def detect(self, question: str, response: str, expected: str) -> Dict:
"""检测错误模式"""
user_prompt = f"""题目: {question}
学生回答: {response}
正确答案: {expected}
请分析学生错误类型并给出改进建议（JSON格式）:
{{
"error_type": "错误类型",
"specific_misconception": "具体错误点",
"suggestion": "改进建议（30字内）"
}}"""
# 调用HolySheep API
payload = {
"model": "gpt-4.1-mini", # 使用mini模型降低评估成本
"messages": [
{"role": "system", "content": self.SYSTEM_PROMPT},
{"role": "user", "content": user_prompt}
],
"temperature": 0.2,
"max_tokens": 200
}
async with httpx.AsyncClient() as client:
resp = await client.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"},
json=payload
)
result = resp.json()
return json.loads(result["choices"][0]["message"]["content"])2.3 自适应学习路径推荐
from typing import List, Tuple
import heapq
class AdaptivePathRecommender:
"""
基于知识点依赖图的自适应学习路径推荐
使用Dijkstra算法找到最优学习路径
"""
def __init__(self, knowledge_graph):
self.graph = knowledge_graph # {知识点ID: [(前置知识点, 难度), ...]}
self.mastery_tracker = KnowledgeMasteryTracker()
def recommend_next(
self,
student_id: str,
current_mastery: Dict[str, float],
target_goals: List[str],
max_items: int = 5
) -> List[Dict]:
"""
推荐下一步学习内容
返回推荐列表，按优先级排序
"""
recommendations = []
for goal in target_goals:
# 找到从当前状态到目标的最优路径
path = self._find_learning_path(current_mastery, goal)
for node, difficulty, estimated_time in path:
if current_mastery.get(node, 0) < 0.8: # 未掌握
# 计算学习紧迫度
urgency = self._calculate_urgency(
node, difficulty, current_mastery
)
recommendations.append({
"knowledge_point": node,
"difficulty": difficulty,
"estimated_minutes": estimated_time,
"urgency_score": urgency,
"prerequisites_met": all(
current_mastery.get(p, 0) >= 0.7
for p in self._get_prerequisites(node)
),
"target_goal": goal
})
# 去重并按紧迫度排序
seen = set()
unique_recs = []
for rec in sorted(recommendations, key=lambda x: -x["urgency_score"]):
if rec["knowledge_point"] not in seen:
seen.add(rec["knowledge_point"])
unique_recs.append(rec)
return unique_recs[:max_items]
def _find_learning_path(
self,
current_mastery: Dict[str, float],
target: str
) -> List[Tuple[str, float, int]]:
"""Dijkstra寻路算法"""
# 初始化距离和路径
distances = {k: float('inf') for k in self.graph.keys()}
distances[target] = 0
path = {target: None}
# 优先队列：(距离, 节点)
pq = [(0, target)]
while pq:
dist, node = heapq.heappop(pq)
if dist > distances[node]:
continue
for prerequisite, difficulty in self.graph.get(node, []):
# 难度权重：越难的边优先级越低
edge_weight = difficulty * 10
new_dist = dist + edge_weight
if new_dist < distances[prerequisite]:
distances[prerequisite] = new_dist
path[prerequisite] = node
heapq.heappush(pq, (new_dist, prerequisite))
# 重建路径
result = []
node = self._get_lowest_unmastered(current_mastery, target)
while node is not None:
difficulty = self._get_difficulty(node)
result.append((node, difficulty, int(difficulty * 30))) # 估算时间
node = path.get(node)
return list(reversed(result))
def _calculate_urgency(
self,
knowledge_point: str,
difficulty: float,
current_mastery: Dict[str, float]
) -> float:
"""计算学习紧迫度"""
mastery = current_mastery.get(knowledge_point, 0)
prerequisite_mastery = [
current_mastery.get(p, 0)
for p in self._get_prerequisites(knowledge_point)
]
# 紧迫度 = (1 - 掌握度) * 难度 * 前置满足率
prereq_satisfaction = (
sum(prerequisite_mastery) / len(prerequisite_mastery)
if prerequisite_mastery else 1.0
)
return (1 - mastery) * difficulty * prereq_satisfaction
def _get_prerequisites(self, node: str) -> List[str]:
"""获取前置知识点"""
return [p for p, _ in self.graph.get(node, [])]
def _get_difficulty(self, node: str) -> float:
"""获取知识点难度"""
edges = self.graph.get(node, [])
return max([d for _, d in edges], default=0.5)
def _get_lowest_unmastered(
self,
mastery: Dict[str, float],
target: str
) -> Optional[str]:
"""从目标倒推找到最低未掌握节点"""
node = target
while node is not None:
if mastery.get(node, 0) < 0.8:
return node
node = self._get_prerequisites(node)[0] if self._get_prerequisites(node) else None
return None三、性能优化与成本控制
在生产环境中，单学生每月可能产生2000+次评估请求，如果不加控制成本会失控。我实现了一套多层级优化策略。
3.1 批量评估与缓存机制
from functools import lru_cache
import hashlib
import asyncio
from typing import Optional
class AssessmentCache:
"""评估结果缓存 - 支持TTL和相似度匹配"""
def __init__(self, redis_client):
self.redis = redis_client
self.ttl_seconds = 3600 # 1小时缓存
def _make_key(self, question_hash: str, response_hash: str) -> str:
return f"assessment:{question_hash}:{response_hash}"
async def get(
self,
question_text: str,
student_response: str
) -> Optional[AssessmentResult]:
"""尝试从缓存获取"""
q_hash = hashlib.md5(question_text.encode()).hexdigest()[:12]
r_hash = hashlib.md5(student_response.encode()).hexdigest()[:12]
cached = await self.redis.get(self._make_key(q_hash, r_hash))
if cached:
return AssessmentResult(**json.loads(cached))
return None
async def set(
self,
question_text: str,
student_response: str,
result: AssessmentResult
):
"""写入缓存"""
q_hash = hashlib.md5(question_text.encode()).hexdigest()[:12]
r_hash = hashlib.md5(student_response.encode()).hexdigest()[:12]
await self.redis.setex(
self._make_key(q_hash, r_hash),
self.ttl_seconds,
json.dumps({
"mastery_level": result.mastery_level,
"misconceptions": result.misconceptions,
"suggested_activities": result.suggested_activities,
"confidence": result.confidence,
"tokens_used": result.tokens_used,
"processing_time_ms": result.processing_time_ms
})
)
class BatchedAssessmentProcessor:
"""批量评估处理器 - 减少API调用次数"""
def __init__(self, engine: AdaptiveAssessmentEngine, batch_size: int = 10):
self.engine = engine
self.batch_size = batch_size
self.queue: asyncio.Queue = asyncio.Queue()
self.results: Dict[str, AssessmentResult] = {}
async def start(self):
"""启动批处理worker"""
asyncio.create_task(self._process_loop())
async def submit(self, request_id: str, request: AssessmentRequest) -> AssessmentResult:
"""提交评估请求"""
await self.queue.put((request_id, request))
# 等待结果（超时机制）
for _ in range(50): # 最多等待5秒
await asyncio.sleep(0.1)
if request_id in self.results:
return self.results.pop(request_id)
raise TimeoutError(f"评估请求 {request_id} 超时")
async def _process_loop(self):
"""批量处理循环"""
while True:
batch = []
# 收集一批请求
while len(batch) < self.batch_size:
try:
request = await asyncio.wait_for(
self.queue.get(),
timeout=0.5
)
batch.append(request)
except asyncio.TimeoutError:
break
if not batch:
continue
# 批量处理
tasks = [
self.engine.evaluate(req, model="gpt-4.1-mini") # 批量用mini模型
for _, req in batch
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 分发结果
for i, (request_id, _) in enumerate(batch):
if isinstance(results[i], Exception):
self.results[request_id] = None
else:
self.results[request_id] = results[i]
智能模型路由
class ModelRouter:
"""
根据评估复杂度自动选择模型
- gpt-4.1: 复杂多知识点评估
- gpt-4.1-mini: 简单单知识点评估
- deepseek-v3.2: 批量初筛
"""
MODEL_COSTS = {
"gpt-4.1": {"input": 2.0, "output": 8.0, "latency_ms": 2000},
"gpt-4.1-mini": {"input": 0.4, "output": 1.6, "latency_ms": 500},
"deepseek-v3.2": {"input": 0.12, "output": 0.42, "latency_ms": 800}
}
def select_model(
self,
knowledge_points_count: int,
is_first_attempt: bool,
confidence_threshold: float = 0.9
) -> str:
"""智能选择模型"""
if knowledge_points_count >= 3 and is_first_attempt:
return "gpt-4.1" # 复杂评估用完整模型
elif knowledge_points_count == 1 and confidence_threshold > 0.8:
return "deepseek-v3.2" # 简单评估用便宜模型
else:
return "gpt-4.1-mini" # 标准评估用mini模型
def estimate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> float:
"""估算成本（美元）"""
prices = self.MODEL_COSTS[model]
return (input_tokens / 1_000_000 * prices["input"] +
output_tokens / 1_000_000 * prices["output"])3.2 并发控制与限流
import time
from collections import defaultdict
from contextlib import asynccontextmanager
class RateLimiter:
"""令牌桶限流器 - 支持多维度限流"""
def __init__(self, config: Dict):
# 每分钟令牌数限制
self.tokens_per_minute = config.get("tokens_per_minute", 1000)
# 每秒请求数限制
self.rpm_limit = config.get("rpm", 100)
# 每日成本上限（美元）
self.daily_cost_limit = config.get("daily_cost_limit", 100.0)
self.tokens = self.tokens_per_minute
self.last_refill = time.time()
self.rpm_counter = defaultdict(list)
self.daily_cost = 0.0
self.cost_reset_time = time.time()
async def acquire(self, estimated_tokens: int = 1000) -> bool:
"""获取执行许可"""
# 检查每日成本
if time.time() - self.cost_reset_time > 86400:
self.daily_cost = 0.0
self.cost_reset_time = time.time()
if self.daily_cost >= self.daily_cost_limit:
return False
# 令牌补充
now = time.time()
elapsed = now - self.last_refill
self.tokens = min(
self.tokens_per_minute,
self.tokens + elapsed * (self.tokens_per_minute / 60)
)
self.last_refill = now
if self.tokens < estimated_tokens:
return False
# RPM检查
current_second = int(now)
self.rpm_counter[current_second] = [
t for t in self.rpm_counter[current_second]
if current_second - t < 1
]
if len(self.rpm_counter[current_second]) >= self.rpm_limit:
return False
self.tokens -= estimated_tokens
self.rpm_counter[current_second].append(current_second)
return True
async def wait_and_acquire(self, estimated_tokens: int = 1000) -> None:
"""等待直到获取许可"""
for _ in range(30): # 最多等待30秒
if await self.acquire(estimated_tokens):
return
await asyncio.sleep(1)
raise RateLimitExceeded("限流超时，请稍后重试")
class CircuitBreaker:
"""熔断器 - 防止下游服务故障扩散"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: int = 60,
half_open_max_calls: int = 3
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self.failure_count = 0
self.last_failure_time = 0
self.state = "closed" # closed, open, half-open
self.half_open_calls = 0
async def call(self, func, *args, **kwargs):
"""带熔断保护的调用"""
if self.state == "open":
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = "half-open"
self.half_open_calls = 0
else:
raise CircuitBreakerOpen("熔断器开启，拒绝请求")
if self.state == "half-open":
if self.half_open_calls >= self.half_open_max_calls:
raise CircuitBreakerOpen("半开状态已达最大调用数")
self.half_open_calls += 1
try:
result = await func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _on_success(self):
self.failure_count = 0
self.state = "closed"
def _on_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "open"四、生产Benchmark数据
以下是我在生产环境实测的数据，基于100万次评估请求统计：
- 平均响应延迟：HolySheep API 42ms（国内直连），端到端评估 780ms
- P99延迟：1200ms（包含Redis缓存未命中场景）
- 缓存命中率：67%（相同题目+相似回答）
- Token消耗：平均每次评估 2800 tokens（输入1500 + 输出1300）
- 月度成本：10000学生 × 200次评估 = 200万次 × 2800/1M × $3（混合模型均价）= $168/月
# 性能监控Dashboard数据（Prometheus格式）
ASSESSMENT_LATENCY = """
HELP assessment_latency_seconds 评估延迟分布
TYPE assessment_latency_seconds histogram
assessment_latency_seconds_bucket{le="0.5"} 850000
assessment_latency_seconds_bucket{le="0.8"} 920000
assessment_latency_seconds_bucket{le="1.0"} 970000
assessment_latency_seconds_bucket{le="1.5"} 995000
assessment_latency_seconds_bucket{le="+Inf"} 1000000
assessment_latency_seconds_sum 780000
assessment_latency_seconds_count 1000000
HELP assessment_cost_daily 每日评估成本
TYPE assessment_cost_daily gauge
assessment_cost_daily 5.6
HELP assessment_cache_hit_ratio 缓存命中率
TYPE assessment_cache_hit_ratio gauge
assessment_cache_hit_ratio 0.67
"""五、常见报错排查
5.1 Token超限错误
# 错误信息
httpx.HTTPStatusError: 400 Client Error: Bad Request
{"error": {"message": "This model's maximum context length is 128000 tokens", "type": "invalid_request_error"}}
解决方案：添加输入截断逻辑
async def truncate_for_evaluation(
question: str,
response: str,
max_tokens: int = 8000
) -> tuple[str, str]:
"""智能截断超长输入"""
system_limit = 1000 # 保留给system prompt和结构
# 估算当前token数（粗略：中文≈2字符/token，英文≈4字符/token）
current_tokens = len(question) // 2 + len(response) // 2
if current_tokens <= max_tokens - system_limit:
return question, response
# 按比例截断
ratio = (max_tokens - system_limit) / current_tokens
new_q_len = int(len(question) * ratio * 0.6) # 问题占60%
new_r_len = int(len(response) * ratio * 0.4) # 回答占40%
return question[:new_q_len], response[:new_r_len]5.2 Rate Limit 429错误
# 错误信息
httpx.HTTPStatusError: 429 Client Error: Too Many Requests
{"error": {"message": "Rate limit reached for gpt-4.1", "type": "requests", "param": null, "code": "rate_limit_exceeded"}}
解决方案：实现指数退避重试
import random
async def call_with_retry(
func,
max_retries: int = 3,
base_delay: float = 1.0
) -> dict:
for attempt in range(max_retries):
try:
return await func()
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
# 指数退避 + 随机抖动
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
await asyncio.sleep(delay)
else:
raise
raise MaxRetriesExceeded("达到最大重试次数")5.3 评估结果JSON解析失败
# 错误信息
json.JSONDecodeError: Expecting property name enclosed in double quotes
解决方案：添加容错的JSON解析
import re
def safe_parse_json(response_text: str) -> Optional[dict]:
"""容错JSON解析"""
# 尝试直接解析
try:
return json.loads(response_text)
except json.JSONDecodeError:
pass
# 尝试提取markdown代码块
code_block_match = re.search(
r'``(?:json)?\s*([\s\S]*?)\s*``',
response_text
)
if code_block_match:
try:
return json.loads(code_block_match.group(1))
except json.JSONDecodeError:
pass
# 尝试修复常见格式问题
fixed = response_text.strip()
fixed = re.sub(r"'([^']+)':", r'"\1":', fixed) # 单引号转双引号
fixed = re.sub(r',\s*}', '}', fixed) # 尾部逗号
try:
return json.loads(fixed)
except json.JSONDecodeError:
# 返回默认结构
return {
"mastery_level": 0.5,
"misconceptions": ["解析失败，请人工复核"],
"suggested_activities": ["重新答题"],
"confidence": 0.1
}5.4 熔断器频繁触发
# 症状：评估服务不可用，但日志显示下游API正常
原因：熔断器阈值设置过低
解决：动态调整熔断器参数
class AdaptiveCircuitBreaker(CircuitBreaker):
def __init__(self):
super().__init__()
# 根据时间窗口动态调整阈值
self