AI智能体项目实战记录（30）：token 消耗统计，并在每轮对话结束后展示给用户
1. 引言
第 15 篇第一次把调用日志落到了 ai_invoke_log / ai_tool_invoke_log，思路是对的：
一次用户提问 = 一个 TASK_BATCH
├─ 一条“主管智能体”汇总行
├─ 每个子智能体（每次工具执行）一条明细行
└─ 每条明细行再关联一条工具调用记录
但那一版留下了一批遗漏和 bug，最典型的是：
| # | 问题 | 现象 |
|---|---|---|
| 1 | 子智能体行 COST_TOKEN_ 恒为 0 | 只能看到总数，拆不出“钱花在哪个工具” |
| 2 | CHILD_AGENT_ID_ 全为空 | 汇总行关联不到执行单元，按智能体聚合全是空 |
| 3 | 并行工具丢记账 | ThreadPoolExecutor.submit 不传播 contextvar，工具线程里 current() 是 None |
| 4 | 口径不全 | 知识库、企业信息、经营分析、网站工具内部的模型调用没计入 |
| 5 | 用量拿不出来 | finalize_run 落完库就把对象丢了，前端拿不到 |
| 6 | 重复调用放大用量 | 主管反复检索同一问题，token 翻 2~3 倍 |
这一篇把这些问题一起修掉，并补上“每轮对话结束展示给用户”的完整链路。
2. 整体流程
POST /chat/v1/chat/completions
│
▼
_handle_stream（controller）
│ begin_run(user_id, thread_id, skill, question)
▼
supervisor_graph.astream()
│
├─ 主管决策 ──── record_model_response() ─┐
│ │
├─ 工具调用（可并行） │ 都写进 _RunLog（父级累加器）
│ begin_tool_scope(name) │ 同时写进当前工具作用域
│ ├─ 知识库 RAG record_model_response()
│ ├─ 企业信息 record_model_response()
│ └─ … │
│ end_tool_scope() → 该工具 tokens ────┤
│ │
└─ 总结生成 ──── record_usage() ─────────┘
│
▼
finalize_run() ── 落库（汇总行 + 子行 + 工具明细）── 返回 summary
│ └─ token_quota_service.add_tokens()
▼
yield {"type": "usage", "content": {..., "usedToday", "limit"}}
│
▼
controller 转 StreamDataResp(type="usage") → 浏览器
│
├─ 前端：气泡下方小字
└─ 兜底：正文尾部追加一行（兼容旧前端）
三条硬约束：
- 先落库、后推送——展示的数字必须等于记账的数字；
- 父级记全量、作用域记明细——双写，两边都不重不漏；
- 任何记账失败都不影响回答——全部
try/except包裹。
3. 数据模型
3.1 三类行
| 行 | 数量 | COST_TOKEN_ | TOOL_INVOKE_COUNT_ | CHILD_AGENT_ID_ |
|---|---|---|---|---|
| 汇总行 | 每轮 1 条 | 本轮总量（含所有子智能体） | 本轮工具数 | agent_supervisor |
| 子智能体行 | 每个工具 1 条 | 该工具自己的消耗 | 1 | 对应智能体 ID |
| 工具明细 | 每个工具 1 条（ai_tool_invoke_log） | — | — | 通过 AI_INVOKE_LOG_ID_ 关联子行 |
汇总行存全量是有意为之：这样
SUM(COST_TOKEN_)按用户/按天统计才是对的；想拆明细就查子行。
3.2 实测数据形态
thread=xxx child_agent_id=agent_supervisor cost_token=8865 tool_invoke_count=1 ← 汇总
thread=xxx child_agent_id=2 cost_token=2830 tool_invoke_count=1 ← 知识库
8865（总） = 2830（知识库） + 6035（主管决策 + 总结）
4. 关键设计
4.1 contextvar 贯穿异步与线程
图节点、工具、LLM 调用分散在 async 与 ThreadPoolExecutor 里，手工传对象不现实，所以用两个 contextvar：
_run_log_var: ContextVar[Optional[_RunLog]] # 整个请求
_tool_scope_var: ContextVar[Optional[_ToolScope]] # 当前工具
4.2 双写：父级 + 工具作用域
def record_usage(prompt_tokens=None, completion_tokens=None):
run_log = current()
if run_log is None:
return
run_log.add_usage(prompt_tokens, completion_tokens) # 父级（总量）
scope = _tool_scope_var.get()
if scope is not None:
scope.add(prompt_tokens, completion_tokens) # 当前工具（明细）
一次记账，两个去处。父级保证总量正确，作用域保证能拆到工具。
4.3 并行工具不丢账
# supervisor_graph._run_async_in_thread_pool
ctx = contextvars.copy_context()
return executor.submit(ctx.run, lambda: asyncio.run(coro)).result()
copy_context() 把当前线程的 contextvar 快照带进子任务，否则工具线程里 current() 是 None，记账静默丢失——这是第 15 篇最隐蔽的 bug。
4.4 CHILD_AGENT_ID_ 为空的真正原因
row = query_one("SELECT ID_ FROM ai_child_agent WHERE NAME = :name ...")
agent_id = (row or {}).get("ID_") # ← 永远取不到
PostgreSQL 会把未加引号的标识符折成小写，返回的字典键是 id_，不是 ID_。MySQL 下没问题，切到 PG 后全空。
修法：用别名，不依赖驱动的大小写行为。
SELECT ID_ AS agent_id FROM ai_child_agent WHERE NAME = :name AND DELETED_ != 1 LIMIT 1
同时补了两件事：
ai_child_agent里没有“主管智能体”这行时自动补建（固定 IDagent_supervisor，幂等）；- 未命中的结果不缓存，补完数据不用重启。
4.5 用量必须能带出来
finalize_run 原本 return None，现在返回：
{
"promptTokens": 7756, "completionTokens": 1109, "totalTokens": 8865,
"costMs": 12345, "toolCount": 1
}
上层再补 usedToday / limit 推给前端。
5. 关键代码
5.1 运行上下文与工具作用域
class _RunLog:
def __init__(self, user_id, thread_id, skill, question):
self.run_id = uuid.uuid4().hex
self.batch_id = uuid.uuid4().hex
self.started_dt = datetime.now()
self.tools = []
self.prompt_tokens = 0
self.completion_tokens = 0
self.lock = threading.Lock()
def add_usage(self, prompt=None, completion=None):
with self.lock:
if prompt:
self.prompt_tokens += int(prompt)
if completion:
self.completion_tokens += int(completion)
@property
def total_tokens(self):
return self.prompt_tokens + self.completion_tokens
class _ToolScope:
"""单次工具执行的 token 计数器"""
def __init__(self, tool_name):
self.tool_name = tool_name
self.prompt_tokens = 0
self.completion_tokens = 0
self.lock = threading.Lock()
begin_tool_scope / end_tool_scope 用工整的 set/reset，必须在 finally 中配对：
def begin_tool_scope(tool_name: str):
return _tool_scope_var.set(_ToolScope(tool_name))
def end_tool_scope(scope_token) -> int:
scope = _tool_scope_var.get()
_tool_scope_var.reset(scope_token)
return scope.total_tokens if scope else 0
5.2 工具节点：作用域包住整段执行
start_dt = datetime.now()
scope_token = invoke_log_service.begin_tool_scope(tool_name)
try:
result = tool._run(...)
clean_result = strip_tool_error(result)
ok = not is_tool_error(result)
except Exception as e:
clean_result = f"工具执行失败: {str(e)}"
ok = False
error_text = str(e)
finally:
tool_tokens = invoke_log_service.end_tool_scope(scope_token)
_record_tool_event(
tool_name, tool_args, start_dt,
clean_result if ok else None,
error_text,
tool_tokens, # ← 子行 COST_TOKEN_ 用这个
)
不管成功、失败还是抛异常，作用域都会出栈，不会污染下一个工具。
5.3 记账入口：三处必调
def record_model_response(resp, prompt_messages, answer_text) -> None:
"""优先取 usage，缺失时按字符估算"""
run_log = current()
if run_log is None:
return
usage = getattr(resp, "usage", None)
if usage is not None:
record_usage(
getattr(usage, "prompt_tokens", None),
getattr(usage, "completion_tokens", None),
)
else:
record_text_estimate(prompt_messages, answer_text)
调用点：
| 位置 | 调用 |
|---|---|
_node_supervisor 每轮决策 | record_model_response(resp, messages_for_selection, raw_output) |
_astream_llm（总结） | 流结束后 record_usage(usage_prompt, usage_completion) |
| 工具内部（RAG / 企业信息 / 经营分析 / 网站） | 各自 record_model_response(...) |
5.4 汇总行与子行落库
# 1) 汇总行
conn.execute(text(text_insert_invoke()), {
"id": run_log.run_id,
"child_agent_id": supervisor_agent_id or "",
"tool_invoke_count": len(run_log.tools),
"cost_token": run_log.total_tokens, # ← 本轮总量
...
})
# 2) 每个子智能体一行
for event in run_log.tools:
child_invoke_id = uuid.uuid4().hex
conn.execute(text(text_insert_invoke()), {
"id": child_invoke_id,
"child_agent_id": _lookup_child_agent_id([event.get("display_name"), event.get("tool_name")]),
"tool_invoke_count": 1,
"cost_token": int(event.get("tokens") or 0), # ← 该工具自己的用量
...
})
# 3) 工具明细
conn.execute(text(text_insert_tool()), {...})
5.5 收尾并返回用量
async def finalize_run(success, answer="", fail_msg=""):
run_log = current()
if run_log is None:
return None
summary = {
"promptTokens": run_log.prompt_tokens,
"completionTokens": run_log.completion_tokens,
"totalTokens": run_log.total_tokens,
"costMs": int((datetime.now() - run_log.started_dt).total_seconds() * 1000),
"toolCount": len(run_log.tools),
}
try:
await asyncio.to_thread(_do_finalize, run_log, success, answer, fail_msg)
finally:
reset_run()
return summary
5.6 图里推送 usage（落库之后）
summary = await invoke_log_service.finalize_run(
success=success, answer=answer_text, fail_msg=fail_msg
)
if summary:
limit = int(setting.chat_daily_token_limit or 0)
used_today = 0
if limit > 0:
try:
used_today = await asyncio.to_thread(
token_quota_service.used_tokens, user_id, "chat"
)
except Exception as e:
logger.warning(f"[usage] 读取今日用量失败: {e}")
yield {
"type": "usage",
"content": {**summary, "usedToday": used_today, "limit": limit},
}
usedToday 从 Redis 读（finalize_run 内部已 add_tokens），所以推送值和当日额度是同一份数据。
5.7 控制器转发 + 旧前端兜底
elif chunk_type == "usage":
usage_info = content if isinstance(content, dict) else None
yield StreamDataResp(data=content, type=StreamDataRespType.USAGE.value).model_dump_json()
...
await _record_conversation(messages, thread_id, user_id, full_answer)
if setting.chat_usage_in_answer and usage_info:
tail = _format_usage_line(usage_info)
if tail:
yield StreamDataResp(data=tail, type=StreamDataRespType.ANSWER.value).model_dump_json()
def _format_usage_line(usage: dict) -> str:
total = int(usage.get("totalTokens") or 0)
if total <= 0:
return ""
parts = [f"本次消耗 {total:,} tokens"]
cost_ms = int(usage.get("costMs") or 0)
if cost_ms > 0:
parts.append(f"用时 {cost_ms / 1000:.1f}s")
limit = int(usage.get("limit") or 0)
used = int(usage.get("usedToday") or 0)
if limit > 0:
parts.append(f"今日剩余 {max(0, limit - used):,}")
return "\n\n---\n" + " · ".join(parts)
两个细节：
- 尾部文案在
_record_conversation之后发，所以不会写进历史； CHAT_USAGE_IN_ANSWER=false可关掉，只用结构化事件。
5.8 前端解析与展示
chat.js 增加分支：
if (type === 'answer') {
onAnswer?.(data)
} else if (type === 'analysis') {
onStatus?.(data)
} else if (type === 'usage') {
onUsage?.(data) // 新增：本次用量
} else if (type === 'complete') {
onDone?.()
}
Chat.vue 接收并挂到消息上：
onUsage: (usage) => {
if (messages.value[assistantIndex]) {
messages.value[assistantIndex].usage = usage || null
}
},
function formatUsage(u) {
if (!u || !u.totalTokens) return ''
const parts = [`本次消耗 ${Number(u.totalTokens).toLocaleString()} tokens`]
if (u.costMs) parts.push(`用时 ${(u.costMs / 1000).toFixed(1)}s`)
if (u.limit > 0 && typeof u.usedToday === 'number') {
const left = Math.max(0, u.limit - u.usedToday)
parts.push(`今日剩余 ${left.toLocaleString()}`)
}
return parts.join(' · ')
}
<div v-if="msg.usage" class="usage-hint">{{ formatUsage(msg.usage) }}</div>
未知 type 会被解析器忽略，所以新增事件天然向后兼容。
6. 避免重复调用放大用量
同一问题被连调 4 次时，用量会翻 2~3 倍。两层防护：
① 提示词规则（软约束）
11. 【避免重复调用】如果“已获取的工具结果”中已有能回答用户问题的内容，不要再次调用相同工具。
② 硬去重（确定性）
def _call_signature(tool_name: str, tool_args: dict) -> str:
args_text = json.dumps(tool_args or {}, sort_keys=True, ensure_ascii=False, default=str)
return f"{tool_name}:{args_text}"
executed = set(state.get("executed_calls") or [])
deduped = []
for tc in tool_calls or []:
sig = _call_signature(tc.get("name") or "", tc.get("args") or {})
if sig in executed:
logger.warning(f"[supervisor] 跳过重复调用: {sig[:120]}")
continue
deduped.append(tc)
tool_calls = deduped
只拦“工具名 + 参数完全相同”，换库补充检索（参数不同）仍然允许。
7. 实测
# 事件顺序
data: {"data":"...","type":"answer"}
data: {"data":{"promptTokens":7756,"completionTokens":1109,"totalTokens":8865,
"costMs":12345,"toolCount":1,"usedToday":8865,"limit":200000},"type":"usage"}
data: {"data":"\n\n---\n本次消耗 8,865 tokens · 用时 12.3s · 今日剩余 191,135","type":"answer"}
data: {"data":"","type":"complete"}
# 去重前后（同一问题）
改前: 4 次工具调用, 决策 5 轮, 21467 tokens, 21.3s
改后: 1 次工具调用, 决策 2 轮, 8865 tokens, 8.2s
# 子行终于有值
{'child_agent_id_': 'agent_supervisor', 'cost_token': 8865, 'tool_invoke_count': 1}
{'child_agent_id_': '2', 'cost_token': 2830, 'tool_invoke_count': 1}
8. 对账口径
-- 某用户某天总消耗（汇总行已含子行，直接 SUM 不会重复）
SELECT SUM(COST_TOKEN_) FROM ai_invoke_log
WHERE CREATE_BY_ = '1001' AND CREATE_TIME_ >= '2026-10-29';
-- 某轮按工具拆分
SELECT CHILD_AGENT_ID_, SUM(COST_TOKEN_), COUNT(*)
FROM ai_invoke_log
WHERE THREAD_ID = 'xxx' AND CHILD_AGENT_ID_ <> 'agent_supervisor'
GROUP BY CHILD_AGENT_ID_;
-- 主管自身 = 汇总行 - 子行合计
和 DeepSeek 控制台对比时注意口径：控制台按“单次 API 请求”计，我们按“一轮用户提问”计，一轮通常包含 3~5 次请求，所以我们的数字是控制台单条的几倍，属正常。
9. 遗留
| # | 问题 | 影响 |
|---|---|---|
| 1 | usage 缺失时按“2 字符 ≈ 1 token”估算 | 中文略偏低 |
| 2 | Hermes 长任务不回传 usage | 该部分未计入 |
| 3 | 历史消息不保存 usage | 重新进入会话看不到用量（只在当轮显示） |
| 4 | 非流式接口（stream: false）未返回 usage | 需要单独接口或改契约 |
| 5 | 报表工具失败时 success_result 为空 | 排查时只能看 FAIL_MSG_ |
10. 附：invoke_log_service.py 完整代码
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @File : invoke_log_service.py
# @Desc : 智能体调用日志落库服务（MySQL）
# - 一次用户请求 = 一个 TASK_BATCH，ai_invoke_log 写一条“主管智能体”汇总行
# - 每次工具执行（=一次子智能体调用）各写一条 ai_invoke_log，执行单元为对应子智能体
# - 每条子智能体行再对应写一条 ai_tool_invoke_log 工具调用明细
# - token 统计：优先取各模型响应 usage，缺失时按字符数估算（约 2 字符 ≈ 1 token）
#
# 通过 contextvar 把“本次运行的日志上下文”透传给编排图各节点与工具线程，
# 结束后统一落库，避免在多线程/异步间手工传递对象。
import asyncio
import contextvars
import threading
import uuid
from datetime import datetime
from typing import Any, Dict, List, Optional
from loguru import logger
from sqlalchemy import text
from app.service.system import db_service, token_quota_service
# 表名（与 Java 后台 ai_invoke_log / ai_tool_invoke_log 对应，无 dzxx_ 前缀）
TABLE_INVOKE_LOG = "ai_invoke_log"
TABLE_TOOL_INVOKE_LOG = "ai_tool_invoke_log"
TABLE_CHILD_AGENT = "ai_child_agent"
# “主管智能体”在 ai_child_agent 表中的名称（用于把汇总行关联到执行单元）
SUPERVISOR_AGENT_NAME = "主管智能体"
# 缺失时自动补一条（固定 ID，便于识别是系统生成的）
SUPERVISOR_AGENT_ID = "agent_supervisor"
# 2 个字符估算为 1 个 token（中文场景的粗略折算）
_CHARS_PER_TOKEN = 2
class _RunLog:
"""一次智能体调用的运行日志上下文"""
def __init__(self, user_id: str, thread_id: str, skill: str, question: str):
self.run_id = uuid.uuid4().hex # ai_invoke_log.ID_
self.batch_id = uuid.uuid4().hex # TASK_BATCH_ID
self.user_id = user_id or ""
self.thread_id = thread_id or ""
self.skill = skill or ""
self.question = (question or "")[:1000] # USER_INPUT varchar(1000)
self.started_dt = datetime.now()
self.tools: List[Dict[str, Any]] = [] # 工具明细
self.prompt_tokens = 0
self.completion_tokens = 0
self.lock = threading.Lock()
self.error_message = ""
def add_tool(self, event: Dict[str, Any]) -> None:
with self.lock:
self.tools.append(event)
def add_usage(self, prompt: Optional[int] = None, completion: Optional[int] = None) -> None:
with self.lock:
if prompt:
self.prompt_tokens += int(prompt)
if completion:
self.completion_tokens += int(completion)
@property
def total_tokens(self) -> int:
return self.prompt_tokens + self.completion_tokens
_run_log_var: contextvars.ContextVar[Optional[_RunLog]] = contextvars.ContextVar("invoke_run_log", default=None)
class _ToolScope:
"""单次工具执行的 token 计数器（用于把用量拆分到子智能体行）"""
def __init__(self, tool_name: str):
self.tool_name = tool_name
self.prompt_tokens = 0
self.completion_tokens = 0
self.lock = threading.Lock()
def add(self, prompt: Optional[int] = None, completion: Optional[int] = None) -> None:
with self.lock:
if prompt:
self.prompt_tokens += int(prompt)
if completion:
self.completion_tokens += int(completion)
@property
def total_tokens(self) -> int:
return self.prompt_tokens + self.completion_tokens
# 当前线程/任务正在执行的工具作用域；并行工具各自在独立线程上下文里，互不干扰
_tool_scope_var: contextvars.ContextVar[Optional[_ToolScope]] = contextvars.ContextVar("invoke_tool_scope", default=None)
_supervisor_id_cache: Optional[str] = None
_supervisor_id_lock = threading.Lock()
# 子智能体（按工具展示名/工具名）ID 解析缓存：name -> ID_（'' 表示未匹配）
_child_agent_id_cache: Dict[str, str] = {}
_child_agent_id_lock = threading.Lock()
# =============================================================================
# 运行上下文管理
# =============================================================================
def begin_run(user_id: str = "", thread_id: str = "", skill: str = "", question: str = "") -> _RunLog:
"""开启一次智能体调用日志上下文（每个请求一次）"""
run_log = _RunLog(user_id, thread_id, skill, question)
_run_log_var.set(run_log)
return run_log
def current() -> Optional[_RunLog]:
"""获取当前请求的日志上下文（图节点/工具线程内调用）"""
return _run_log_var.get()
def reset_run() -> None:
"""请求结束后清理上下文"""
_run_log_var.set(None)
def begin_tool_scope(tool_name: str):
"""进入一次工具执行的作用域（返回用于 end_tool_scope 的 token）"""
return _tool_scope_var.set(_ToolScope(tool_name))
def end_tool_scope(scope_token) -> int:
"""
退出工具作用域并返回该工具消耗的 token。
必须在 finally 中调用，保证异常路径也能出栈。
"""
scope = _tool_scope_var.get()
_tool_scope_var.reset(scope_token)
return scope.total_tokens if scope else 0
# =============================================================================
# token 统计辅助（供各 LLM 调用点使用）
# =============================================================================
def _estimate_tokens(text: Any) -> int:
text = "" if text is None else str(text)
return max(1, (len(text) + _CHARS_PER_TOKEN - 1) // _CHARS_PER_TOKEN)
def record_usage(prompt_tokens: Optional[int] = None, completion_tokens: Optional[int] = None) -> None:
"""把模型返回的 usage 计入当前运行（usage 字段缺失时调用方传 None，则不会累加）"""
run_log = current()
if run_log is None:
return
run_log.add_usage(prompt_tokens, completion_tokens)
# 同步计入当前工具作用域，便于按子智能体拆分统计
scope = _tool_scope_var.get()
if scope is not None:
scope.add(prompt_tokens, completion_tokens)
def record_text_estimate(prompt_messages: List[Dict[str, Any]], answer_text: Any) -> None:
"""模型未返回 usage 时按字符估算 prompt/completion token 并计入"""
run_log = current()
if run_log is None:
return
prompt_text = ""
for msg in prompt_messages or []:
content = msg.get("content")
if isinstance(content, str):
prompt_text += content
prompt = _estimate_tokens(prompt_text)
completion = _estimate_tokens(answer_text)
record_usage(prompt, completion)
def record_model_response(resp: Any, prompt_messages: List[Dict[str, Any]], answer_text: Any) -> None:
"""
统一的模型响应 token 记账入口：
优先使用 resp.usage（OpenAI 兼容响应对象），缺失时按字符估算。
"""
run_log = current()
if run_log is None:
return
usage = getattr(resp, "usage", None)
if usage is not None:
record_usage(
getattr(usage, "prompt_tokens", None),
getattr(usage, "completion_tokens", None),
)
else:
record_text_estimate(prompt_messages, answer_text)
# =============================================================================
# 落库
# =============================================================================
def _query_supervisor_agent_id() -> Optional[str]:
"""查询“主管智能体”在 ai_child_agent 中的 ID（进程内缓存；缺失时自动补一条）"""
global _supervisor_id_cache
with _supervisor_id_lock:
if _supervisor_id_cache is not None:
return _supervisor_id_cache
if not db_service.is_configured():
return None
try:
# 注意：PostgreSQL 会把未加引号的列名转为小写，必须用别名读取，
# 否则 row.get("ID_") 永远取不到值（历史 bug）
row = db_service.query_one(
f"SELECT ID_ AS agent_id FROM {TABLE_CHILD_AGENT} "
"WHERE NAME = :name AND DELETED_ != 1 LIMIT 1",
{"name": SUPERVISOR_AGENT_NAME},
)
agent_id = (row or {}).get("agent_id") or ""
if not agent_id:
agent_id = _create_supervisor_agent() or ""
_supervisor_id_cache = agent_id or None
except Exception as e:
logger.warning(f"[invoke_log] 查询主管智能体ID失败: {e}")
_supervisor_id_cache = None
return _supervisor_id_cache
def _create_supervisor_agent() -> Optional[str]:
"""ai_child_agent 缺少“主管智能体”时自动补一条（汇总行需要关联到它）"""
try:
now = datetime.now()
with db_service.get_engine().begin() as conn:
conn.execute(
text(
f"INSERT INTO {TABLE_CHILD_AGENT} "
"(ID_, NAME, CATEGORY, STATUS, DUTY, DELETED_, CREATE_TIME_, UPDATE_TIME_) "
"VALUES (:id, :name, 1, 1, '1', 0, :now, :now)"
),
{"id": SUPERVISOR_AGENT_ID, "name": SUPERVISOR_AGENT_NAME, "now": now},
)
logger.info(f"[invoke_log] 已自动创建「{SUPERVISOR_AGENT_NAME}」记录 id={SUPERVISOR_AGENT_ID}")
return SUPERVISOR_AGENT_ID
except Exception as e:
logger.warning(f"[invoke_log] 自动创建主管智能体记录失败（汇总行将不带 agent_id）: {e}")
return None
def _safe_text(value: Any, limit: Optional[int] = None) -> Optional[str]:
if value is None:
return None
text = str(value)
if limit is not None and len(text) > limit:
text = text[:limit]
return text
def _lookup_child_agent_id(names: List[str]) -> str:
"""按名称优先级解析子智能体在 ai_child_agent 中的 ID（展示名优先，其次工具名）"""
if not db_service.is_configured():
return ""
for name in names or []:
if not name:
continue
with _child_agent_id_lock:
if name in _child_agent_id_cache:
return _child_agent_id_cache.get(name, "")
agent_id = ""
try:
# 同样用别名读取，避免 PostgreSQL 小写列名导致取不到值
row = db_service.query_one(
f"SELECT ID_ AS agent_id FROM {TABLE_CHILD_AGENT} "
"WHERE NAME = :name AND DELETED_ != 1 LIMIT 1",
{"name": name},
)
agent_id = (row or {}).get("agent_id") or ""
except Exception as e:
logger.warning(f"[invoke_log] 查询子智能体[{name}]ID失败: {e}")
# 只缓存命中结果：未命中不缓存，补数据后无需重启即可生效
if agent_id:
_child_agent_id_cache[name] = agent_id
else:
logger.warning(f"[invoke_log] ai_child_agent 中未找到名称为「{name}」的智能体")
return agent_id
return ""
def _event_user_input(event: Dict[str, Any]) -> str:
"""从工具参数中提取用于 USER_INPUT 的任务描述（优先 question/query 等关键参数）"""
args = event.get("args")
if isinstance(args, dict):
for key in ("question", "query", "company_name", "city", "date_str", "datetime", "date", "content"):
value = args.get(key)
if value is not None and str(value).strip():
return str(value)[:1000]
return _safe_text(event.get("tool_params"), 500) or ""
def _do_finalize(run_log: _RunLog, success: bool, answer: str, fail_msg: str = "") -> None:
"""把本次运行的汇总行与工具明细行写入 MySQL（在 to_thread 中执行）"""
if not db_service.is_configured():
return
try:
end_dt = datetime.now()
cost_time = round((end_dt - run_log.started_dt).total_seconds(), 6)
supervisor_agent_id = _query_supervisor_agent_id()
with db_service.get_engine().begin() as conn:
# 1) ai_invoke_log 汇总行（主管智能体）
conn.execute(
text(text_insert_invoke()),
{
"id": run_log.run_id,
"thread_id": _safe_text(run_log.thread_id, 64) or "",
"task_batch_id": run_log.batch_id,
"child_agent_id": supervisor_agent_id or "",
"user_input": run_log.question,
"tool_invoke_count": len(run_log.tools),
"status": 1 if success else -1,
"cost_time": cost_time,
"cost_token": run_log.total_tokens,
"fail_msg": _safe_text(fail_msg, 500),
"success_result": _safe_text(answer) if success else None,
"skill_id": _safe_text(run_log.skill, 64),
"tenant_id": "",
"create_dep_id": "",
"create_by": _safe_text(run_log.user_id, 64) or "",
"create_time": run_log.started_dt,
"update_by": _safe_text(run_log.user_id, 64) or "",
"update_time": end_dt,
},
)
# 2) 每个子智能体（每次工具执行）各写一条 ai_invoke_log，并把工具明细关联到该行
for event in run_log.tools:
child_invoke_id = uuid.uuid4().hex
child_agent_id = _lookup_child_agent_id([
event.get("display_name"),
event.get("tool_name"),
])
event_ok = event.get("status", 1) == 1
conn.execute(
text(text_insert_invoke()),
{
"id": child_invoke_id,
"thread_id": _safe_text(run_log.thread_id, 64) or "",
"task_batch_id": run_log.batch_id,
"child_agent_id": child_agent_id,
"user_input": _event_user_input(event),
"tool_invoke_count": 1,
"status": event.get("status", 1),
"cost_time": event.get("cost_time") or 0,
"cost_token": int(event.get("tokens") or 0),
"fail_msg": _safe_text(event.get("fail_msg"), 500),
"success_result": _safe_text(event.get("success_result")) if event_ok else None,
"skill_id": _safe_text(run_log.skill, 64),
"tenant_id": "",
"create_dep_id": "",
"create_by": _safe_text(run_log.user_id, 64) or "",
"create_time": event.get("start_time"),
"update_by": _safe_text(run_log.user_id, 64) or "",
"update_time": event.get("end_time"),
},
)
# 3) 子智能体行对应的工具调用明细
display_name = event.get("display_name") or event.get("tool_name") or ""
conn.execute(
text(text_insert_tool()),
{
"id": uuid.uuid4().hex,
"invoke_log_id": child_invoke_id,
"tool_name": _safe_text(display_name, 255) or "",
"tool_params": _safe_text(event.get("tool_params")),
"start_time": event.get("start_time"),
"end_time": event.get("end_time"),
"cost_time": event.get("cost_time") or 0,
"status": event.get("status", 1),
"success_result": _safe_text(event.get("success_result")) if event_ok else None,
"fail_msg": _safe_text(event.get("fail_msg"), 2000),
"create_by": _safe_text(run_log.user_id, 64) or "",
"create_time": event.get("start_time"),
"update_by": _safe_text(run_log.user_id, 64) or "",
"update_time": event.get("end_time"),
},
)
logger.info(
f"[invoke_log] 已写入调用日志 run_id={run_log.run_id} batch={run_log.batch_id} "
f"tools={len(run_log.tools)} tokens={run_log.total_tokens} status={'成功' if success else '失败'}"
)
# 累加当天普通对话额度（MCP 调用在 MCP Server 侧单独累加）
token_quota_service.add_tokens(run_log.user_id, run_log.total_tokens, "chat")
except Exception as e:
logger.error(f"[invoke_log] 写入调用日志失败: {e}")
def log_external_invoke(
user_id: str,
question: str,
answer: str,
tokens: int,
source: str = "mcp:knowledge-base",
success: bool = True,
fail_msg: str = "",
) -> None:
"""
记录一次由外部客户端（MCP Server / 第三方系统）发起的调用。
与对话链路的区别：没有主管/子智能体编排，只落一行汇总日志；
SKILL_ID_ 用 mcp 前缀标记，便于和普通对话额度区分统计。
"""
if not db_service.is_configured():
return
now = datetime.now()
try:
with db_service.get_engine().begin() as conn:
conn.execute(
text(text_insert_invoke()),
{
"id": uuid.uuid4().hex,
"thread_id": "",
"task_batch_id": uuid.uuid4().hex,
"child_agent_id": "",
"user_input": (question or "")[:1000],
"tool_invoke_count": 1,
"status": 1 if success else -1,
"cost_time": 0,
"cost_token": int(tokens or 0),
"fail_msg": _safe_text(fail_msg, 500),
"success_result": _safe_text(answer) if success else None,
"skill_id": _safe_text(source, 64),
"tenant_id": "",
"create_dep_id": "",
"create_by": _safe_text(user_id, 64) or "mcp",
"create_time": now,
"update_by": _safe_text(user_id, 64) or "mcp",
"update_time": now,
},
)
logger.info(
f"[invoke_log] 已写入外部调用日志 source={source} user={user_id or 'mcp'} tokens={tokens}"
)
# 台账写成功后再累加 MCP 额度，避免日志失败时额度被多扣
token_quota_service.add_tokens(user_id, tokens, "mcp")
except Exception as e:
logger.error(f"[invoke_log] 写入外部调用日志失败: {e}")
async def finalize_run(success: bool, answer: str = "", fail_msg: str = "") -> Optional[Dict[str, Any]]:
"""
异步收尾：在 to_thread 中执行落库并清理上下文。
Returns:
本次运行的用量汇总（供上层推送给前端）；无运行上下文时返回 None
"""
run_log = current()
if run_log is None:
return None
summary = {
"promptTokens": run_log.prompt_tokens,
"completionTokens": run_log.completion_tokens,
"totalTokens": run_log.total_tokens,
"costMs": int((datetime.now() - run_log.started_dt).total_seconds() * 1000),
"toolCount": len(run_log.tools),
}
try:
await asyncio.to_thread(_do_finalize, run_log, success, answer, fail_msg)
except Exception as e:
logger.error(f"[invoke_log] 异步收尾失败: {e}")
finally:
reset_run()
return summary
def text_insert_invoke() -> str:
return (
f"INSERT INTO {TABLE_INVOKE_LOG} ("
"ID_, THREAD_ID, TASK_BATCH_ID, CHILD_AGENT_ID_, USER_INPUT, TOOL_INVOKE_COUNT, "
"STATUS, COST_TIME, COST_TOKEN, FAIL_MSG, SUCCESS_RESULT, SKILL_ID_, "
"TENANT_ID_, CREATE_DEP_ID_, CREATE_BY_, CREATE_TIME_, UPDATE_BY_, UPDATE_TIME_"
") VALUES ("
":id, :thread_id, :task_batch_id, :child_agent_id, :user_input, :tool_invoke_count, "
":status, :cost_time, :cost_token, :fail_msg, :success_result, :skill_id, "
":tenant_id, :create_dep_id, :create_by, :create_time, :update_by, :update_time"
")"
)
def text_insert_tool() -> str:
return (
f"INSERT INTO {TABLE_TOOL_INVOKE_LOG} ("
"ID_, AI_INVOKE_LOG_ID_, TOOL_NAME, TOOL_PARAMS, START_TIME, END_TIME, "
"COST_TIME, STATUS, SUCCESS_RESULT, FAIL_MSG, CREATE_BY_, CREATE_TIME_, UPDATE_BY_, UPDATE_TIME_"
") VALUES ("
":id, :invoke_log_id, :tool_name, :tool_params, :start_time, :end_time, "
":cost_time, :status, :success_result, :fail_msg, :create_by, :create_time, :update_by, :update_time"
")"
)
__all__ = [
"begin_run",
"current",
"reset_run",
"begin_tool_scope",
"end_tool_scope",
"record_usage",
"record_text_estimate",
"record_model_response",
"finalize_run",
"log_external_invoke",
]
11. 小结
这一篇修的不是"加个功能"，而是把第 15 篇留下的三处硬伤补齐：
| 修复 | 价值 |
|---|---|
| 工具作用域双写 | 从"只有总数"变成"能拆到每个工具" |
SELECT ID_ AS agent_id + 自动补建 | 汇总行终于能关联执行单元，按智能体聚合可用 |
copy_context() + 全链路记账 | 并行不丢账，口径覆盖所有模型调用 |
finalize_run 返回 summary | 前端能拿到，且与落库一致 |
| 提示词规则 + 硬去重 | 消除重复检索，token 直接降 60% |
一句话：token 统计的可信度，取决于"每个调用点都记账"和"记账与展示同源"这两件事，缺一个数字就不可信。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/qq_29467891/article/details/169457075
