AI模型服务监控:FastAPI中间件实现性能与元信息仪表盘
2026/8/27 4:03:07 网站建设 项目流程

1. 项目概述:为什么我们需要一个模型监控仪表盘?

在AI应用开发,尤其是大模型应用落地的过程中,我们常常会陷入一种“黑盒”状态。模型部署上线了,接口能调通,返回结果看起来也对,但心里总是不踏实:这个模型处理一次请求到底花了多长时间?它的响应速度稳定吗?随着用户量增长,性能瓶颈会出现在哪里?当前服务的版本是什么?健康状态如何?这些看似基础的问题,如果缺乏有效的监控手段,就变成了盲人摸象。

“实现模型响应耗时统计 + 基础元信息展示”这个项目,正是为了解决这个痛点。它不是一个复杂的性能分析平台,而是一个轻量级、高内聚的监控仪表盘。核心目标就两个:第一,精准地统计每一次模型推理的耗时,从毫秒级精度洞察性能波动;第二,清晰地展示服务的基础元信息,比如模型版本、服务启动时间、当前负载等,让开发者对服务状态一目了然。

这适合谁呢?如果你是算法工程师,刚把训练好的模型封装成API服务,需要评估其在线性能;如果你是后端开发,正在构建一个集成多个AI能力的应用,需要统一监控各个模型的响应情况;或者你是运维工程师,需要为AI服务添加可观测性指标。这个项目提供的思路和代码,可以直接集成到你的Flask、FastAPI、Django甚至是异步框架的服务中,用最小的成本获得最关键的性能透视能力。接下来,我会拆解从设计思路到代码落地的全过程,并分享我在实际部署中踩过的坑和总结的经验。

2. 整体设计与核心思路拆解

2.1 核心需求与方案选型

这个项目的需求非常明确,但实现路径有多种选择。我们需要统计耗时,那么就要在请求进入和离开时打点;我们需要展示元信息,就需要一个途径能随时获取这些信息。常见的方案有:

  1. 中间件(Middleware)拦截方案:这是最主流、侵入性较低的方式。在Web框架的请求处理管道中,插入一个中间件。请求到达时记录开始时间,响应返回时计算耗时,并存储起来。同时,该中间件也可以响应一个特定的监控端点(如/metrics/health)。
  2. 装饰器(Decorator)方案:在具体的模型预测函数上添加装饰器,专门统计该函数的执行时间。这种方式更精准,直接度量核心逻辑,但需要对每个需要监控的函数进行修饰,侵入性稍高。
  3. APM(应用性能监控)集成方案:使用像Prometheus、StatsD、Datadog等专业监控工具。功能强大,但架构复杂,对于只需要基础监控的小型服务来说略显笨重。

为什么我们选择中间件方案?对于模型服务来说,一个请求的耗时不仅包含模型前向推理的纯计算时间,还包括数据预处理、后处理、序列化/反序列化、网络IO等。中间件在框架层面拦截,能统计到整个HTTP请求/响应周期的总耗时,这反映了终端用户感受到的真实延迟,价值更大。此外,中间件天然适合统一处理所有路由,便于集中展示元信息,无需修改业务代码。

技术栈选择:我们以最常用的Python Web框架FastAPI为例进行实现。选择FastAPI是因为其高性能、易用性以及对异步的原生支持,在现代AI服务中应用广泛。监控数据存储,我们选择内存中的数据结构(如字典、列表),并配合简单的文件缓存或周期上报,以保持轻量。对于需要持久化和聚合分析的场景,可以很容易地扩展为写入数据库或推送到监控系统。

2.2 架构设计:数据流与模块职责

整个监控模块可以划分为三个核心部分:

  1. 数据采集器(Metrics Collector):核心是中间件。负责在request开始时记录时间戳、请求ID(用于追踪),在response结束时计算耗时,并将本次请求的元数据(路径、方法、状态码、耗时)存入一个线程安全的数据结构。同时,它自身也维护着服务的静态元信息(如版本号、启动时间戳)和动态元信息(如总请求数、最近N次请求平均耗时)。
  2. 数据存储器(Metrics Store):负责临时存放采集到的指标数据。考虑到并发访问,我们必须使用线程安全的数据结构,如threading.Lock保护的字典,或者直接使用collections.deque(双端队列)来保存最近N次的耗时记录,其appendpopleft操作是原子性的。对于历史数据,可以定期聚合(如计算1分钟内的P50、P95、P99耗时)后记录到日志文件或发送出去。
  3. 数据展示器(Metrics Exporter):提供一个HTTP端点(例如/admin/metrics),当访问该端点时,从数据存储器中读取实时数据,并组织成人类可读(HTML)或机器可读(JSON)的格式返回。基础元信息如服务版本、运行时长等也在此一并返回。

