i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 09:20:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: 增加内存保护措施，精简启动任务

Browse Source 
- 流式响应不再存储完整文本，仅记录长度用于 token 估算
- complete_response 文本累积增加 64KB 上限保护
- 通知缓冲区(email/webhook)增加溢出保护，超限丢弃旧通知
- 熔断器淘汰策略改进，全部 open/half-open 时按最旧失败时间淘汰
- 移除启动时清理任务和统计聚合回填，减少启动负担
- token 估算简化为基于长度的方法，避免持有完整文本
This commit is contained in:
fawney19
2026-03-10 16:08:11 +08:00
parent 6ec8df97e8
commit 7b0c80a0c4
5 changed files with 62 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
src/core/resilience.py
View File

@@ -208,11 +208,21 @@ class ResilienceManager:
def get_circuit_breaker(self, key: str) -> CircuitBreaker:
"""获取或创建熔断器"""
if key not in self.circuit_breakers:
# 淘汰已恢复的旧熔断器，防止无界增长
# 淘汰旧熔断器，防止无界增长
if len(self.circuit_breakers) >= self._MAX_CIRCUIT_BREAKERS:
# 优先淘汰已恢复(closed)的
closed_keys = [k for k, cb in self.circuit_breakers.items() if cb.state == "closed"]
for k in closed_keys:
del self.circuit_breakers[k]
if closed_keys:
for k in closed_keys:
del self.circuit_breakers[k]
else:
# 全部处于 open/half-open，按最后失败时间淘汰最旧的一半
sorted_keys = sorted(
self.circuit_breakers,
key=lambda cb_key: self.circuit_breakers[cb_key].last_failure_time or 0,
)
for k in sorted_keys[: len(sorted_keys) // 2 or 1]:
del self.circuit_breakers[k]
self.circuit_breakers[key] = CircuitBreaker()
return self.circuit_breakers[key]

7
src/plugins/notification/email.py
View File

@@ -55,6 +55,7 @@ class EmailNotificationPlugin(NotificationPlugin):
# 缓冲配置
self._buffer: list[Notification] = []
self._buffer_max_size = config.get("buffer_max_size", 500) if config else 500
self._lock = asyncio.Lock()
self._flush_task: asyncio.Task[None] | None = None
@@ -285,6 +286,12 @@ class EmailNotificationPlugin(NotificationPlugin):
"""
# 添加到缓冲区
async with self._lock:
# 缓冲区溢出保护：丢弃最旧的通知
if len(self._buffer) >= self._buffer_max_size:
drop_count = len(self._buffer) - self._buffer_max_size + 1
del self._buffer[:drop_count]
logger.warning("Email 通知缓冲区溢出，丢弃 {} 条旧通知", drop_count)
self._buffer.append(notification)
# 如果是严重通知，立即发送

7
src/plugins/notification/webhook.py
View File

@@ -52,6 +52,7 @@ class WebhookNotificationPlugin(NotificationPlugin):
# 缓冲配置
self._buffer: list[Notification] = []
self._buffer_max_size = config.get("buffer_max_size", 500) if config else 500
self._lock = asyncio.Lock()
self._session: aiohttp.ClientSession | None = None
self._flush_task = None
@@ -199,6 +200,12 @@ class WebhookNotificationPlugin(NotificationPlugin):
"""
# 添加到缓冲区
async with self._lock:
# 缓冲区溢出保护：丢弃最旧的通知
if len(self._buffer) >= self._buffer_max_size:
drop_count = len(self._buffer) - self._buffer_max_size + 1
del self._buffer[:drop_count]
logger.warning("Webhook 通知缓冲区溢出，丢弃 {} 条旧通知", drop_count)
self._buffer.append(notification)
# 如果是严重通知，立即发送

21
src/services/system/maintenance_scheduler.py
View File

@@ -144,15 +144,6 @@ class MaintenanceScheduler:
name="统计小时数据聚合",
timezone="UTC",
)
# 统计聚合补偿任务 - 每 30 分钟检查缺失并回填
scheduler.add_interval_job(
self._scheduled_stats_aggregation,
minutes=30,
job_id="stats_aggregation_backfill",
name="统计数据聚合补偿",
backfill=True,
)
# 清理任务 - 凌晨 3 点执行
scheduler.add_cron_job(
self._scheduled_cleanup,
@@ -254,24 +245,12 @@ class MaintenanceScheduler:
except Exception as e:
logger.debug("启动时刷新 Antigravity UA 版本失败（不影响运行）: {}", e)
try:
logger.info("启动时执行首次清理任务...")
await self._perform_cleanup()
except Exception as e:
logger.exception(f"启动时清理任务执行出错: {e}")
try:
logger.info("启动时清理残留的 pending/streaming 请求...")
await self._perform_pending_cleanup()
except Exception as e:
logger.exception(f"启动时 pending 清理执行出错: {e}")
try:
logger.info("启动时检查统计数据...")
await self._perform_stats_aggregation(backfill=True)
except Exception as e:
logger.exception(f"启动时统计聚合任务出错: {e}")
async def stop(self) -> Any:
"""停止调度器"""
if not self.running:

60
src/services/usage/stream.py
View File

@@ -7,7 +7,6 @@ from __future__ import annotations
import asyncio
import json
import re
from collections import deque
from collections.abc import AsyncIterator
from typing import Any
@@ -103,9 +102,11 @@ class StreamUsageTracker:
self.cache_read_input_tokens = 0
self.cache_creation_input_tokens_5m = 0
self.cache_creation_input_tokens_1h = 0
self.accumulated_content = ""
self._accumulated_content_len = 0 # 仅记录长度，不存储实际文本
# 完整响应跟踪（仅用于内部统计，不记录到数据库）
self._complete_response_content_chars = 0 # content 累计字符数
self._COMPLETE_RESPONSE_CONTENT_MAX_CHARS = 64 * 1024 # 64KB 上限
self.complete_response = {
"id": None,
"type": "message",
@@ -173,7 +174,7 @@ class StreamUsageTracker:
)
def _update_complete_response(self, chunk: dict[str, Any]) -> None:
"""根据响应块更新完整响应结构"""
"""根据响应块更新完整响应结构（文本累积受 64KB 上限保护）"""
try:
# 更新响应ID
if chunk.get("id"):
@@ -211,18 +212,26 @@ class StreamUsageTracker:
current_block = self.complete_response["content"][index]
# 超过上限后停止累积文本/JSON，仅保留前缀用于调试
over_limit = (
self._complete_response_content_chars
>= self._COMPLETE_RESPONSE_CONTENT_MAX_CHARS
)
if delta.get("type") == "text_delta":
# 文本增量
if current_block.get("type") == "text":
current_block["text"] = current_block.get("text", "") + delta.get(
"text", ""
)
text = delta.get("text", "")
self._complete_response_content_chars += len(text)
if not over_limit and current_block.get("type") == "text":
current_block["text"] = current_block.get("text", "") + text
elif delta.get("type") == "input_json_delta":
# 工具调用输入增量
if current_block.get("type") == "tool_use":
partial = delta.get("partial_json", "")
self._complete_response_content_chars += len(partial)
if not over_limit and current_block.get("type") == "tool_use":
current_input = current_block.get("input", {})
if isinstance(current_input, str):
current_input += delta.get("partial_json", "")
current_input += partial
current_block["input"] = current_input
elif event_type == "content_block_stop":
@@ -543,9 +552,9 @@ class StreamUsageTracker:
content, usage = self.parse_stream_chunk(chunk)
if content:
self.accumulated_content += content
self._accumulated_content_len += len(content)
# 实时估算输出tokens
self.output_tokens = max(1, len(self.accumulated_content) // 4)
self.output_tokens = max(1, self._accumulated_content_len // 4)
if usage:
# 如果响应中包含准确的usage信息，使用它
@@ -569,7 +578,7 @@ class StreamUsageTracker:
logger.debug(
f"ID:{self.request_id} | 流式响应结束 | 共处理{chunk_count}个chunks | "
f"累积内容长度:{len(self.accumulated_content)} | 输出tokens:{self.output_tokens}"
f"累积内容长度:{self._accumulated_content_len} | 输出tokens:{self.output_tokens}"
)
# 检查是否收到了有效数据
@@ -643,8 +652,8 @@ class StreamUsageTracker:
response_time_ms = None
# 如果没有准确的token计数，使用估算值
if self.output_tokens == 0 and self.accumulated_content:
self.output_tokens = max(1, len(self.accumulated_content) // 4)
if self.output_tokens == 0 and self._accumulated_content_len > 0:
self.output_tokens = max(1, self._accumulated_content_len // 4)
# 使用完整的响应体（包含所有信息，包括工具调用）
# 更新最终的usage信息
@@ -668,7 +677,7 @@ class StreamUsageTracker:
"stream": True,
"total_chunks": total_chunks,
"stored_chunks": stored_chunks,
"content_length": len(self.accumulated_content),
"content_length": self._accumulated_content_len,
"response_time_ms": response_time_ms,
}
if stored_chunks < total_chunks:
@@ -782,7 +791,7 @@ class StreamUsageTracker:
response_time_ms=response_time_ms,
status_code=self.status_code, # 使用实际的状态码
error_message=self.error_message, # 使用实际的错误消息
metadata={"stream": True, "content_length": len(self.accumulated_content)},
metadata={"stream": True, "content_length": self._accumulated_content_len},
request_body=self.request_data,
request_headers=self.request_headers,
provider_request_headers=self.provider_request_headers,
@@ -937,13 +946,14 @@ class EnhancedStreamUsageTracker(StreamUsageTracker):
except Exception as e:
logger.warning(f"Token encoding failed: {e}")
# 回退到估算方法
# 中文字符通常是2个token，英文单词约1.3个token
chinese_chars = len(re.findall(r"[\u4e00-\u9fff]", text))
english_words = len(re.findall(r"\b\w+\b", text))
# 回退到基于长度的估算
return self.count_tokens_by_len(len(text))
estimated_tokens = chinese_chars * 2 + english_words * 1.3
return max(1, int(estimated_tokens))
@staticmethod
def count_tokens_by_len(text_len: int) -> int:
"""基于文本长度估算 token 数（避免持有完整文本）"""
# 混合语言平均约 3 字符/token
return max(1, text_len // 3)
def estimate_input_tokens(self, messages: list) -> int:
"""
@@ -1048,9 +1058,9 @@ class EnhancedStreamUsageTracker(StreamUsageTracker):
content, usage = self.parse_stream_chunk(chunk)
if content:
self.accumulated_content += content
self._accumulated_content_len += len(content)
# 使用更准确的方法计算输出tokens
self.output_tokens = self.count_tokens(self.accumulated_content)
self.output_tokens = self.count_tokens_by_len(self._accumulated_content_len)
if usage:
# 如果响应中包含准确的usage信息，优先使用
@@ -1069,7 +1079,7 @@ class EnhancedStreamUsageTracker(StreamUsageTracker):
logger.debug(
f"ID:{self.request_id} | 流式响应结束 | 共处理{chunk_count}个chunks | "
f"累积内容长度:{len(self.accumulated_content)} | 输出tokens:{self.output_tokens}"
f"累积内容长度:{self._accumulated_content_len} | 输出tokens:{self.output_tokens}"
)
# 检查是否收到了有效数据