diff --git a/app/agent/orchestrator.py b/app/agent/orchestrator.py index 973413539..154bc5b1f 100644 --- a/app/agent/orchestrator.py +++ b/app/agent/orchestrator.py @@ -3,6 +3,7 @@ import hashlib import inspect import json import re +import time import traceback import uuid import warnings @@ -69,6 +70,7 @@ from app.agent.tools.impl.query_system_settings import QuerySystemSettingsTool from app.chain.agent import AgentChain from app.runtime.config import settings from app.runtime.events import eventmanager +from app.runtime.observability import record_metric from app.application.plugin.runtime import get_plugin_manager @@ -90,6 +92,37 @@ from app.foundation.identity import SYSTEM_INTERNAL_USER_ID warnings.filterwarnings("ignore", message=".*allowed_objects.*") +_KNOWN_AGENT_PROVIDER_TYPES = ( + "anthropic", + "azure", + "deepseek", + "gemini", + "ollama", + "openai", +) + + +def _agent_provider_metric_type(provider: object) -> str: + """把可配置 provider 名称收敛为有限指标类别，避免泄露自定义名称。""" + normalized = str(provider or "").strip().lower() + if not normalized: + return "unknown" + for provider_type in _KNOWN_AGENT_PROVIDER_TYPES: + if provider_type in normalized: + return provider_type + return "custom" + + +def _agent_task_metric_type(source: object, channel: object = None) -> str: + """把 Agent 来源归一为交互、调度、后台三类稳定标签。""" + normalized = str(source or "").strip().lower() + if normalized in {"scheduler", "scheduled", "heartbeat", "agent_task"}: + return "scheduled" + if channel: + return "interactive" + return "background" + + def _finish_processing_status(status: Optional[dict], user_id: Optional[str] = None) -> None: """结束入站消息的渠道处理状态。""" if not status: @@ -707,6 +740,24 @@ class MoviePilotAgent: self._session_usage.total_cache_write_input_tokens += cache_write_input_tokens self._session_usage.total_uncached_input_tokens += uncached_input_tokens self._session_usage.cache_usage_available |= cache_usage_available + provider_type = _agent_provider_metric_type( + (self._llm_provider_selection or {}).get("provider") + or settings.LLM_PROVIDER + ) + if input_tokens: + record_metric( + "agent.token_usage", + input_tokens, + provider_type=provider_type, + direction="input", + ) + if output_tokens: + record_metric( + "agent.token_usage", + output_tokens, + provider_type=provider_type, + direction="output", + ) if not is_current_request: return @@ -2279,6 +2330,7 @@ class MoviePilotAgent: """ execution_success = False execution_error: Optional[str] = None + metric_started_at = time.perf_counter() self._agent_started_at = datetime.now() self._llm_runtime_config = None self._llm_provider_selection = {} @@ -2435,6 +2487,15 @@ class MoviePilotAgent: await self._dispatch_execution_notice(friendly_message) return friendly_message, {} finally: + selection = self._llm_provider_selection or {} + record_metric( + "agent.provider.duration", + time.perf_counter() - metric_started_at, + provider_type=_agent_provider_metric_type( + selection.get("provider") or settings.LLM_PROVIDER + ), + outcome="success" if execution_success else "error", + ) self._send_agent_tokens_usage_event( success=execution_success, error=execution_error, @@ -2885,6 +2946,8 @@ class AgentManager: logger.debug(f"会话 {session_id} 的消息队列空闲，worker退出") break + task_type = _agent_task_metric_type(task.source, task.channel) + active_metric_recorded = False try: if task.enqueued_at is not None: queue_wait_ms = max( @@ -2900,6 +2963,12 @@ class AgentManager: 3, ) await self._start_task_processing_status(task) + record_metric( + "agent.active_tasks", + 1, + task_type=task_type, + ) + active_metric_recorded = True result = await self._process_message_internal(task) if task.completion_future and not task.completion_future.done(): if ( @@ -2923,6 +2992,12 @@ class AgentManager: if task.completion_future and not task.completion_future.done(): task.completion_future.set_exception(e) finally: + if active_metric_recorded: + record_metric( + "agent.active_tasks", + -1, + task_type=task_type, + ) await self._finish_task_processing_status(task) queue.task_done() if session_id in self._session_cancel_requested: @@ -3060,6 +3135,15 @@ class AgentManager: async def _stop_current_task_locked(self, session_id: str): """在 lifecycle 互斥域内停止会话 worker。""" stopped = False + active_agent = self.active_agents.get(session_id) + task_type = ( + _agent_task_metric_type( + getattr(active_agent, "source", None), + getattr(active_agent, "channel", None), + ) + if active_agent + else "unknown" + ) worker = self._session_workers.get(session_id) queue = self._session_queues.get(session_id) @@ -3101,6 +3185,12 @@ class AgentManager: else: logger.debug(f"会话 {session_id} 没有正在执行的Agent任务") + record_metric( + "agent.cancel", + task_type=task_type, + outcome="stopped" if stopped else "not_found", + ) + return stopped async def clear_session(self, session_id: str, user_id: str): diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 82a4bba03..698b1b8e7 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -866,6 +866,10 @@ OTel 初始化只能位于 Startup/Adapter；Domain/Application 只依赖 no-op- - `app.adapters.observability.otel` 只在组合根显式读取 `MOVIEPILOT_OTEL_METRICS=1` 后懒加载 OTel API； 未安装可选包时稳定回退 no-op，不给核心层增加 SDK 依赖。HTTP Adapter 通过路由匹配输出模板，绝不以原始 request path 充当 label。 +- 2026-08-22 扩展接线覆盖 SQLAlchemy checkout/checkin、异步回退配额 wait/timeout、Module 真实 + `TimeoutError`、插件 start/initialize/stop/reload，以及 Agent 活跃任务、取消结果、供应商耗时和输入/ + 输出 token。自定义 Agent provider 统一归类为 `custom`，不会暴露配置名称；Scheduler retry/dead-letter + 属于本轮明确暂停的 Outbox worker 范围，目录合同保留但不在本轮接线。 - 专项测试覆盖 exporter 缺失、非法标签、全目录高基数审计、成功/失败 outcome、动态 URL 路由模板； 既有 API、Event、Module、Scheduler 与健康探针回归保持通过。 diff --git a/tests/test_agent_tokens_events.py b/tests/test_agent_tokens_events.py index 364ac1aac..7421feafb 100644 --- a/tests/test_agent_tokens_events.py +++ b/tests/test_agent_tokens_events.py @@ -94,6 +94,36 @@ def test_initialize_llm_uses_chain_event_selection(monkeypatch) -> None: assert agent._llm_provider_selection["selected_provider_id"] == "provider-1" +def test_record_usage_emits_low_cardinality_token_metrics() -> None: + """Token 指标只暴露归一供应商类别与输入输出方向。""" + agent = MoviePilotAgent(session_id="usage-metrics", user_id="user-1") + agent._llm_provider_selection = {"provider": "private-provider-name"} + + with patch("app.agent.orchestrator.record_metric") as record_metric: + agent._record_usage( + { + "has_usage": True, + "input_usage_available": True, + "input_tokens": 120, + "output_tokens": 30, + "total_tokens": 150, + } + ) + + record_metric.assert_any_call( + "agent.token_usage", + 120, + provider_type="custom", + direction="input", + ) + record_metric.assert_any_call( + "agent.token_usage", + 30, + provider_type="custom", + direction="output", + ) + + def test_execute_agent_broadcasts_usage_on_success() -> None: """Agent 执行成功后应广播聚合 token 用量事件。""" agent = MoviePilotAgent(session_id="usage-success", user_id="user-1") diff --git a/tests/test_async_db_pooling.py b/tests/test_async_db_pooling.py index 4b2c191a0..b985b5390 100644 --- a/tests/test_async_db_pooling.py +++ b/tests/test_async_db_pooling.py @@ -14,6 +14,7 @@ Connection 与 aiosqlite 的线程都绑定在创建它的循环上。因此池 """ import asyncio import threading +from unittest.mock import patch import pytest @@ -163,8 +164,16 @@ def test_fallback_slot_times_out_when_exhausted(monkeypatch): async def run(): db_module._fallback_slots.acquire() # 占满唯一名额 - with pytest.raises(TimeoutError): - await db_module._acquire_fallback_slot() + with patch("app.db.session.record_metric") as record_metric: + with pytest.raises(TimeoutError): + await db_module._acquire_fallback_slot() + record_metric.assert_any_call("db.pool.timeout", backend="sqlite") + record_metric.assert_any_call( + "db.pool.wait", + pytest.approx(0.05, abs=0.03), + backend="sqlite", + outcome="timeout", + ) asyncio.run(run()) diff --git a/tests/test_observability.py b/tests/test_observability.py index c98885896..9020915c7 100644 --- a/tests/test_observability.py +++ b/tests/test_observability.py @@ -5,12 +5,16 @@ from typing import Mapping import httpx import pytest +from sqlalchemy import create_engine from starlette.applications import Starlette from starlette.responses import PlainTextResponse from starlette.routing import Route from app.adapters.observability import otel from app.adapters.web.metrics import HttpMetricsMiddleware +from app.db.engine import _register_database_pool_metrics +from app.runtime.extensions.plugin.lifecycle import observe_plugin_lifecycle +from app.schemas.plugin import PluginRuntimeStatus from app.runtime.observability import ( METRIC_SPECS, MetricSpec, @@ -137,3 +141,36 @@ def test_optional_otel_adapter_falls_back_to_noop(monkeypatch) -> None: monkeypatch.setattr(otel.importlib, "import_module", missing) assert isinstance(otel.build_observation_port(), NoopObservationPort) + + +def test_database_pool_checkout_updates_gauge() -> None: + """真实 SQLAlchemy checkout/checkin 应成对维护连接借出量。""" + port = RecordingObservationPort() + configure_observation(port) + engine = create_engine("sqlite://") + _register_database_pool_metrics(engine) + + with engine.connect(): + pass + engine.dispose() + + records = [record for record in port.records if record[0].name == "db.pool.checked_out"] + assert [record[1] for record in records] == [1.0, -1.0] + assert all(record[2] == {"backend": "sqlite"} for record in records) + + +def test_plugin_lifecycle_failed_status_records_error_outcome() -> None: + """被插件生命周期内部收敛的加载失败仍应记录 error。""" + port = RecordingObservationPort() + configure_observation(port) + + @observe_plugin_lifecycle("start") + def load_plugin() -> dict[str, PluginRuntimeStatus]: + """返回插件加载失败状态。""" + return {"Example": PluginRuntimeStatus.LOAD_FAILED} + + load_plugin() + + spec, _, labels = port.records[-1] + assert spec.name == "plugin.lifecycle.duration" + assert labels == {"operation": "start", "outcome": "error"}