这个架构的优点是解耦清晰。采集器只管收数据,存储器只管存数据,展示器只管取数据并渲染。未来如果我们想把数据存到Redis里,或者用Prometheus的CounterHistogram来暴露指标,只需要替换存储器和展示器的部分实现即可,采集器的逻辑基本不变。

3. 核心细节解析与实操要点

3.1 高精度耗时统计的实现陷阱

统计耗时听起来简单,用time.time()减一下就行,但魔鬼在细节里。

第一,时间函数的选择。Python中常用的有:

  • time.time():返回自纪元(1970-01-01 UTC)以来的秒数(浮点数)。它受系统时钟调整的影响(如NTP同步),可能导致时间倒流或跳跃。
  • time.perf_counter():返回性能计数器的值(以秒为单位),用于测量短时间间隔。它具有最高可用分辨率,且是单调递增的(不会倒退),最适合用于基准测试和耗时统计。
  • time.process_time():返回当前进程的系统和用户CPU时间之和。不包含睡眠时间,适合测量CPU工作时间。

关键选择:对于网络服务响应耗时,我们应该使用time.perf_counter()。因为它测量的是墙上时钟时间(wall-clock time),反映了用户等待的真实时间,并且是单调的,避免了因系统时间调整而产生的负耗时这种荒谬情况。

第二,中间件的执行位置与异常处理。中间件必须确保即使视图函数抛出异常,耗时统计也能完成。这意味着结束时间的记录和耗时计算必须放在finally块或异步的try/except/finally结构中。否则,一旦应用内部报错,这次请求的耗时数据就会丢失,导致统计偏差(丢失的往往是耗时长的错误请求)。

第三,异步服务的特殊考量。在FastAPI等异步框架中,如果中间件和路径操作函数是异步的,使用time.perf_counter()仍然是正确的。但需要注意的是,在异步上下文中,如果存在await,可能会发生事件循环切换,perf_counter测量的是总耗时,这符合我们的需求。我们不需要使用asyncio的时钟函数。

3.2 线程安全的数据存储设计

Web服务器通常是多线程(如Gunicorn + sync worker)或多进程/异步(如Uvicorn + ASGI)的。我们的内存存储必须考虑并发安全。

方案一:使用threading.Lock这是最直接的方案。我们创建一个全局的存储字典和一个锁。

import threading import time from collections import deque # 全局存储和锁 _request_metrics_lock = threading.Lock() _request_metrics = { “total_requests”: 0, “total_time”: 0.0, “recent_durations”: deque(maxlen=1000) # 只保留最近1000次请求的耗时 } # 在中间件中更新数据 start_time = time.perf_counter() try: # ... 执行请求 ... finally: end_time = time.perf_counter() duration = end_time - start_time with _request_metrics_lock: # 获取锁,确保原子更新 _request_metrics[“total_requests”] += 1 _request_metrics[“total_time”] += duration _request_metrics[“recent_durations”].append(duration)

方案二:使用collections.dequedequeappendpopleft操作是线程安全的(得益于GIL,在单个操作上是原子的)。对于简单的追加历史记录场景,可以不用锁。但像“total_requests += 1”这种“读-改-写”操作不是原子的,仍需配合锁或使用threading.Atomic(Python 3.11+)?实际上Python没有内置的Atomic整数,所以对计数器的更新仍需锁。

实操心得:对于轻量级监控,我推荐“锁 + deque”组合。用锁保护几个关键聚合变量(总请求数、总耗时),用deque自动管理固定长度的历史耗时记录,避免内存无限增长。dequemaxlen参数非常有用,设成1000或5000,就实现了滑动窗口。

3.3 元信息的管理与展示

基础元信息分为静态和动态两类:

  • 静态信息:在服务启动时确定,后续不变。如:service_name,version(可从pyproject.toml__version__读取),model_name,model_version,startup_time
  • 动态信息:随时间变化。如:uptime(运行时长,由当前时间减startup_time计算),total_requests,average_latency(总耗时/总请求数),以及基于最近N次请求计算的性能分位数。

