fix: Improved evals

CHANGELOG.md

@@ -11,12 +11,19 @@
 - Tool and subagent calls are now exposed as `artifact-update` A2A events so any SSE consumer can observe invocations, arguments, and results; opt-in per request via `userMessage.metadata["tool-status-update"]`; also added to preview
 - Preview now shows context size and token usage
 - OpenAI provider support via llm kind `openai`
+- Model and model parameters are shown in evaluation runs in MLflow
+
+### Changed
+
+- Consider static assertions in tests now in "output_correctness" and "success_rate" metrics of evaluation runs
 
 ### Fixed
 
 - Correctly handle non `@cap-js/mcp` tools during PII masking
 - Correctly report A2A messages as failed when AI Core is configured but no AI Core credentials can be found
 - Correctly report input tokens for Anthropic models on AI Core. OpenTelemetry specifies that input_tokens includes cached tokens, but Anthropic did not include the cached tokens in input_tokens
+- Summarize on timeout prompts are now correctly shown in MLflow
+- Evaluation run metrics are now correctly aggregated in MLflow
 - Thinking steps are now correctly separated also inside a turn
 - Tasks are now moved into failed state on server crash to avoid quota issue on unstable servers
 - `@agent.llm` is now correctly considered for feature toggled services

lib/agents/summarize-on-timeout.js

