Merge branch 'ji'
This commit is contained in:
commit
cf8185498a
8 files changed
+211
-73
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.
|
// otherwise wait until usage arrives so the bar doesn't render on its own.
|
||||||
const usage = getUsageMetadata(message);
|
const usage = getUsageMetadata(message);
|
||||||
const showUsage = enabled && !isLoading && Boolean(usage);
|
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) {
|
if (!showUsage && !showSpeed) {
|
||||||
return null;
|
return null;
|
||||||
|
|||||||
@@ -36,6 +36,13 @@ export interface MessageMetric {
|
|||||||
* only to add a "~" prefix when the value is approximate.
|
* only to add a "~" prefix when the value is approximate.
|
||||||
*/
|
*/
|
||||||
fromUsage: boolean;
|
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 };
|
type Sample = { at: number; chars: number };
|
||||||
@@ -161,6 +168,25 @@ export function useMessageMetrics(
|
|||||||
return () => window.clearInterval(interval);
|
return () => window.clearInterval(interval);
|
||||||
}, [isLoading]);
|
}, [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
|
// Build the snapshot map. We compute on every render so consumers see the
|
||||||
// freshest duration; the underlying timing storage is the source of truth.
|
// freshest duration; the underlying timing storage is the source of truth.
|
||||||
const out = new Map<string, MessageMetric>();
|
const out = new Map<string, MessageMetric>();
|
||||||
@@ -212,6 +238,7 @@ export function useMessageMetrics(
|
|||||||
durationMs,
|
durationMs,
|
||||||
isActive,
|
isActive,
|
||||||
fromUsage,
|
fromUsage,
|
||||||
|
isTurnFinal: turnFinalIds.has(id),
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -71,67 +71,38 @@ function MessageRow({ message }: { message: AdminMessage }) {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Render one memory bucket: context summaries + facts. */
|
/** Render one Memory V2 bucket: a flat list of USER.md / MEMORY.md entries. */
|
||||||
function MemoryBucketView({ bucket }: { bucket: AdminMemoryBucket }) {
|
function MemoryBucketView({ bucket }: { bucket: AdminMemoryBucket }) {
|
||||||
const data = bucket.data as {
|
const raw = (bucket.data as { entries?: unknown }).entries;
|
||||||
user?: Record<string, { summary?: string }>;
|
const entries = Array.isArray(raw)
|
||||||
history?: Record<string, { summary?: string }>;
|
? raw.filter(
|
||||||
facts?: Array<{ content?: string; category?: string; confidence?: number }>;
|
(e): e is string => typeof e === "string" && e.trim().length > 0,
|
||||||
};
|
)
|
||||||
const sections: Array<[string, string]> = [];
|
: [];
|
||||||
const labels: Record<string, string> = {
|
|
||||||
workContext: "工作背景",
|
|
||||||
personalContext: "个人背景",
|
|
||||||
topOfMind: "近期关注",
|
|
||||||
recentMonths: "近几个月",
|
|
||||||
earlierContext: "更早记录",
|
|
||||||
longTermBackground: "长期背景",
|
|
||||||
};
|
|
||||||
for (const group of [data.user, data.history]) {
|
|
||||||
for (const [key, value] of Object.entries(group ?? {})) {
|
|
||||||
const summary = value?.summary?.trim();
|
|
||||||
if (summary) sections.push([labels[key] ?? key, summary]);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
const facts = (data.facts ?? []).filter((f) => f.content?.trim());
|
|
||||||
|
|
||||||
return (
|
return (
|
||||||
<div className="rounded-md border p-4">
|
<div className="rounded-md border p-4">
|
||||||
<div className="mb-3 flex items-center gap-2">
|
<div className="mb-3 flex items-center gap-2">
|
||||||
<Badge variant={bucket.name === "global" ? "default" : "secondary"}>
|
<Badge variant={bucket.name === "global" ? "default" : "secondary"}>
|
||||||
{bucket.name === "global" ? "全局记忆" : `Agent: ${bucket.name}`}
|
{bucket.name === "global" ? "用户画像" : `Agent: ${bucket.name}`}
|
||||||
</Badge>
|
</Badge>
|
||||||
|
{entries.length > 0 && (
|
||||||
|
<span className="text-muted-foreground text-xs">
|
||||||
|
{entries.length} 条
|
||||||
|
</span>
|
||||||
|
)}
|
||||||
</div>
|
</div>
|
||||||
{sections.length === 0 && facts.length === 0 && (
|
{entries.length === 0 ? (
|
||||||
<p className="text-muted-foreground text-sm">该记忆库暂无内容。</p>
|
<p className="text-muted-foreground text-sm">该记忆库暂无内容。</p>
|
||||||
)}
|
) : (
|
||||||
{sections.length > 0 && (
|
<ul className="flex flex-col gap-1.5">
|
||||||
<div className="flex flex-col gap-2">
|
{entries.map((entry, i) => (
|
||||||
{sections.map(([label, summary]) => (
|
<li key={i} className="flex items-start gap-2 text-sm">
|
||||||
<div key={label}>
|
<span className="text-muted-foreground">·</span>
|
||||||
<span className="text-muted-foreground text-xs">{label}</span>
|
<span className="flex-1 whitespace-pre-wrap">{entry}</span>
|
||||||
<p className="text-sm">{summary}</p>
|
</li>
|
||||||
</div>
|
|
||||||
))}
|
))}
|
||||||
</div>
|
</ul>
|
||||||
)}
|
|
||||||
{facts.length > 0 && (
|
|
||||||
<div className="mt-3">
|
|
||||||
<span className="text-muted-foreground text-xs">事实记录({facts.length})</span>
|
|
||||||
<ul className="mt-1 flex flex-col gap-1">
|
|
||||||
{facts.map((fact, i) => (
|
|
||||||
<li key={i} className="flex items-start gap-2 text-sm">
|
|
||||||
<span className="text-muted-foreground">·</span>
|
|
||||||
<span className="flex-1">{fact.content}</span>
|
|
||||||
{fact.category && (
|
|
||||||
<Badge variant="outline" className="shrink-0 text-xs">
|
|
||||||
{fact.category}
|
|
||||||
</Badge>
|
|
||||||
)}
|
|
||||||
</li>
|
|
||||||
))}
|
|
||||||
</ul>
|
|
||||||
</div>
|
|
||||||
)}
|
)}
|
||||||
</div>
|
</div>
|
||||||
);
|
);
|
||||||
|
|||||||
@@ -32,6 +32,7 @@ export default defineConfig({
|
|||||||
},
|
},
|
||||||
],
|
],
|
||||||
},
|
},
|
||||||
|
"base":"./",
|
||||||
server: {
|
server: {
|
||||||
port: 5173,
|
port: 5173,
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -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`)
|
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
|
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)
|
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
|
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)
|
10. **SummarizationMiddleware** - Context reduction when approaching token limits (optional, if enabled)
|
||||||
11. **TodoListMiddleware** - Task tracking with `write_todos` tool (optional, if plan_mode)
|
11. **TodoListMiddleware** - Task tracking with `write_todos` tool (optional, if plan_mode)
|
||||||
|
|||||||
@@ -23,7 +23,7 @@ from app.gateway.deps import (
|
|||||||
get_current_user_from_request,
|
get_current_user_from_request,
|
||||||
get_thread_store,
|
get_thread_store,
|
||||||
)
|
)
|
||||||
from deerflow.agents.memory.updater import get_memory_data
|
from deerflow.agents.memory import BuiltinFileProvider
|
||||||
from deerflow.config.agents_config import AGENT_NAME_PATTERN
|
from deerflow.config.agents_config import AGENT_NAME_PATTERN
|
||||||
from deerflow.config.paths import get_paths
|
from deerflow.config.paths import get_paths
|
||||||
from deerflow.runtime import serialize_channel_values
|
from deerflow.runtime import serialize_channel_values
|
||||||
@@ -136,24 +136,40 @@ class AdminMemoryResponse(BaseModel):
|
|||||||
buckets: list[AdminMemoryBucket] = Field(default_factory=list)
|
buckets: list[AdminMemoryBucket] = Field(default_factory=list)
|
||||||
|
|
||||||
|
|
||||||
|
def _read_memory_bucket(user_id: str, agent_id: str | None) -> dict:
|
||||||
|
"""Read one Memory V2 file into a JSON-able ``{"entries": [...]}`` dict.
|
||||||
|
|
||||||
|
``agent_id is None`` -> the user's ``USER.md`` (profile, cross-agent);
|
||||||
|
otherwise that agent's ``MEMORY.md`` (per-agent working notes).
|
||||||
|
"""
|
||||||
|
provider = BuiltinFileProvider()
|
||||||
|
provider.initialize(user_id=user_id, agent_id=agent_id)
|
||||||
|
target = "user" if agent_id is None else "memory"
|
||||||
|
return {"entries": provider.read(target)}
|
||||||
|
|
||||||
|
|
||||||
@router.get("/{user_id}/memory", response_model=AdminMemoryResponse)
|
@router.get("/{user_id}/memory", response_model=AdminMemoryResponse)
|
||||||
async def get_user_memory(user_id: str, request: Request) -> AdminMemoryResponse:
|
async def get_user_memory(user_id: str, request: Request) -> AdminMemoryResponse:
|
||||||
"""Return every memory bucket (global + per-agent) for one user."""
|
"""Return every memory bucket for one user.
|
||||||
|
|
||||||
|
Memory V2 layout: the ``global`` bucket is ``USER.md`` (user profile);
|
||||||
|
each per-agent bucket is that agent's ``MEMORY.md`` working notes.
|
||||||
|
"""
|
||||||
await _require_admin(request)
|
await _require_admin(request)
|
||||||
|
|
||||||
buckets: list[AdminMemoryBucket] = [
|
buckets: list[AdminMemoryBucket] = [
|
||||||
AdminMemoryBucket(name="global", data=get_memory_data(None, user_id=user_id))
|
AdminMemoryBucket(name="global", data=_read_memory_bucket(user_id, None))
|
||||||
]
|
]
|
||||||
|
|
||||||
agents_root = get_paths().user_dir(user_id) / "agents"
|
agents_root = get_paths().user_dir(user_id) / "agents"
|
||||||
if agents_root.is_dir():
|
if agents_root.is_dir():
|
||||||
for entry in sorted(agents_root.iterdir(), key=lambda p: p.name.lower()):
|
for entry in sorted(agents_root.iterdir(), key=lambda p: p.name.lower()):
|
||||||
if not entry.is_dir() or not (entry / "memory.json").exists():
|
if not entry.is_dir() or not (entry / "MEMORY.md").exists():
|
||||||
continue
|
continue
|
||||||
if not AGENT_NAME_PATTERN.match(entry.name):
|
if not AGENT_NAME_PATTERN.match(entry.name):
|
||||||
continue
|
continue
|
||||||
try:
|
try:
|
||||||
data = get_memory_data(entry.name, user_id=user_id)
|
data = _read_memory_bucket(user_id, entry.name)
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.warning("Failed to read memory bucket %s for user", entry.name)
|
logger.warning("Failed to read memory bucket %s for user", entry.name)
|
||||||
continue
|
continue
|
||||||
|
|||||||
+8
@@ -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 /
|
coarse ``error_category`` (rate_limited / ip_blocked / auth_error / timeout /
|
||||||
network_error / not_found / empty_result / tool_error) so administrators can
|
network_error / not_found / empty_result / tool_error) so administrators can
|
||||||
spot retrieval outages, blocked IPs, and quota problems at a glance.
|
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
|
from deerflow.persistence.tool_metrics.base import ToolMetricRecord, ToolMetricsStore
|
||||||
|
|||||||
+124
-15
@@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import asyncio
|
||||||
import logging
|
import logging
|
||||||
from collections.abc import AsyncIterator
|
from collections.abc import AsyncIterator
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
@@ -15,11 +16,99 @@ from deerflow.persistence.tool_metrics.model import ToolCallMetricRow
|
|||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
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:
|
def _normalize_created_at(value: datetime) -> datetime:
|
||||||
return value if value.tzinfo else value.replace(tzinfo=UTC)
|
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):
|
class SqlToolMetricsStore(ToolMetricsStore):
|
||||||
"""SQLite/Postgres/MySQL-backed implementation; writes are fire-and-forget."""
|
"""SQLite/Postgres/MySQL-backed implementation; writes are fire-and-forget."""
|
||||||
|
|
||||||
@@ -27,28 +116,48 @@ class SqlToolMetricsStore(ToolMetricsStore):
|
|||||||
self._sf = session_factory
|
self._sf = session_factory
|
||||||
|
|
||||||
async def record(self, metric: ToolMetricRecord) -> None:
|
async def record(self, metric: ToolMetricRecord) -> None:
|
||||||
row = ToolCallMetricRow(
|
# Skill calls are merged per run: the model may touch one skill across
|
||||||
id=metric.id,
|
# several tool calls / retries, and the admin wants a single row per
|
||||||
created_at=_normalize_created_at(metric.created_at),
|
# (run, skill) instead of one per call. Non-skill tool calls, and skill
|
||||||
user_id=metric.user_id,
|
# calls with no run_id, fall back to a plain insert.
|
||||||
thread_id=metric.thread_id,
|
if metric.skill_name and metric.run_id:
|
||||||
run_id=metric.run_id,
|
await self._record_skill_merged(metric)
|
||||||
agent_name=metric.agent_name,
|
return
|
||||||
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,
|
|
||||||
)
|
|
||||||
try:
|
try:
|
||||||
async with self._sf() as session:
|
async with self._sf() as session:
|
||||||
session.add(row)
|
session.add(_row_from_metric(metric))
|
||||||
await session.commit()
|
await session.commit()
|
||||||
except Exception:
|
except Exception:
|
||||||
# Metrics are best-effort; never surface a write failure to the runtime.
|
# 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)
|
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):
|
def _apply_filters(self, stmt, query: ToolMetricsQuery):
|
||||||
if query.user_id:
|
if query.user_id:
|
||||||
stmt = stmt.where(ToolCallMetricRow.user_id == query.user_id)
|
stmt = stmt.where(ToolCallMetricRow.user_id == query.user_id)
|
||||||
|
|||||||
Reference in new issue
Block a user