展示端点(如/admin/metrics)的设计:

  1. 格式:通常提供两种。application/json格式供其他系统(如健康检查、自动化监控)消费;text/html格式供人类在浏览器中直观查看。可以通过请求头Accept或查询参数format来区分。
  2. 内容:JSON格式应结构清晰。HTML格式则可以利用简单的表格和进度条,将耗时以毫秒为单位显示,并用颜色区分(如<100ms绿色,100-500ms黄色,>500ms红色),让状态一目了然。
  3. 性能:该端点本身不应该复杂计算影响性能。所有聚合计算(如平均耗时、分位数)可以在数据更新时异步计算,或者在该端点被请求时实时计算,但要注意如果历史数据量很大(如deque长度10万),实时计算P99可能较慢。因此,保持deque在一个合理的大小(如1000)是关键。

4. 基于FastAPI的完整实现与代码详解

下面我们一步步实现一个功能完整的监控中间件和展示端点。

4.1 项目结构与依赖

首先,创建一个简单的项目结构。我们主要需要两个文件:main.py(主应用)和monitoring.py(监控模块)。

your_model_service/ ├── main.py ├── monitoring.py └── requirements.txt

requirements.txt内容:

fastapi>=0.104.0 uvicorn[standard]>=0.24.0

4.2 监控模块(monitoring.py)实现

这是核心代码,我们将其模块化。

