文档

文档

Datadog + ModelRiver

统一的 AI 可观测性。将 ModelRiver 请求指标、追踪记录和错误事件发送至您的 Datadog 仪表盘。

概览

Datadog 是一个企业级可观测性平台。通过将 ModelRiver 请求数据转发给 Datadog，您可以建立仪表盘，直观地将 AI 的运行指标 (如延迟、词元 (tokens) 消耗量、提供商使用情况) 与您本地应用程序原有的运行遥测数据相互绑定和对比。

实现方式

ModelRiver 拥有一套属于自己的 内置可观测性 仪表盘与详尽的请求日志 (Request Logs)。如果要与 Datadog 实现集成，您需要对应用程序的代码进行相关的埋点处理，使得在每次发起针对 ModelRiver API 的请求调用前后，都能自动发送自定义相关的遥测指标或追踪标记记录。

利用 DogStatsD 上报自定义的各项性能指标

Python

PYTHON
1from datadog import statsd
2from openai import OpenAI
3import time
4 
5client = OpenAI(
6 base_url="https://api.modelriver.com/v1",
7 api_key="mr_live_YOUR_API_KEY",
8)
9 
10def chat_with_metrics(workflow: str, messages: list) -> str:
11 start = time.time()
12 tags = [f"workflow:{workflow}"]
13 
14 try:
15 response = client.chat.completions.create(
16 model=workflow,
17 messages=messages,
18 )
19 
20 duration = (time.time() - start) * 1000
21 tokens = response.usage.total_tokens
22 
23 # 发送相关指标数据
24 statsd.histogram("modelriver.request.duration_ms", duration, tags=tags)
25 statsd.increment("modelriver.request.count", tags=tags + ["status:success"])
26 statsd.histogram("modelriver.request.tokens", tokens, tags=tags)
27 
28 return response.choices[0].message.content
29 
30 except Exception as e:
31 duration = (time.time() - start) * 1000
32 statsd.increment("modelriver.request.count", tags=tags + ["status:error"])
33 statsd.histogram("modelriver.request.duration_ms", duration, tags=tags)
34 raise

Node.js

JAVASCRIPT
1import { StatsD } from "hot-shots";
2import OpenAI from "openai";
3 
4const statsd = new StatsD();
5const client = new OpenAI({
6 baseURL: "https://api.modelriver.com/v1",
7 apiKey: "mr_live_YOUR_API_KEY",
8});
9 
10async function chatWithMetrics(workflow, messages) {
11 const start = Date.now();
12 const tags = [`workflow:${workflow}`];
13 
14 try {
15 const response = await client.chat.completions.create({
16 model: workflow,
17 messages,
18 });
19 
20 const duration = Date.now() - start;
21 statsd.histogram("modelriver.request.duration_ms", duration, tags);
22 statsd.increment("modelriver.request.count", [...tags, "status:success"]);
23 statsd.histogram("modelriver.request.tokens", response.usage.total_tokens, tags);
24 
25 return response.choices[0].message.content;
26 } catch (error) {
27 statsd.increment("modelriver.request.count", [...tags, "status:error"]);
28 throw error;
29 }
30}

接入应用性能管理 (APM traces)

Python (使用 ddtrace)

PYTHON
1from ddtrace import tracer
2from openai import OpenAI
3 
4client = OpenAI(
5 base_url="https://api.modelriver.com/v1",
6 api_key="mr_live_YOUR_API_KEY",
7)
8 
9@tracer.wrap(service="modelriver", resource="chat")
10def chat(workflow: str, messages: list):
11 span = tracer.current_span()
12 span.set_tag("modelriver.workflow", workflow)
13 
14 response = client.chat.completions.create(
15 model=workflow,
16 messages=messages,
17 )
18 
19 span.set_tag("modelriver.tokens", response.usage.total_tokens)
20 span.set_tag("modelriver.model", response.model)
21 return response.choices[0].message.content

基于 Webhook 的收发与转投应用

您也可以在本地通过建立 Webhook 收集口的方式，将 ModelRiver 发来的带事件信息的 Webhook 数据，顺手发给 Datadog，当作记录留存。

PYTHON
1# 一个专门负责收取后发往 Datadog 的 FastAPI Webhook 后端示例程序
2from fastapi import FastAPI, Request
3import httpx
4 
5app = FastAPI()
6 
7DATADOG_LOG_URL = "https://http-intake.logs.datadoghq.com/api/v2/logs"
8DD_API_KEY = "YOUR_DATADOG_API_KEY"
9 
10@app.post("/webhooks/modelriver")
11async def forward_to_datadog(request: Request):
12 body = await request.json()
13 
14 # 转接上报给 Datadog Logs 集成口
15 async with httpx.AsyncClient() as http:
16 await http.post(
17 DATADOG_LOG_URL,
18 headers={"DD-API-KEY": DD_API_KEY},
19 json=[{
20 "ddsource": "modelriver",
21 "service": "ai-gateway",
22 "hostname": "modelriver",
23 "message": body,
24 }],
25 )
26 
27 return {"status": "ok"}

建议使用的仪表盘数据查选模块方式

图层模块形式所用测量基参数值具体能实现的意义目标
折线与追踪图 (Timeseries)modelriver.request.duration_ms查看各项响应情况的时间波动延迟走势
各项名榜单榜首榜单排列 (Top list)按 workflow 这个参选下的 modelriver.request.count看谁被调取或者运作的总执行点量数值占据最多名首居
分布态势热力色彩图 (Heat map)modelriver.request.tokensTokens 所产生掉烧散的利用状况排配大看视图表显示
基本计数计次大框图表器 (Counter)这个只按着 status:error 算时 modelriver.request.count{status:error}查询错误占比大发散出错比值
基本大统计单行总查显示出值框 (Query value)avg:modelriver.request.duration_ms显示长久以内的均摊调用时长反应等回候回应数

最佳实践指南

  1. 依工作流做各种上报标签: 如此这般能够让仪表盘图块和设立出来的各样报错事件通知有确凿具体的针对性。
  2. 盯随错误报告以及错误率: 让触发点专门就针对于那些 status:error 情况忽然异位长飙时作哨叫发警声。
  3. 同各项自家软件本有的那些调用路口线相关联在同一面网: 将属于该请求生成的那些时长轨迹记录 (spans) 打套嵌于自身应用的整个记录之内作为它里一部分展现。
  4. 把依靠于 Webhook 获取来的并利用之留存作为各项长录: 这是由于 Webhook 内送的信息都会富含详细并完整的向内各项挂包元数数据能够用作为最佳录本查询之源。
  5. 请专门确切为以上各种配好设置出专做哨卫的警管机哨口: 特别设定在如果某些处理应答总反应超出其时长许可值或是抛有错误异样巨幅时要向人员发生抛抛上报错发告提示等动作。

下一步的查阅探索与集成扩展指引