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

feat: 请求日志和 usage 记录中追踪代理节点信息

Browse Source 
- 新增 resolve_proxy_info/get_proxy_label 提取脱敏的代理摘要
- StreamContext 增加 proxy_info 字段，流式/非流式全路径写入 metadata
- ProxyNode 缓存补充 name 字段，日志输出代理节点标识
- aether-proxy 连接和转发日志从 debug 提升为 info 级别
- TaskService 代理不可用日志补充 node_id 细节
This commit is contained in:
fawney19
2026-02-08 00:49:43 +08:00
parent 5384ffd403
commit 4a1029961a
9 changed files with 153 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
src/api/handlers/base/chat_handler_base.py
View File

@@ -913,9 +913,15 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
# Capture the selected base_url from transport (used by some envelopes for failover).
ctx.selected_base_url = envelope.capture_selected_base_url() if envelope else None
# 记录代理信息
from src.clients.http_client import get_proxy_label, resolve_proxy_info
ctx.proxy_info = resolve_proxy_info(provider.proxy)
proxy_label = get_proxy_label(ctx.proxy_info)
logger.debug(
f" [{self.request_id}] 发送流式请求: Provider={provider.name}, "
f"模型={ctx.model} -> {mapped_model or '无映射'}"
f"模型={ctx.model} -> {mapped_model or '无映射'}, 代理={proxy_label}"
)
# If upstream is forced to non-stream mode, we execute a sync request and then
@@ -1307,6 +1313,10 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
# 失败时返回给客户端的是 JSON 错误响应
client_response_headers = {"content-type": "application/json"}
stream_fail_metadata: dict[str, Any] | None = None
if ctx.proxy_info:
stream_fail_metadata = {"proxy": ctx.proxy_info}
await self.telemetry.record_failure(
provider=ctx.provider_name or "unknown",
model=ctx.model,
@@ -1324,6 +1334,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
endpoint_api_format=ctx.provider_api_format or None,
has_format_conversion=ctx.has_format_conversion,
target_model=ctx.mapped_model,
request_metadata=stream_fail_metadata,
)
# ==================== 非流式处理 ====================
@@ -1372,6 +1383,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
endpoint_id: str | None = None # Endpoint ID（用于失败记录）
key_id: str | None = None # Key ID（用于失败记录）
mapped_model_result: str | None = None # 映射后的目标模型名（用于 Usage 记录）
sync_proxy_info: dict[str, Any] | None = None # 代理信息（用于 Usage 记录）
async def sync_request_func(
provider: Provider,
@@ -1382,6 +1394,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
nonlocal provider_name, response_json, status_code, response_headers
nonlocal provider_request_headers, provider_request_body, mapped_model_result
nonlocal provider_api_format_for_error, client_api_format_for_error, needs_conversion_for_error
nonlocal sync_proxy_info
provider_name = str(provider.name)
provider_api_format = str(endpoint.api_format or api_format)
@@ -1533,9 +1546,16 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
# 非流式：必须在 build_provider_url 调用后立即缓存（避免 contextvar 被后续调用覆盖）
selected_base_url_cached = envelope.capture_selected_base_url() if envelope else None
# 记录代理信息
from src.clients.http_client import get_proxy_label, resolve_proxy_info
sync_proxy_info = resolve_proxy_info(provider.proxy)
_proxy_label = get_proxy_label(sync_proxy_info)
logger.info(
f" [{self.request_id}] 发送{'上游流式(聚合)' if upstream_is_stream else '非流式'}请求: "
f"Provider={provider.name}, 模型={model} -> {mapped_model or '无映射'}"
f"Provider={provider.name}, 模型={model} -> {mapped_model or '无映射'}, "
f"代理={_proxy_label}"
)
logger.debug(f" [{self.request_id}] 请求URL: {redact_url_for_log(url)}")
@@ -1775,7 +1795,9 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
client_response_headers = filter_proxy_response_headers(response_headers)
client_response_headers["content-type"] = "application/json"
request_metadata = self._build_request_metadata()
request_metadata = self._build_request_metadata() or {}
if sync_proxy_info:
request_metadata["proxy"] = sync_proxy_info
total_cost = await self.telemetry.record_success(
provider=provider_name,
model=model,
@@ -1803,7 +1825,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
provider_api_key_id=key_id,
# 模型映射信息
target_model=mapped_model_result,
request_metadata=request_metadata,
request_metadata=request_metadata or None,
)
logger.debug(f"{self.FORMAT_ID} 非流式响应完成")
@@ -1826,7 +1848,9 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
# 记录实际发送给 Provider 的请求体，便于排查问题根因
response_time_ms = self.elapsed_ms()
actual_request_body = provider_request_body or original_request_body
request_metadata = self._build_request_metadata()
request_metadata = self._build_request_metadata() or {}
if sync_proxy_info:
request_metadata["proxy"] = sync_proxy_info
await self.telemetry.record_failure(
provider=provider_name or "unknown",
model=model,
@@ -1836,7 +1860,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
request_body=actual_request_body,
error_message=str(e),
is_stream=False,
request_metadata=request_metadata,
request_metadata=request_metadata or None,
)
client_format = (client_api_format_for_error or "").upper()
provider_format = (provider_api_format_for_error or client_format).upper()
@@ -1851,7 +1875,9 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
except UpstreamClientException as e:
response_time_ms = self.elapsed_ms()
actual_request_body = provider_request_body or original_request_body
request_metadata = self._build_request_metadata()
request_metadata = self._build_request_metadata() or {}
if sync_proxy_info:
request_metadata["proxy"] = sync_proxy_info
await self.telemetry.record_failure(
provider=provider_name or "unknown",
model=model,
@@ -1903,7 +1929,9 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
elif isinstance(e, httpx.HTTPStatusError) and hasattr(e, "response"):
error_response_headers = dict(e.response.headers)
request_metadata = self._build_request_metadata()
request_metadata = self._build_request_metadata() or {}
if sync_proxy_info:
request_metadata["proxy"] = sync_proxy_info
await self.telemetry.record_failure(
provider=provider_name or "unknown",
model=model,

40
src/api/handlers/base/cli_handler_base.py
View File

@@ -890,6 +890,12 @@ class CliMessageHandlerBase(BaseMessageHandler):
# Capture the selected base_url from transport (used by some envelopes for failover).
ctx.selected_base_url = envelope.capture_selected_base_url() if envelope else None
# 记录代理信息（sync-bridge 路径，早于流式路径执行）
from src.clients.http_client import get_proxy_label as _gpl
from src.clients.http_client import resolve_proxy_info as _rpi
ctx.proxy_info = _rpi(provider.proxy)
# If upstream is forced to non-stream mode, we execute a sync request and then
# simulate streaming to the client (sync -> stream bridge).
if not upstream_is_stream:
@@ -1073,12 +1079,14 @@ class CliMessageHandlerBase(BaseMessageHandler):
# 优先使用 Provider 配置，否则使用全局配置
request_timeout = provider.stream_first_byte_timeout or config.stream_first_byte_timeout
_proxy_label = _gpl(ctx.proxy_info)
logger.debug(
f" └─ [{self.request_id}] 发送流式请求: "
f"Provider={provider.name}, Endpoint={endpoint.id[:8] if endpoint.id else 'N/A'}..., "
f"Key=***{key.api_key[-4:] if key.api_key else 'N/A'}, "
f"原始模型={ctx.model}, 映射后={mapped_model or '无映射'}, URL模型={url_model}, "
f"timeout={request_timeout}s"
f"timeout={request_timeout}s, 代理={_proxy_label}"
)
# 创建 HTTP 客户端（支持代理配置，从 Provider 读取）
@@ -2802,6 +2810,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
mapped_model_result = None # 映射后的目标模型名（用于 Usage 记录）
response_metadata_result: dict[str, Any] = {} # Provider 响应元数据
needs_conversion = False # 是否需要格式转换（由 candidate 决定）
sync_proxy_info: dict[str, Any] | None = None # 代理信息
# 可变请求体容器：允许 TaskService 在遇到 Thinking 签名错误时整流请求体后重试
# 结构: {"body": 实际请求体, "_rectified": 是否已整流, "_rectified_this_turn": 本轮是否整流}
@@ -2813,7 +2822,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
key: ProviderAPIKey,
candidate: ProviderCandidate,
) -> dict[str, Any]:
nonlocal provider_name, response_json, status_code, response_headers, provider_api_format, provider_request_headers, provider_request_body, mapped_model_result, response_metadata_result, needs_conversion
nonlocal provider_name, response_json, status_code, response_headers, provider_api_format, provider_request_headers, provider_request_body, mapped_model_result, response_metadata_result, needs_conversion, sync_proxy_info
provider_name = str(provider.name)
provider_api_format = str(endpoint.api_format) if endpoint.api_format else ""
@@ -2948,11 +2957,18 @@ class CliMessageHandlerBase(BaseMessageHandler):
# 非流式：必须在 build_provider_url 调用后立即缓存（避免 contextvar 被后续调用覆盖）
selected_base_url_cached = envelope.capture_selected_base_url() if envelope else None
# 记录代理信息
from src.clients.http_client import get_proxy_label, resolve_proxy_info
sync_proxy_info = resolve_proxy_info(provider.proxy)
_proxy_label = get_proxy_label(sync_proxy_info)
logger.info(
f" └─ [{self.request_id}] 发送{'上游流式(聚合)' if upstream_is_stream else '非流式'}请求: "
f"Provider={provider.name}, Endpoint={endpoint.id[:8] if endpoint.id else 'N/A'}..., "
f"Key=***{key.api_key[-4:] if key.api_key else 'N/A'}, "
f"原始模型={model}, 映射后={mapped_model or '无映射'}, URL模型={url_model}"
f"原始模型={model}, 映射后={mapped_model or '无映射'}, URL模型={url_model}, "
f"代理={_proxy_label}"
)
# 获取复用的 HTTP 客户端（支持代理配置，从 Provider 读取）
@@ -3190,7 +3206,9 @@ class CliMessageHandlerBase(BaseMessageHandler):
client_response_headers = filter_proxy_response_headers(response_headers)
client_response_headers["content-type"] = "application/json"
request_metadata = self._build_request_metadata()
request_metadata = self._build_request_metadata() or {}
if sync_proxy_info:
request_metadata["proxy"] = sync_proxy_info
total_cost = await self.telemetry.record_success(
provider=provider_name,
model=model,
@@ -3219,7 +3237,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
target_model=mapped_model_result,
# Provider 响应元数据（如 Gemini 的 modelVersion）
response_metadata=response_metadata_result if response_metadata_result else None,
request_metadata=request_metadata,
request_metadata=request_metadata or None,
)
logger.info(f"{self.FORMAT_ID} 非流式响应处理完成")
@@ -3236,7 +3254,9 @@ class CliMessageHandlerBase(BaseMessageHandler):
# 记录实际发送给 Provider 的请求体，便于排查问题根因
response_time_ms = int((time.time() - sync_start_time) * 1000)
actual_request_body = provider_request_body or original_request_body
request_metadata = self._build_request_metadata()
request_metadata = self._build_request_metadata() or {}
if sync_proxy_info:
request_metadata["proxy"] = sync_proxy_info
await self.telemetry.record_failure(
provider=provider_name or "unknown",
model=model,
@@ -3247,7 +3267,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
error_message=str(e),
is_stream=False,
api_format=api_format,
request_metadata=request_metadata,
request_metadata=request_metadata or None,
)
raise
@@ -3272,7 +3292,9 @@ class CliMessageHandlerBase(BaseMessageHandler):
elif isinstance(e, httpx.HTTPStatusError) and hasattr(e, "response"):
error_response_headers = dict(e.response.headers)
request_metadata = self._build_request_metadata()
request_metadata = self._build_request_metadata() or {}
if sync_proxy_info:
request_metadata["proxy"] = sync_proxy_info
await self.telemetry.record_failure(
provider=provider_name or "unknown",
model=model,
@@ -3292,7 +3314,7 @@ class CliMessageHandlerBase(BaseMessageHandler):
has_format_conversion=is_format_converted(provider_api_format, str(api_format)),
# 模型映射信息
target_model=mapped_model_result,
request_metadata=request_metadata,
request_metadata=request_metadata or None,
)
raise

4
src/api/handlers/base/stream_context.py
View File

@@ -112,6 +112,9 @@ class StreamContext:
perf_sampled: bool = False
perf_metrics: dict[str, Any] = field(default_factory=dict)
# 代理信息（用于 usage 记录和日志）
proxy_info: dict[str, Any] | None = None
# 流式格式转换状态（跨 chunk 追踪）
stream_conversion_state: StreamState | None = None
@@ -142,6 +145,7 @@ class StreamContext:
self.response_id = None
self.final_usage = None
self.final_response = None
self.proxy_info = None
self.stream_conversion_state = None
self.needs_conversion = False
self.selected_base_url = None

6
src/api/handlers/base/stream_telemetry.py
View File

@@ -201,6 +201,8 @@ class StreamTelemetryRecorder:
metadata: dict[str, Any] = {"stream": True, "content_length": ctx.data_count}
if ctx.perf_metrics:
metadata["perf"] = ctx.perf_metrics
if ctx.proxy_info:
metadata["proxy"] = ctx.proxy_info
await writer.record_success(
provider=ctx.provider_name or "unknown",
@@ -250,6 +252,8 @@ class StreamTelemetryRecorder:
metadata: dict[str, Any] = {"stream": True, "content_length": ctx.data_count}
if ctx.perf_metrics:
metadata["perf"] = ctx.perf_metrics
if ctx.proxy_info:
metadata["proxy"] = ctx.proxy_info
await writer.record_failure(
provider=ctx.provider_name or "unknown",
@@ -297,6 +301,8 @@ class StreamTelemetryRecorder:
metadata: dict[str, Any] = {"stream": True, "content_length": ctx.data_count}
if ctx.perf_metrics:
metadata["perf"] = ctx.perf_metrics
if ctx.proxy_info:
metadata["proxy"] = ctx.proxy_info
await writer.record_cancelled(
provider=ctx.provider_name or "unknown",