# monitoring.py import time import threading from collections import deque from typing import Dict, Any, Deque, Optional from contextlib import contextmanager from dataclasses import dataclass, asdict import json @dataclass class ServiceMetadata: """服务静态元数据""" service_name: str = “AI-Model-Service” version: str = “1.0.0” model_name: str = “gpt-2-small” model_version: str = “v1” startup_time: float = time.time() # 记录启动时刻的墙上时钟时间 class MetricsStore: """指标存储中心(线程安全)""" def __init__(self, window_size: int = 1000): self._lock = threading.Lock() self.window_size = window_size # 核心指标 self.total_requests: int = 0 self.total_duration: float = 0.0 self.recent_durations: Deque[float] = deque(maxlen=window_size) # 状态码统计(可选) self.status_codes: Dict[int, int] = {} def record_request(self, duration: float, status_code: int): """记录一次请求的耗时和状态码""" with self._lock: self.total_requests += 1 self.total_duration += duration self.recent_durations.append(duration) self.status_codes[status_code] = self.status_codes.get(status_code, 0) + 1 def get_summary(self) -> Dict[str, Any]: """获取指标摘要(计算平均耗时、分位数等)""" with self._lock: avg_latency = (self.total_duration / self.total_requests) if self.total_requests > 0 else 0.0 recent_list = list(self.recent_durations) # 计算P50, P90, P95, P99(百分位数) sorted_durations = sorted(recent_list) count = len(sorted_durations) def get_percentile(p: float) -> float: if count == 0: return 0.0 index = int(p * count) return sorted_durations[min(index, count - 1)] return { “total_requests”: self.total_requests, “average_latency_ms”: round(avg_latency * 1000, 2), # 转为毫秒 “p50_latency_ms”: round(get_percentile(0.5) * 1000, 2), “p90_latency_ms”: round(get_percentile(0.9) * 1000, 2), “p95_latency_ms”: round(get_percentile(0.95) * 1000, 2), “p99_latency_ms”: round(get_percentile(0.99) * 1000, 2), “recent_window_size”: count, “status_codes”: dict(self.status_codes), } # 全局单例实例 _metrics_store = MetricsStore() _service_meta = ServiceMetadata() def get_metrics_store() -> MetricsStore: return _metrics_store def get_service_metadata() -> ServiceMetadata: return _service_meta class TimingMiddleware: """统计耗时的中间件""" def __init__(self, app, metrics_store: MetricsStore): self.app = app self.metrics_store = metrics_store async def __call__(self, scope, receive, send): # 只处理HTTP请求 if scope[“type”] != “http”: await self.app(scope, receive, send) return start_time = time.perf_counter() status_code = 200 # 默认状态码,如果异常会被覆盖 # 定义一个自定义的send函数来拦截状态码 async def wrapped_send(message): nonlocal status_code if message[“type”] == “http.response.start”: status_code = message[“status”] await send(message) try: await self.app(scope, receive, wrapped_send) except Exception: # 如果发生异常,状态码可能在异常处理中设置,这里我们标记为500 # 更精细的做法可以从异常处理中间件获取状态码,这里简化处理 status_code = 500 raise finally: # 无论成功与否,都记录耗时 end_time = time.perf_counter() duration = end_time - start_time self.metrics_store.record_request(duration, status_code) def create_metrics_router(): """创建并返回一个包含监控端点的APIRouter""" from fastapi import APIRouter, Response from fastapi.responses import HTMLResponse import json router = APIRouter(tags=[“monitoring”]) @router.get(“/health”) async def health_check(): """基础健康检查端点""" return {“status”: “healthy”, “timestamp”: time.time()} @router.get(“/metrics”, summary=“获取服务监控指标”) async def get_metrics(format: Optional[str] = None, response: Response = None): """ 获取详细的性能指标和元信息。 可通过查询参数 `format=html` 获取人类可读的HTML页面。 默认返回JSON格式。 """ metrics_data = get_metrics_store().get_summary() meta = get_service_metadata() uptime = time.time() - meta.startup_time result = { “metadata”: { **asdict(meta), “uptime_seconds”: round(uptime, 2), “uptime_human”: _format_uptime(uptime), }, “metrics”: metrics_data, } # 根据format参数返回不同格式 if format == “html”: html_content = _generate_html_dashboard(result) return HTMLResponse(content=html_content) # 默认返回JSON return result return router def _format_uptime(seconds: float) -> str: """将秒数格式化为易读的字符串,如 '1天 03:45:20'""" days, remainder = divmod(int(seconds), 86400) hours, remainder = divmod(remainder, 3600) minutes, seconds = divmod(remainder, 60) if days > 0: return f“{days}天 {hours:02d}:{minutes:02d}:{seconds:02d}” else: return f“{hours:02d}:{minutes:02d}:{seconds:02d}” def _generate_html_dashboard(data: Dict) -> str: """生成一个简单的HTML监控面板""" meta = data[“metadata”] metrics = data[“metrics”] # 简单的HTML模板,实际中可以更美观 html = f“”“ <!DOCTYPE html> <html> <head> <title>服务监控面板 - {meta[‘service_name’]}</title> <style> body {{ font-family: sans-serif; margin: 20px; }} .card {{ border: 1px solid #ccc; border-radius: 5px; padding: 15px; margin-bottom: 15px; }} .metric-label {{ font-weight: bold; }} .metric-value {{ color: #333; }} .latency-good {{ color: green; }} .latency-warn {{ color: orange; }} .latency-bad {{ color: red; }} table {{ border-collapse: collapse; width: 100%%; }} th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }} th {{ background-color: #f2f2f2; }} </style> </head> <body> <h1>服务监控面板</h1> <div class=“card”> <h2>服务元信息</h2> <p><span class=“metric-label”>服务名称:</span> {meta[‘service_name’]}</p> <p><span class=“metric-label”>版本:</span> {meta[‘version’]}</p> <p><span class=“metric-label”>模型:</span> {meta[‘model_name’]} ({meta[‘model_version’]})</p> <p><span class=“metric-label”>启动时间:</span> {time.strftime(‘%Y-%m-%d %H:%M:%S’, time.localtime(meta[‘startup_time’]))}</p> <p><span class=“metric-label”>运行时长:</span> {meta[‘uptime_human’]}</p> </div> <div class=“card”> <h2>性能指标</h2> <table> <tr><th>指标</th><th>值</th></tr> <tr><td>总请求数</td><td>{metrics[‘total_requests’]}</td></tr> <tr><td>平均延迟</td><td>{metrics[‘average_latency_ms’]} ms</td></tr> <tr><td>P50延迟</td><td>{metrics[‘p50_latency_ms’]} ms</td></tr> <tr><td>P90延迟</td><td>{metrics[‘p90_latency_ms’]} ms</td></tr> <tr><td>P95延迟</td><td>{metrics[‘p95_latency_ms’]} ms</td></tr> <tr><td>P99延迟</td><td>{metrics[‘p99_latency_ms’]} ms</td></tr> <tr><td>滑动窗口大小</td><td>{metrics[‘recent_window_size’]}</td></tr> </table> </div> <div class=“card”> <h2>状态码分布</h2> <table> <tr><th>状态码</th><th>计数</th></tr> {“”.join(f“<tr><td>{code}</td><td>{count}</td></tr>” for code, count in metrics.get(‘status_codes’, {}).items())} </table> </div> <p><small>页面自动刷新: <a href=“javascript:location.reload()”>刷新</a> | <a href=“/metrics?format=json”>查看JSON</a></small></p> <script> // 每30秒自动刷新页面 setTimeout(() => location.reload(), 30000); </script> </body> </html> ”“” return html

