企业官网建设流程全解析

AI智能体项目实战记录（15）：日志记录 —— 在主管与子图添加日志切面，实现智能体调用记录查询与 token 统计

1. 引言

1.1 为什么智能体项目必须记录调用日志

一个对话请求背后往往发生了很多事：

用户提问 → 主管调用一次大模型做决策 → 可能调用 1~N 个工具（每个工具内部还可能有 HTTP / 数据库 / 向量库 / 大模型） → 主管再决策一次 → 总结节点再调用一次大模型，流式输出

如果没有日志，下面这些问题都无法回答：

• 这次请求到底调用了哪些工具？每个工具耗时多少？
• 主管决策用掉多少 token？总结用掉多少 token？
• 哪个工具最慢、失败率最高？
• 某个用户/某个会话的完整调用链路是什么？
• 子图内部的步骤有没有真正执行？

1.2 三层记录模型

本文的日志方案把一次请求拆成三层，分别落到两张表：

三层通过同一个TASK_BATCH_ID串起来：

1.3 设计原则

1. 一次请求一个上下文：用contextvar保存_RunLog，节点、线程、子图都能拿到，不用层层传参；
2. 切面式埋点：业务节点只关心"开始时间 / 结果 / 异常"，日志写入统一交给_record_tool_event（主管）和_record_step_tool（子图）；
3. 失败不影响主流程：日志写入异常只记 warning/error，不阻断对话；
4. token 优先真实值：模型返回usage就用真实值，没有就按字符估算；
5. 敏感参数脱敏：authorization、callback、ai_data_key不落库。

2. 表结构与字段说明

2.1ai_invoke_log（智能体调用记录）

2.2ai_tool_invoke_log（工具调用明细）

说明：SKILL_ID_当前存的是技能名而不是数值 ID，便于直接按rag、web-search统计；如果后续技能迁移到数据库，可以改存ai_skill.id。

3. 运行上下文：_RunLog与contextvar

3.1_RunLog结构

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_IDself.user_id=user_idor""self.thread_id=thread_idor""self.skill=skillor""self.question=(questionor"")[:1000]# USER_INPUT varchar(1000)self.started_dt=datetime.now()self.tools:List[Dict[str,Any]]=[]# 工具明细self.prompt_tokens=0self.completion_tokens=0self.lock=threading.Lock()self.error_message=""defadd_tool(self,event:Dict[str,Any])->None:withself.lock:self.tools.append(event)defadd_usage(self,prompt:Optional[int]=None,completion:Optional[int]=None)->None:withself.lock:ifprompt:self.prompt_tokens+=int(prompt)ifcompletion:self.completion_tokens+=int(completion)@propertydeftotal_tokens(self)->int:returnself.prompt_tokens+self.completion_tokens

关键点：

• tools是并发写的：主管图会用线程池并行执行多个工具，因此add_tool/add_usage都用threading.Lock保护；
• batch_id与run_id分开：run_id是汇总行主键，batch_id用来把同一请求下的所有行关联起来；
• 问题截断到 1000 字符：与USER_INPUT varchar(1000)对齐。

3.2 contextvar 与三个入口

_run_log_var:contextvars.ContextVar[Optional[_RunLog]]=contextvars.ContextVar("invoke_run_log",default=None)defbegin_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)returnrun_logdefcurrent()->Optional[_RunLog]:"""获取当前请求的日志上下文（图节点/工具线程内调用）"""return_run_log_var.get()defreset_run()->None:"""请求结束后清理上下文"""_run_log_var.set(None)

为什么用contextvar而不是把run_log作为参数层层传递？

• 图节点签名是 LangGraph 固定的，不能随便加参数；
• 工具执行可能在其它线程；
• 子图内部节点也需要写入同一条请求的日志，传参要穿透多层。

contextvar由asyncio.to_thread自动复制到工作线程；主管图里跨线程跑子图时用copy_context()显式复制（见 6.5）。

4. token 记账

4.1 优先使用模型返回的usage

defrecord_usage(prompt_tokens:Optional[int]=None,completion_tokens:Optional[int]=None)->None:"""把模型返回的 usage 计入当前运行（usage 字段缺失时调用方传 None，则不会累加）"""run_log=current()ifrun_logisNone:returnrun_log.add_usage(prompt_tokens,completion_tokens)

4.2 缺失时按字符估算

