fix: 速度胶囊每轮只显示一次 + 技能调用按 run 合并统计

速度胶囊:原本每条 AI 消息都渲染,agentic 多轮 LLM 调用会出现多个胶囊。改为只在每轮 turn-final 消息(下一条 human 之前的最后一条 AI 消息)显示。

技能统计:tool_call_metrics 改为按 (run_id, skill_name) upsert 合并——同一 run 内同一技能的多次调用/重试合并为一行;任一成功即记成功,全部失败则记一行 error 并累积所有报错信息;非技能调用不受影响。

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
吉鑫升andClaude Opus 4.7 committed 2026-05-19 17:22:17 +08:00
1 parent 4c38055d38
commit 028c5a34e6
5 files changed
+167 -17

No files matched your search

@@ -88,7 +88,13 @@ export function MessageTokenUsage({
// otherwise wait until usage arrives so the bar doesn't render on its own.
const usage = getUsageMetadata(message);
const showUsage = enabled && !isLoading && Boolean(usage);
const showSpeed = speedEnabled && resolvedMetric !== undefined && resolvedMetric.durationMs > 0;
// The speed pill is shown once per turn — only on the turn-final AI message
// — so a turn with several LLM calls no longer renders a pill per call.
const showSpeed =
speedEnabled &&
resolvedMetric !== undefined &&
resolvedMetric.durationMs > 0 &&
resolvedMetric.isTurnFinal;
if (!showUsage && !showSpeed) {
return null;
@@ -36,6 +36,13 @@ export interface MessageMetric {
* only to add a "~" prefix when the value is approximate.
*/
fromUsage: boolean;
/**
* True iff this is the turn-final AI message — the last AI message before
* the next human message. An agentic turn makes several LLM calls, each its
* own AI message; the speed pill is shown only on the turn-final one so a
* turn renders a single pill instead of one per call.
*/
isTurnFinal: boolean;
}
type Sample = { at: number; chars: number };
@@ -161,6 +168,25 @@ export function useMessageMetrics(
return () => window.clearInterval(interval);
}, [isLoading]);
// Identify the turn-final AI message of each turn — the last AI message
// (with a recorded metric) before the next human message. Only these show
// the speed pill, collapsing a multi-LLM-call turn to a single pill.
const turnFinalIds = new Set<string>();
let pendingTurnFinal: string | null = null;
for (let i = 0; i < messages.length; i++) {
const message = messages[i];
if (!message) continue;
if (message.type === "human") {
if (pendingTurnFinal) turnFinalIds.add(pendingTurnFinal);
pendingTurnFinal = null;
continue;
}
if (isAiMessage(message) && timingRef.current.has(getMessageId(message, i))) {
pendingTurnFinal = getMessageId(message, i);
}
}
if (pendingTurnFinal) turnFinalIds.add(pendingTurnFinal);
// Build the snapshot map. We compute on every render so consumers see the
// freshest duration; the underlying timing storage is the source of truth.
const out = new Map<string, MessageMetric>();
@@ -212,6 +238,7 @@ export function useMessageMetrics(
durationMs,
isActive,
fromUsage,
isTurnFinal: turnFinalIds.has(id),
});
}
+1 -1
View File
@@ -173,7 +173,7 @@ Lead-agent middlewares are assembled in strict append order across `packages/har
16. **SubagentLimitMiddleware** - Truncates excess `task` tool calls from model response to enforce `MAX_CONCURRENT_SUBAGENTS` limit (optional, if `subagent_enabled`)
17. **LoopDetectionMiddleware** - Detects repeated tool-call loops; hard-stop responses clear both structured `tool_calls` and raw provider tool-call metadata before forcing a final text answer
18. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last)
8a. **ToolMetricsMiddleware** - Appended immediately after `ToolErrorHandlingMiddleware` in `_build_runtime_middlewares`, so it sits *deepest* in the `wrap_tool_call` chain. Records one `tool_call_metrics` row per tool/skill/MCP invocation — timing duration and bucketing failures into a coarse `error_category` (rate_limited / ip_blocked / auth_error / timeout / network_error / not_found / empty_result / tool_error). It catches both raised exceptions (before #8 converts them) and *self-handled* errors returned as error `ToolMessage`s (search tools that swallow HTTP 429/403). It also captures the targeted Agent Skill into `tool_call_metrics.skill_name`: the lead agent invokes a skill by `read_file`-ing its `/mnt/skills/<category>/<name>/SKILL.md` path (per the `<skill_system>` prompt block — there is no dedicated "use skill" tool), so `skill_name` is recovered by scanning every tool call's string arguments for the `skills/{public,custom}/{name}/` path shape (`_SKILL_PATH_RE`), plus the explicit `name` argument of `skill_view`/`skill_manage`/`skill_archive`. `skill_view`-specific success detection handles its swallowed failures (`Skill 'x' not found.` returned as a plain string rather than raised). Regression coverage: `tests/test_tool_metrics_middleware.py`. Writes are best-effort fire-and-forget. Surfaced by the admin `GET /api/admin/tool-metrics` router (which has a `kind=skill` filter and a per-skill `/summary` breakdown).
8a. **ToolMetricsMiddleware** - Appended immediately after `ToolErrorHandlingMiddleware` in `_build_runtime_middlewares`, so it sits *deepest* in the `wrap_tool_call` chain. Records one `tool_call_metrics` row per tool/skill/MCP invocation — timing duration and bucketing failures into a coarse `error_category` (rate_limited / ip_blocked / auth_error / timeout / network_error / not_found / empty_result / tool_error). It catches both raised exceptions (before #8 converts them) and *self-handled* errors returned as error `ToolMessage`s (search tools that swallow HTTP 429/403). It also captures the targeted Agent Skill into `tool_call_metrics.skill_name`: the lead agent invokes a skill by `read_file`-ing its `/mnt/skills/<category>/<name>/SKILL.md` path (per the `<skill_system>` prompt block — there is no dedicated "use skill" tool), so `skill_name` is recovered by scanning every tool call's string arguments for the `skills/{public,custom}/{name}/` path shape (`_SKILL_PATH_RE`), plus the explicit `name` argument of `skill_view`/`skill_manage`/`skill_archive`. `skill_view`-specific success detection handles its swallowed failures (`Skill 'x' not found.` returned as a plain string rather than raised). Regression coverage: `tests/test_tool_metrics_middleware.py`. Writes are best-effort fire-and-forget. **Per-run skill merging**: `SqlToolMetricsStore.record` upserts skill rows by `(run_id, skill_name)` instead of inserting one row per call — a skill touched by several tool calls / retries inside one run collapses to a single row. Any success in the run wins (a skill that failed then finally succeeded is recorded as one `success`); a skill that only ever failed yields one `error` row whose `error_message` accumulates every attempt's message; durations are summed. Non-skill tool calls, and skill calls with no `run_id`, keep one row each. The read→merge→write is guarded by a per-event-loop `asyncio.Lock`. Surfaced by the admin `GET /api/admin/tool-metrics` router (which has a `kind=skill` filter and a per-skill `/summary` breakdown).
9. **PromptPrefixMiddleware** - Injects the admin-configured prompt prefix into the model request only. The prefix is tagged onto the first human message's `additional_kwargs['prompt_prefix']` by the Gateway (`app.gateway.services`) and prepended to content here at call time, so the persisted/displayed message stays prefix-free
10. **SummarizationMiddleware** - Context reduction when approaching token limits (optional, if enabled)
11. **TodoListMiddleware** - Task tracking with `write_todos` tool (optional, if plan_mode)
@@ -5,6 +5,14 @@ writes one row recording duration and outcome. Failures are bucketed into a
coarse ``error_category`` (rate_limited / ip_blocked / auth_error / timeout /
network_error / not_found / empty_result / tool_error) so administrators can
spot retrieval outages, blocked IPs, and quota problems at a glance.
**Skill-call merging**: a skill is often touched by several tool calls inside
one run (read ``SKILL.md``, run a script, retry after a failure). Rather than
one row per call, ``SqlToolMetricsStore.record`` upserts a single row per
``(run_id, skill_name)``: any success in the run wins (a skill that failed then
finally succeeded is recorded as one success), and a skill that only ever
failed yields one ``error`` row whose ``error_message`` accumulates every
attempt's message. Non-skill tool calls keep one row each.
"""
from deerflow.persistence.tool_metrics.base import ToolMetricRecord, ToolMetricsStore
@@ -2,6 +2,7 @@
from __future__ import annotations
import asyncio
import logging
from collections.abc import AsyncIterator
from datetime import UTC, datetime
@@ -15,11 +16,99 @@ from deerflow.persistence.tool_metrics.model import ToolCallMetricRow
logger = logging.getLogger(__name__)
# Separator joining every failed attempt's message inside one merged skill row.
_ERROR_SEP = "\n--- \n"
# Hard cap on the accumulated ``error_message`` of a merged row.
_MERGED_ERROR_LIMIT = 2000
# A single skill invocation often spans several tool calls (read SKILL.md, run
# a script, ...), and the model may retry a skill within the same run. Those
# fire-and-forget writes can interleave on the event loop, so the read → merge →
# write of a skill row is guarded by a lock. ``asyncio.Lock`` binds to the loop
# it is used on; the lock is therefore (re)created whenever the running loop
# changes (the Gateway keeps one persistent loop, while the ``asyncio.run``
# fallback path spins up a throwaway loop per call — already serialized).
_merge_lock: asyncio.Lock | None = None
_merge_lock_loop: object = None
def _skill_merge_lock() -> asyncio.Lock:
global _merge_lock, _merge_lock_loop
loop = asyncio.get_running_loop()
if _merge_lock is None or _merge_lock_loop is not loop:
_merge_lock = asyncio.Lock()
_merge_lock_loop = loop
return _merge_lock
def _normalize_created_at(value: datetime) -> datetime:
return value if value.tzinfo else value.replace(tzinfo=UTC)
def _merge_error_messages(existing: str | None, new: str | None) -> str | None:
"""Append a failed attempt's message to the accumulated error text.
All failures are kept ("都统计下来"), separated and capped at
:data:`_MERGED_ERROR_LIMIT` characters.
"""
new = (new or "").strip()
if not new:
return existing
existing = (existing or "").strip()
if not existing:
return new
if new in existing:
# Identical retry failure — no point storing the same line twice.
return existing
combined = existing + _ERROR_SEP + new
if len(combined) > _MERGED_ERROR_LIMIT:
combined = combined[: _MERGED_ERROR_LIMIT - 3] + "..."
return combined
def _merge_skill_row(existing: ToolCallMetricRow, metric: ToolMetricRecord) -> None:
"""Fold another same-run, same-skill call into an existing metric row.
Outcome rules:
- any success in the run -> the row becomes a single ``success`` and earlier
failures are dropped ("先报错、最后成功 -> 记一次成功");
- all calls failed -> a single ``error`` row whose ``error_message``
accumulates every attempt's message ("一直报错 -> 记一次,报错信息都统计下来").
"""
existing.duration_ms = (existing.duration_ms or 0) + (metric.duration_ms or 0)
if metric.status == "success":
if existing.status != "success":
existing.status = "success"
existing.error_category = None
existing.error_message = None
return
# metric.status == "error"
if existing.status == "success":
# The skill already succeeded once this run — keep that outcome.
return
existing.status = "error"
if metric.error_category:
existing.error_category = metric.error_category
existing.error_message = _merge_error_messages(existing.error_message, metric.error_message)
def _row_from_metric(metric: ToolMetricRecord) -> ToolCallMetricRow:
return ToolCallMetricRow(
id=metric.id,
created_at=_normalize_created_at(metric.created_at),
user_id=metric.user_id,
thread_id=metric.thread_id,
run_id=metric.run_id,
agent_name=metric.agent_name,
tool_name=metric.tool_name,
skill_name=metric.skill_name,
duration_ms=metric.duration_ms,
status=metric.status,
error_category=metric.error_category,
error_message=metric.error_message,
)
class SqlToolMetricsStore(ToolMetricsStore):
"""SQLite/Postgres/MySQL-backed implementation; writes are fire-and-forget."""
@@ -27,28 +116,48 @@ class SqlToolMetricsStore(ToolMetricsStore):
self._sf = session_factory
async def record(self, metric: ToolMetricRecord) -> None:
row = ToolCallMetricRow(
id=metric.id,
created_at=_normalize_created_at(metric.created_at),
user_id=metric.user_id,
thread_id=metric.thread_id,
run_id=metric.run_id,
agent_name=metric.agent_name,
tool_name=metric.tool_name,
skill_name=metric.skill_name,
duration_ms=metric.duration_ms,
status=metric.status,
error_category=metric.error_category,
error_message=metric.error_message,
)
# Skill calls are merged per run: the model may touch one skill across
# several tool calls / retries, and the admin wants a single row per
# (run, skill) instead of one per call. Non-skill tool calls, and skill
# calls with no run_id, fall back to a plain insert.
if metric.skill_name and metric.run_id:
await self._record_skill_merged(metric)
return
try:
async with self._sf() as session:
session.add(row)
session.add(_row_from_metric(metric))
await session.commit()
except Exception:
# Metrics are best-effort; never surface a write failure to the runtime.
logger.warning("Failed to persist tool_call_metrics row %s", metric.id, exc_info=True)
async def _record_skill_merged(self, metric: ToolMetricRecord) -> None:
"""Upsert a per-(run, skill) row, folding repeated calls into one."""
try:
async with _skill_merge_lock():
async with self._sf() as session:
existing = await session.scalar(
select(ToolCallMetricRow)
.where(
ToolCallMetricRow.run_id == metric.run_id,
ToolCallMetricRow.skill_name == metric.skill_name,
)
.order_by(ToolCallMetricRow.created_at)
.limit(1)
)
if existing is None:
session.add(_row_from_metric(metric))
else:
_merge_skill_row(existing, metric)
await session.commit()
except Exception:
logger.warning(
"Failed to merge tool_call_metrics for skill %s (run %s)",
metric.skill_name,
metric.run_id,
exc_info=True,
)
def _apply_filters(self, stmt, query: ToolMetricsQuery):
if query.user_id:
stmt = stmt.where(ToolCallMetricRow.user_id == query.user_id)