4.3 主应用(main.py)集成

现在,在FastAPI主应用中集成我们的监控模块。

# main.py from fastapi import FastAPI from monitoring import TimingMiddleware, get_metrics_store, create_metrics_router import uvicorn # 1. 创建FastAPI应用实例 app = FastAPI(title=“AI模型服务”, version=“1.0.0”) # 2. 获取全局指标存储实例 metrics_store = get_metrics_store() # 3. 将自定义中间件添加到应用 # 注意:中间件的添加顺序很重要,TimingMiddleware应该尽可能早地添加,以便捕获最完整的耗时。 app.add_middleware(TimingMiddleware, metrics_store=metrics_store) # 4. 将监控路由挂载到应用 metrics_router = create_metrics_router() app.include_router(metrics_router, prefix=“/admin”) # 所有监控端点都在 /admin 路径下 # 5. 定义你的业务端点(模型预测端点) @app.post(“/predict”) async def predict(input_text: str): """ 模拟一个模型预测接口。 在实际应用中,这里会加载你的模型并进行推理。 """ # 模拟一些处理时间,比如模型推理 import asyncio import random # 模拟一个50ms到200ms之间的随机延迟 await asyncio.sleep(random.uniform(0.05, 0.2)) # 模拟返回结果 return {“result”: f“Processed: {input_text}”, “confidence”: random.random()} @app.get(“/“) async def root(): return {“message”: “AI Model Service is running. Check /admin/metrics for monitoring.”} if __name__ == “__main__”: # 启动服务 uvicorn.run(app, host=“0.0.0.0”, port=8000)

4.4 运行与测试

  1. 安装依赖:pip install -r requirements.txt
  2. 启动服务:python main.py
  3. 访问业务接口:用curl或浏览器访问http://localhost:8000/predict(POST方法,需传参) 或http://localhost:8000/
  4. 查看监控面板:
    • JSON格式指标http://localhost:8000/admin/metrics
    • HTML监控面板http://localhost:8000/admin/metrics?format=html
    • 健康检查http://localhost:8000/admin/health

多调用几次/predict接口后,再刷新监控面板,你就能看到总请求数、平均延迟、P95/P99延迟等指标在不断更新。HTML页面还会每30秒自动刷新,方便实时观察。

5. 生产环境进阶考量与问题排查

5.1 性能影响与优化

在中间件中增加计时和存储操作,理论上会对性能有极微小的损耗(每次请求多几次函数调用和锁操作)。但在实际测试中,这个损耗对于毫秒级响应的服务来说通常可以忽略不计(< 0.1ms)。如果实在担心,可以考虑以下优化:

  • 减少锁竞争MetricsStore.record_request中的锁保护了多个变量的更新。如果并发极高,可以尝试使用更细粒度的锁,或者使用__slots__来优化内存访问。但对于绝大多数场景,一个锁足够了。
  • 异步写入:将耗时的操作(如计算分位数、写入日志文件)放到后台线程或异步任务中执行,不阻塞主请求响应。例如,可以使用asyncio.create_task()来异步执行record_request中非关键的部分(如更新状态码分布),但核心的计数和deque操作仍需同步以保证数据一致性。
  • 采样:对于超高QPS(每秒数万请求)的服务,记录每一次请求的耗时可能产生大量数据。可以考虑采样,例如只记录1%的请求,或者只记录慢请求(如耗时大于100ms的)。

踩坑记录:我曾在一个QPS约3000的服务中直接记录了每次请求的完整URL路径到deque中,导致内存快速增长。切记,不要在每次请求中存储过大的对象(如完整的请求体、长URL)。只存储数值型的指标(耗时、状态码),必要时对路径进行聚合(如按路由模式聚合,而不是完整路径)。

5.2 数据持久化与可视化