# 2 个字符估算为 1 个 token（中文场景的粗略折算）_CHARS_PER_TOKEN=2def_estimate_tokens(text:Any)->int:text=""iftextisNoneelsestr(text)returnmax(1,(len(text)+_CHARS_PER_TOKEN-1)//_CHARS_PER_TOKEN)defrecord_text_estimate(prompt_messages:List[Dict[str,Any]],answer_text:Any)->None:"""模型未返回 usage 时按字符估算 prompt/completion token 并计入"""run_log=current()ifrun_logisNone:returnprompt_text=""formsginprompt_messagesor[]:content=msg.get("content")ifisinstance(content,str):prompt_text+=content prompt=_estimate_tokens(prompt_text)completion=_estimate_tokens(answer_text)run_log.add_usage(prompt,completion)

关键点：

• 向上取整：(len + 1) // 2，避免短文本被算成 0；
• prompt 只拼字符串内容：多模态内容（列表）会被忽略，这是估算的已知局限；
• 估算只是兜底：真实成本统计应以usage为准。

4.3 统一入口record_model_response

defrecord_model_response(resp:Any,prompt_messages:List[Dict[str,Any]],answer_text:Any)->None:""" 统一的模型响应 token 记账入口： 优先使用 resp.usage（OpenAI 兼容响应对象），缺失时按字符估算。 """run_log=current()ifrun_logisNone:returnusage=getattr(resp,"usage",None)ifusageisnotNone:record_usage(getattr(usage,"prompt_tokens",None),getattr(usage,"completion_tokens",None),)else:record_text_estimate(prompt_messages,answer_text)

这样主管决策节点只需要一行：

invoke_log_service.record_model_response(resp,messages_for_selection,raw_output)

4.4 两个 token 埋点位置

流式部分的代码：

# 统计本轮总结生成消耗的 token（usage 优先，缺失按字符估算）ifusage_promptisnotNone:invoke_log_service.record_usage(usage_prompt,usage_completion)else:invoke_log_service.record_text_estimate(messages,full_answer)

4.5 估算的局限

• 中文按 2 字符 ≈ 1 token 只是经验值，真实值随模型分词器变化；
• 工具内部的 LLM 调用（如 RAG 生成、企业信息总结）目前没有接入record_model_response，因此汇总行的COST_TOKEN只包含"主管决策 + 最终总结"；
• 若需要把工具内部 token 也统计进来，应在工具的模型调用后补一行invoke_log_service.record_model_response(resp, messages, raw)（见 10.4 的可选增强）。

5. 主管图的日志切面

对应文件：app/graph/supervisor_graph.py

5.1 请求入口：begin_run

# 开启本次“智能体调用日志”上下文（主管汇总行 + 工具明细行共用）invoke_log_service.begin_run(user_id=user_id,thread_id=thread_id,skill=skill,question=latest_question,)

位置在SupervisorGraph.astream里、真正跑图之前。这样整个请求期间，任何节点、任何工具线程都能通过current()拿到同一个_RunLog。

5.2 主管决策记账

# 统计本轮主管决策消耗的 token（优先 usage，缺失按字符估算）invoke_log_service.record_model_response(resp,messages_for_selection,raw_output)

每轮supervisor都会记一次，所以多轮决策的 token 会累加，最终落到汇总行的COST_TOKEN。

5.3 工具切面：_record_tool_event

def_record_tool_event(tool_name:str,tool_args:Dict[str,Any],start_dt:datetime,result:Any,error:str|None)->None:"""把一次工具执行写入当前运行的调用日志上下文（ai_tool_invoke_log 明细）"""run_log=invoke_log_service.current()ifrun_logisNone:returnend_dt=datetime.now()try:# 工具参数脱敏：不记录 authorization/callback/ai_data_key 等敏感字段params={k:vfork,vin(tool_argsor{}).items()ifknotin("authorization","callback","ai_data_key")}exceptException:params={}try:params_json=json.dumps(params,ensure_ascii=False,default=str)exceptException:params_json=""display_name=tool_name tool_desc=TOOL_MAP.get(tool_name)iftool_descisnotNone:display_name=tool_desc.display_name cost_time=round((end_dt-start_dt).total_seconds(),6)status=1iferrorisNoneelse-1run_log.add_tool({"tool_name":tool_name,"display_name":display_name,"args":params,"tool_params":params_json,"start_time":start_dt,"end_time":end_dt,"cost_time"