@@ -60,14 +60,17 @@ export async function summarizePartialWork({
       reason === "timeOut"
         ? `Agent task was not completed within its time limit. Write short, precise progress summary so user can decide whether to continue. State completed work and immediate next work. Do not claim work not shown. Start with: Agent did not finish within time! End with: Continue running or stop? No other questions to the user allowed!`
         : `Agent task was interrupted. Reason: ${reason}. Based on conversation history, provide brief summary of completed work and remaining work. Be concise.`
-    summaryPrompt += `\n\n Conversation Snippet: \n\n ${conversationSnippet.trim()}`
+    summaryPrompt += `\n\n Conversation Snippet: \n\n`
 
     let summaryTimer
     try {
       const response = await Promise.race([
         (async () => {
+          const instruction = new HumanMessage(summaryPrompt)
+          instruction.name = "cap-js-agents-summarize"
           const model = await getModel()
-          return model.invoke([new HumanMessage(summaryPrompt)])
+          // Two messages, so the first one can be logged as the prompt for summarization in MLflow
+          return model.invoke([instruction, new HumanMessage(conversationSnippet.trim())])
         })(),
         new Promise((_, reject) => {
           summaryTimer = setTimeout(() => reject(new Error("Summary LLM call timed out")), timeout)

lib/eval/eval-run.js

@@ -5,6 +5,7 @@ import {
   createEvalRun,
   closeEvalRun,
   logMlflowMetrics,
+  logMlflowRunMetadata,
 } from "../telemetry/mlflow/evaluation.js"
 import { flushMlflowTraces } from "../telemetry/mlflow/tracing.js"
 
@@ -23,6 +24,9 @@ export function evalRun(opts = {}) {
       runId,
       mlflowRunId,
       validationsByTask: new Map(),
+      metricKeys: new Set(),
+      mlflowMetadataLogged: false,
+      prompts: [],
     }
   }
 
@@ -34,15 +38,22 @@ export function evalRun(opts = {}) {
   })
 
   if (typeof afterEach === "function") {
-    afterEach(async () => {
-      if (state) await _flushValidations(state)
+    afterEach(async (testState) => {
+      if (state) {
+        await _flushValidations(state)
+        // Report test failure/success, so aggregated output_correctness respects static asserts
+        _addValidation(
+          { _evalState: state, taskId: "code_asserts" },
+          testState.task.result.state === "pass",
+        )
+      }
     })
   }
 
   afterAll(async () => {
     if (state) await _flushValidations(state)
     await flushMlflowTraces()
-    await closeEvalRun(state?.mlflowRunId).catch(() => {})
+    await closeEvalRun(state).catch(() => {})
     if (cds._activeEvalRun === state) cds._activeEvalRun = null
     state = null
   })
@@ -148,5 +159,40 @@ async function _postAssessmentScore(result, score, comment, config) {
 export async function logMlflowMetricsForResult(result, state = null) {
   state = state ?? cds._activeEvalRun
   if (!state?.mlflowRunId) return
-  await logMlflowMetrics(state.mlflowRunId, result.metrics).catch(() => {})
+  state.prompts = state.prompts.concat(_extractPrompts(result?.spans))
+  await _logMlflowRunMetadataOnce(result, state)
+  await logMlflowMetrics(state, result.metrics).catch(() => {})
+}
+
+// REVISIT: Properly log models as Registered Models in an eval Run to cover the change that an eval run contains multiple models
+async function _logMlflowRunMetadataOnce(result, state) {
+  if (state.mlflowMetadataLogged) return
+  const metadata = _extractMlflowRunMetadata(result?.spans)
+  if (!metadata) return
+  state.mlflowMetadataLogged = true
+  await logMlflowRunMetadata(state.mlflowRunId, metadata).catch(() => {})
+}
+
+function _extractMlflowRunMetadata(spans) {
+  const attrs = spans?.find(
+    (span) => span.attributes?.["gen_ai.operation.name"] === "chat",
+  )?.attributes
+  if (!attrs) return null
+
+  const model = attrs["gen_ai.response.model"]
+  const provider = attrs["gen_ai.provider.name"]
+  const params =
+    attrs["gen_ai.request.model_params"] && JSON.parse(attrs["gen_ai.request.model_params"])
+
+  if (!model && !provider && !Object.keys(params).length) return null
+  return { model, provider, params }
+}
+
+function _extractPrompts(spans) {
+  const attrs = spans?.find(
+    (span) => span.attributes?.["gen_ai.operation.name"] === "invoke_agent",
+  )?.attributes
+  if (!attrs) return []
+  const prompts = JSON.parse(attrs["mlflow.traceTag.mlflow.linkedPrompts"]) ?? []
+  return prompts
 }

lib/telemetry/chat-tracing.js

@@ -308,6 +308,8 @@ export function setLLMSpanStartAttrs(
   if (params?.temperature != null)
     span.setAttribute("gen_ai.request.temperature", params.temperature)
   if (params?.max_tokens != null) span.setAttribute("gen_ai.request.max_tokens", params.max_tokens)
+  const modelParams = pickModelParams(params)
+  if (modelParams) span.setAttribute("gen_ai.request.model_params", JSON.stringify(modelParams))
   if (cds.context?.["agent.context.id"])
     span.setAttribute("gen_ai.conversation.id", cds.context["agent.context.id"])
   if (streaming) span.setAttribute("gen_ai.request.stream", true)

lib/telemetry/mlflow/evaluation.js

@@ -6,17 +6,36 @@ export async function createEvalRun({ name } = {}) {
   if (!exporter) return null
   const creds = cds.env.requires?.mlflow?.credentials || {}
   const experimentId = creds.MLFLOW_EXPERIMENT_ID || process.env.MLFLOW_EXPERIMENT_ID || "0"
-  return exporter.createRun(experimentId, name)
+  const tags = [
+    ...(await getSourceTags()),
+    { key: "mlflow.user", value: "@cap-js/agents evaluation" },
+  ]
+  return exporter.createRun(experimentId, name, tags)
 }
 
-export async function closeEvalRun(runId) {
+export async function closeEvalRun({ mlflowRunId: runId, metricKeys = [], prompts = [] }) {
   if (!runId) return
-  getMlflowExporter()?.closeRun(runId)
+  const exporter = getMlflowExporter()
+  if (!exporter) return
+  await logFinalMlflowMetrics(runId, metricKeys, exporter)
+  const uniquePrompts = []
+  for (const prompt of prompts) {
+    if (!uniquePrompts.some((p) => p.version === prompt.version && p.name === prompt.name)) {
+      uniquePrompts.push(prompt)
+    }
+  }
+  if (uniquePrompts.length)
+    await exporter.linkPromptVersionsToRun(runId, uniquePrompts).catch(() => {})
+  await exporter.closeRun(runId)
 }
 
 // Log a flat metrics object; null/undefined values are skipped.
-export async function logMlflowMetrics(runId, metrics) {
+export async function logMlflowMetrics({ mlflowRunId: runId, metricKeys }, metrics) {
   if (!runId) return
+  metricKeys ??= new Set()
+  for (const [key, value] of Object.entries(metrics)) {
+    if (value != null) metricKeys.add(key)
+  }
   const exporter = getMlflowExporter()
   if (!exporter) return
   await Promise.allSettled(
@@ -26,6 +45,42 @@ export async function logMlflowMetrics(runId, metrics) {
   )
 }
 
+export async function logMlflowRunMetadata(runId, metadata, exporter = getMlflowExporter()) {
+  if (!runId || !exporter || !metadata) return
+  const params = {
+    ...(metadata.model && { "llm.model": metadata.model }),
+    ...(metadata.provider && { "llm.provider": metadata.provider }),
+  }
+  for (const [key, value] of Object.entries(metadata.params ?? {})) {
+    if (value != null) params[`llm.param.${key}`] = _stringifyMlflowValue(value)
+  }
+
+  await Promise.allSettled(
+    Object.entries(params).map(([key, value]) => exporter.logParam(runId, key, value)),
+  )
+}
+
+const AVG_METRICS = { success_rate: 1, output_correctness: 1, latency_ms: 1 }
+
+export async function logFinalMlflowMetrics(runId, metricKeys, exporter = getMlflowExporter()) {
+  if (!runId || !exporter) return
+  const keys = Array.from(metricKeys ?? []).concat(["success_rate", "output_correctness"])
+  await Promise.allSettled(
+    keys.map(async (key) => {
+      const history = await exporter.getMetricHistory(runId, key)
+      const values = history
+        .filter((metric) => metric?.step !== 1)
+        .map((metric) => Number(metric?.value))
+        .filter(Number.isFinite)
+      if (!values.length) return
+
+      const total = values.reduce((sum, value) => sum + value, 0)
+      const aggregate = AVG_METRICS[key] ? total / values.length : total
+      await exporter.logMetric(runId, key, aggregate, { step: 1 })
+    }),
+  )
+}
+
 export async function postMlflowAssessment(
   traceId,
   score,
@@ -36,3 +91,27 @@ export async function postMlflowAssessment(
 ) {
   getMlflowExporter()?.postAssessment(traceId, score, rationale, assessmentName, sourceId, opts)
 }
+
+function _stringifyMlflowValue(value) {
+  return typeof value === "string" ? value : JSON.stringify(value)
+}
+
+function isCI() {
+  return process.env.GITHUB_ACTIONS === "true"
+}
+
+async function getSourceTags() {
+  const serverUrl = process.env.GITHUB_SERVER_URL?.replace(/\/$/, "")
+  const sourceName =
+    serverUrl && process.env.GITHUB_REPOSITORY
+      ? `${serverUrl}/${process.env.GITHUB_REPOSITORY}`
+      : undefined
+  const branch = process.env.GITHUB_HEAD_REF
+  const commit = process.env.GITHUB_SHA
+  return [
+    sourceName && { key: "mlflow.source.name", value: sourceName },
+    { key: "mlflow.source.type", value: isCI() ? "JOB" : "LOCAL" },
+    branch && { key: "mlflow.source.git.branch", value: branch },
+    commit && { key: "mlflow.source.git.commit", value: commit },
+  ].filter(Boolean)
+}

lib/telemetry/mlflow/exporter/DatabricksExporter.js

@@ -1,6 +1,28 @@
 import { MlflowExporter } from "./MlflowExporter.js"
 
 export class DatabricksExporter extends MlflowExporter {
+  async getMetricHistory(runId, key) {
+    const metrics = []
+    let pageToken
+    do {
+      const query = new URLSearchParams({
+        run_id: runId,
+        metric_key: key,
+        max_results: 1000,
+        ...(pageToken && { page_token: pageToken }),
+      }).toString()
+      // eslint-disable-next-line no-await-in-loop
+      const data = await this._fetch(
+        `/api/2.0/mlflow/metrics/get-history?${query}`,
+        undefined,
+        "GET",
+      )
+      metrics.push(...(data?.metrics ?? []))
+      pageToken = data?.next_page_token
+    } while (pageToken)
+    return metrics
+  }
+
   async postAssessment(
     traceId,
     score,
@@ -31,23 +53,35 @@ export class DatabricksExporter extends MlflowExporter {
     })
   }
 
+  async linkPromptVersionsToRun(runId, prompts = []) {
+    if (!prompts.length) return
+    await super.linkPromptVersionsToRun(runId, prompts)
+    await this._fetch("/api/2.0/mlflow/unity-catalog/prompt-versions/links-to-runs", {
+      prompt_versions: prompts,
+      run_ids: [runId],
+    })
+  }
+
   // Returns { tags: [{key,value}], latestVersion: {version, tags} | null }.
   async ensurePrompt(name, description) {
     let res = await this._fetch(
-      `/mlflow/unity-catalog/prompts/${encodeURIComponent(name)}`,
+      `/api/2.0/mlflow/unity-catalog/prompts/${encodeURIComponent(name)}`,
       undefined,
       "GET",
     )
     if (!res) {
-      res = await this._fetch("/mlflow/unity-catalog/prompts", { name, prompt: { description } })
+      res = await this._fetch("/api/2.0/mlflow/unity-catalog/prompts", {
+        name,
+        prompt: { description },
+      })
     }
     const latestVersion = await this._getLatestUcVersion(name)
     return { tags: res?.tags ?? [], latestVersion }
   }
 
   async createPromptVersion(name, description, tags = [], template = "") {
     const res = await this._fetch(
-      `/mlflow/unity-catalog/prompts/${encodeURIComponent(name)}/versions`,
+      `/api/2.0/mlflow/unity-catalog/prompts/${encodeURIComponent(name)}/versions`,
       {
         prompt_version: { template, description, tags },
       },
@@ -56,7 +90,7 @@ export class DatabricksExporter extends MlflowExporter {
   }
 
   async setRegisteredModelTag(name, key, value) {
-    await this._fetch(`/mlflow/unity-catalog/prompts/${encodeURIComponent(name)}/tags`, {
+    await this._fetch(`/api/2.0/mlflow/unity-catalog/prompts/${encodeURIComponent(name)}/tags`, {
       key,
       value,
     })
@@ -65,7 +99,7 @@ export class DatabricksExporter extends MlflowExporter {
   // Returns { version, tags } of the latest UC prompt version, or null.
   async _getLatestUcVersion(name) {
     const res = await this._fetch(
-      `/mlflow/unity-catalog/prompts/${encodeURIComponent(name)}/versions/search`,
+      `/api/2.0/mlflow/unity-catalog/prompts/${encodeURIComponent(name)}/versions/search`,
       { max_results: 1 },
     )
     const pv = res?.prompt_versions?.[0]