内存存储的数据在服务重启后会丢失。对于长期监控,需要持久化。

  • 定期日志输出:最简单的方式是配置一个后台线程,每分钟将MetricsStore.get_summary()的结果以JSON格式打印到日志文件(如metrics.log)。然后可以用ELK(Elasticsearch, Logstash, Kibana)或Grafana Loki来收集和可视化日志。
  • 推送到时序数据库:更专业的做法是将数据推送到Prometheus、InfluxDB或TimescaleDB。我们需要将内存中的计数器(total_requests)和直方图(recent_durations)映射成Prometheus的CounterHistogram指标。可以创建一个额外的端点/admin/metrics/prometheus,返回Prometheus格式的指标数据,然后由Prometheus Server来抓取。
  • 集成OpenTelemetry:对于大型分布式系统,建议直接采用OpenTelemetry标准。OpenTelemetry提供了统一的API来收集指标、链路追踪和日志。我们可以用OpenTelemetry的Python SDK来创建Meter,记录请求耗时直方图,然后通过OTLP协议导出到Jaeger、Prometheus等后端。

5.3 常见问题排查实录

问题1:监控端点/admin/metrics访问变慢,甚至超时。

  • 可能原因get_summary()函数中,对recent_durations进行排序(sorted(recent_list))的操作复杂度是O(n log n)。如果window_size设置得非常大(例如10万),并且该端点被频繁访问,CPU消耗会很大。
  • 解决方案
    1. 限制窗口大小:将window_size保持在合理范围,如1000-5000。
    2. 缓存计算结果:在MetricsStore中增加一个缓存字段(如_cached_summary)和缓存时间戳。只有当recent_durations有更新,或者缓存超过一定时间(如5秒)后,才重新计算摘要。在get_summary()中先检查缓存有效性。
    3. 使用增量计算:维护一个排序后的数据结构(如bisect.insort插入已排序列表),但插入成本变高。需要权衡。

问题2:在多进程部署模式下(如Gunicorn + 多个worker),监控数据不准确,每个worker只有自己的数据。

  • 原因:我们的MetricsStore是进程内内存对象。Gunicorn等WSGI服务器会启动多个worker进程,每个进程有自己独立的内存空间,数据不共享。
  • 解决方案
    1. 使用进程间共享存储:将数据存储在外部的Redis或Memcached中。每个worker在记录和读取时都操作共享缓存。需要注意原子性问题,可以使用Redis的INCR命令和LPUSH/LTRIM列表命令来实现原子操作。
    2. 聚合展示:为每个worker分配一个独立的ID,在监控端点中,分别查询每个worker的指标(如果worker有独立的管理端口),或者通过一个中心化的Agent来收集所有worker的数据后再聚合展示。这种方式更复杂。
    3. 使用专门的监控Agent:放弃在应用内做聚合,每个worker只将原始指标数据(如单次请求耗时)发送到一个中心化的监控Agent(如StatsD daemon, Prometheus Pushgateway),由Agent负责聚合和存储。这是最推荐的生产环境做法。

问题3:记录的耗时包含了大文件上传/下载的时间,导致指标失真,无法反映模型本身的性能。

  • 原因:中间件统计的是整个HTTP请求的生命周期。如果有一个上传图片的接口,用户上传一个10MB的图片可能需要2秒,但这2秒主要是网络IO,不是模型推理时间。
  • 解决方案
    1. 路径排除:在中间件中,检查请求路径。如果是文件上传/下载的特定路径,可以选择不记录其耗时,或者记录到一个单独的指标集里。
    2. 装饰器辅助:在核心的模型推理函数上额外添加一个装饰器,专门记录纯模型推理时间。这样我们就有了两个指标:总请求耗时模型推理耗时。在监控面板上可以同时展示,对比分析。
    3. 流式响应处理:对于流式响应(如SSE, WebSocket),中间件的finally块可能在第一个数据块发送后就执行了,计时不准确。这种情况需要更复杂的处理,可能需要在响应生成器内部打点。

问题4:如何为不同的模型路由(如/v1/model_a/v1/model_b)分别统计指标?

  • 解决方案:在MetricsStore中,不要只用一个全局的recent_durations。可以改为一个嵌套字典,以路由路径(或路由标签)为键。
    self.route_metrics: Dict[str, Dict] = {} # 键为路由路径,值为该路由的指标字典
    record_request时,从请求的scope[“path”]中提取路径,然后更新对应路径的指标存储。在get_summary时,可以返回每个路由的独立摘要,也可以返回全局摘要。这样就能清晰看到哪个模型接口慢。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询