AI Agent 出问题后怎么追责？先把 Event Log 设计好-平芜编程栈

很多团队做 AI Agent，最开始关注的是两个问题：

它能不能理解任务？
它能不能调用工具？

但一旦 Agent 进入真实业务，你会发现还有一个更重要的问题：

它出问题之后，怎么查？

比如：

它为什么给用户发了那封邮件？
它为什么调用了这个工具？
它为什么选择了这个参数？
它读过哪些上下文？
它有没有触发过人工确认？
它失败后有没有重试？
它最终输出是由哪一步生成的？

如果这些问题答不上来，AI Agent 就不适合进入生产环境。

所以，Agent 系统除了 Prompt、工具和权限，还必须设计 Event Log。

Event Log 的目标不是简单记录一行日志，而是把 Agent 的关键行为变成可追踪事件。

一个最小 Agent 执行链路可能是：

User Request -> Context Load -> Planning -> Tool Selection -> Argument Generation -> Tool Call -> Tool Result -> Final Response

每一步都应该有事件。

先定义事件类型：

fromenumimportEnumfromdataclassesimportdataclass,fieldfromtypingimportDict,Any,OptionalfromdatetimeimportdatetimeimportuuidclassAgentEventType(str,Enum):USER_REQUEST="user_request"CONTEXT_LOADED="context_loaded"PLAN_CREATED="plan_created"TOOL_SELECTED="tool_selected"ARGUMENTS_GENERATED="arguments_generated"TOOL_CALLED="tool_called"TOOL_RESULT="tool_result"HUMAN_APPROVAL="human_approval"FINAL_RESPONSE="final_response"ERROR="error"@dataclassclassAgentEvent:event_id:strtrace_id:strsession_id:struser_id:strevent_type:AgentEventType timestamp:strpayload:Dict[str,Any]=field(default_factory=dict)parent_event_id:Optional[str]=None

每一次用户请求生成一个 trace_id。

defnew_trace_id()->str:returnf"trace_{uuid.uuid4().hex}"defnow_iso()->str:returndatetime.utcnow().isoformat()

写一个事件记录函数：

EVENT_STORE=[]deflog_event(trace_id:str,session_id:str,user_id:str,event_type:AgentEventType,payload:Dict[str,Any],parent_event_id:Optional[str]=None)->AgentEvent:event=AgentEvent(event_id=f"evt_{uuid.uuid4().hex}",trace_id=trace_id,session_id=session_id,user_id=user_id,event_type=event_type,timestamp=now_iso(),payload=payload,parent_event_id=parent_event_id)EVENT_STORE.append(event)returnevent

一个用户请求进来时：

trace_id=new_trace_id()session_id="session_001"user_id="user_123"request_event=log_event(trace_id=trace_id,session_id=session_id,user_id=user_id,event_type=AgentEventType.USER_REQUEST,payload={"message":"请帮我整理上周客户反馈，并生成三条跟进建议"})

加载上下文时：

context_event=log_event(trace_id=trace_id,session_id=session_id,user_id=user_id,event_type=AgentEventType.CONTEXT_LOADED,payload={"sources":["crm_feedback_summary","support_tickets"],"record_count":42,"contains_sensitive_data":False},parent_event_id=request_event.event_id)

Agent 生成计划时：

plan_event=log_event(trace_id=trace_id,session_id=session_id,user_id=user_id,event_type=AgentEventType.PLAN_CREATED,payload={"steps":["summarize_feedback","cluster_common_issues","generate_follow_up_suggestions"],"risk_level":"medium"},parent_event_id=context_event.event_id)

工具调用前后都要记录：

tool_event=log_event(trace_id=trace_id,session_id=session_id,user_id=user_id,event_type=AgentEventType.TOOL_CALLED,payload={"tool_name":"cluster_feedback","arguments":{"source":"support_tickets","range":"last_week"}},parent_event_id=plan_event.event_id)result_event=log_event(trace_id=trace_id,session_id=session_id,user_id=user_id,event_type=AgentEventType.TOOL_RESULT,payload={"tool_name":"cluster_feedback","ok":True,"result_summary":{"clusters":5,"top_issue":"response_time"}},parent_event_id=tool_event.event_id)

最终回复也要记录：

final_event=log_event(trace_id=trace_id,session_id=session_id,user_id=user_id,event_type=AgentEventType.FINAL_RESPONSE,payload={"response_type":"suggestions","suggestion_count":3,"requires_human_review":True},parent_event_id=result_event.event_id)

这样，当用户或者团队负责人追问“这条建议怎么来的”，系统不是靠猜，而是可以根据 trace_id 找到完整链路。

defget_trace(trace_id:str):return[eventforeventinEVENT_STOREifevent.trace_id==trace_id]foreventinget_trace(trace_id):print(event.timestamp,event.event_type,event.payload)

真实生产环境里，Event Log 应该进入数据库、消息队列或日志系统，而不是存在内存里。

可以设计表结构：

CREATETABLEagent_events(event_idVARCHAR(64)PRIMARYKEY,trace_idVARCHAR(64)NOTNULL,session_idVARCHAR(64)NOTNULL,user_idVARCHAR(64)NOTNULL,event_typeVARCHAR(64)NOTNULL,parent_event_idVARCHAR(64),payload JSONBNOTNULL,created_atTIMESTAMPNOTNULL);CREATEINDEXidx_agent_events_trace_idONagent_events(trace_id);CREATEINDEXidx_agent_events_user_idONagent_events(user_id);CREATEINDEXidx_agent_events_typeONagent_events(event_type);