## 概述

2026 年，大模型应用正在从“能不能用”进入“稳不稳定”的新阶段。

过去，企业关注大模型应用时，更多看重模型效果，例如回答是否准确、生成速度是否够快、是否能接入知识库、是否支持多轮对话。但当大模型真正进入客服、办公、研发、运维、数据分析等场景后，新的问题开始出现。

模型偶尔响应变慢怎么办？

一次请求为什么消耗了大量 Token？

RAG 检索为什么没有召回正确资料？

Agent 调用工具失败后，系统能不能定位是哪一步出错？

这些问题都指向一个新的方向：AI 可观测性。

传统可观测性主要关注服务器、容器、接口、数据库和中间件，例如 CPU、内存、请求耗时、错误率、日志和调用链。而大模型应用的运行链路更复杂，它不仅包括普通接口，还包括 Prompt、模型调用、Token 消耗、向量检索、工具调用、Agent 执行和最终生成结果。

因此，AI 可观测性不只是“看日志”，而是要把每一次 AI 请求拆成可追踪、可分析、可优化的运行链路。

----

## 一、为什么 AI 可观测性开始重要？

当 AI 应用只是 Demo 时，偶尔失败并不会造成太大影响。

但当它进入真实业务系统后，稳定性就会变得非常关键。

例如：

1. 客服机器人不能长时间无响应；

2. 代码助手不能频繁生成失败；

3. RAG 问答不能经常召回空内容；

4. Agent 不能在工具调用失败后没有记录；

5. 企业需要知道每次模型调用消耗了多少 Token。

这些问题如果没有监控，就很难排查。

所以，大模型应用需要一套新的指标体系，包括请求耗时、模型耗时、输入 Token、输出 Token、工具调用成功率、RAG 召回数量、错误类型和单次请求状态。

下面用 Python 写一个简化版 AI 可观测性系统。

----

二、基础配置：定义指标日志文件

第一步是定义日志文件和基础工具函数。

这里使用 JSONL 文件记录每一步指标。JSONL 的特点是一行一条数据，适合后续导入日志系统或分析平台。

```

import time

import json

import random

from datetime import datetime

from functools import wraps

METRIC_FILE = "ai_observability_metrics.jsonl"

def now_ms():

return int(time.time() * 1000)

def write_metric(metric):

with open(METRIC_FILE, "a", encoding="utf-8") as file:

file.write(

json.dumps(metric, ensure_ascii=False) "n"

)

```

这部分代码相当于最小化的指标采集层。

真实系统中，这些数据可以写入日志服务、监控平台、时序数据库或数据仓库。

----

三、链路追踪：记录每个步骤耗时

第二步是给关键函数加上追踪能力。

这里使用装饰器记录函数开始时间、结束时间、耗时、执行状态和错误信息。这样每一个步骤都会留下运行记录。

```

def trace_step(step_name):

def decorator(func):

@wraps(func)

def wrapper(*args, **kwargs):

trace_id = kwargs.get("trace_id", "unknown")

start_time = now_ms()

success = True

error_message = None

try:

return func(*args, **kwargs)

except Exception as error:

success = False

error_message = str(error)

raise

finally:

end_time = now_ms()

write_metric({

"trace_id": trace_id,

"step_name": step_name,

"start_time": start_time,

"end_time": end_time,

"duration_ms": end_time - start_time,

"success": success,

"error_message": error_message,

"timestamp": datetime.now().isoformat()

})

return wrapper

return decorator

```

这个装饰器可以复用在模型调用、向量检索、权限校验、工具调用等多个环节。

当请求变慢时，系统可以通过 `trace_id` 找到到底是哪一步耗时过高。

----

## 四、模拟 RAG 检索：记录召回数量

第三步是模拟 RAG 检索。

RAG 系统不仅要记录检索耗时，还要记录召回数量。如果召回结果为空，说明知识库、查询语义或向量索引可能存在问题。

```

@trace_step("rag_retrieval")

def rag_retrieval(query, trace_id=None):

time.sleep(random.uniform(0.05, 0.2))

if random.random() < 0.1:

docs = []

else:

docs = [

"知识片段一：这里是与问题相关的资料。",

"知识片段二：这里是另一个召回结果。"

]

write_metric({

"trace_id": trace_id,

"step_name": "rag_retrieval_result",

"query": query,

"doc_count": len(docs),

"timestamp": datetime.now().isoformat()

})

return docs

```

RAG 召回数量是 AI 应用中非常重要的指标。

如果用户提问经常召回不到内容，模型就可能开始猜测，最终影响回答可靠性。

----

## 五、模拟模型调用：记录 Token 和延迟

第四步是模拟大模型调用。

这里记录输入 Token、输出 Token、总 Token 和模型延迟。真实系统中，这些字段通常来自模型接口返回结果。

```

def estimate_tokens(text):

if not text:

return 0

chinese_chars = sum(

1 for char in text

if "u4e00" <= char <= "u9fff"

)

other_chars = len(text) - chinese_chars

return int(chinese_chars * 1.2 other_chars * 0.4)

@trace_step("llm_call")

def call_large_model(prompt, trace_id=None):

time.sleep(random.uniform(0.2, 0.8))

if random.random() < 0.08:

raise RuntimeError("model request timeout")

input_tokens = estimate_tokens(prompt)

output = "这是大模型生成的模拟回答。"

output_tokens = estimate_tokens(output)

write_metric({

"trace_id": trace_id,

"step_name": "llm_token_usage",

"input_tokens": input_tokens,

"output_tokens": output_tokens,

"total_tokens": input_tokens output_tokens,

"timestamp": datetime.now().isoformat()

})

return {

"answer": output,

"input_tokens": input_tokens,

"output_tokens": output_tokens

}

```

Token 指标的价值不只是成本统计。

它还能帮助团队判断 Prompt 是否过长、上下文是否冗余、输出是否异常，以及是否需要做提示词压缩。

----

## 六、模拟工具调用：记录 Agent 执行状态

第五步是模拟 Agent 工具调用。

在智能体系统中，模型经常需要调用搜索、数据库、文件、代码执行等工具。每一次工具调用都应该记录成功率和错误信息。

```

@trace_step("agent_tool_call")

def call_agent_tool(tool_name, payload, trace_id=None):

time.sleep(random.uniform(0.05, 0.3))

if random.random() < 0.12:

raise RuntimeError(f"tool call failed: {tool_name}")

return {

"tool_name": tool_name,

"payload": payload,

"result": "工具执行成功"

}

```

Agent 的稳定性，很大程度上取决于工具调用稳定性。

如果工具失败没有记录，整个系统就很难排查问题。

----

七、请求入口：串联完整 AI 链路

第六步是构建完整请求流程。

一次 AI 请求可能包含权限校验、RAG 检索、工具调用、Prompt 组装和模型生成。这里用一个函数把这些步骤串起来。

```

def build_prompt(query, docs, tool_result):

context = "n".join(docs)

prompt = f"""

你是一个企业 AI 助手。

请基于以下资料回答用户问题。

用户问题：

{query}

知识库资料：

{context}

工具结果：

{tool_result}

要求：

1. 回答准确；

2. 不编造事实；

3. 如果资料不足，请说明无法判断。

"""

return prompt

def handle_ai_request(user_id, query):

trace_id = f"trace-{now_ms()}-{random.randint(1000, 9999)}"

request_start = now_ms()

status = "success"

answer = None

try:

docs = rag_retrieval(

query=query,

trace_id=trace_id

)

tool_result = call_agent_tool(

tool_name="search_tool",

payload={"query": query},

trace_id=trace_id

)

prompt = build_prompt(

query=query,

docs=docs,

tool_result=tool_result["result"]

)

model_result = call_large_model(

prompt=prompt,

trace_id=trace_id

)

answer = model_result["answer"]

except Exception as error:

status = "failed"

answer = f"请求失败：{error}"

finally:

request_end = now_ms()

write_metric({

"trace_id": otterly.cn

"step_name": "request_summary",

"user_id": user_id,

"query": query,

"status": status,

"total_duration_ms": request_end - request_start,

"timestamp": datetime.now().isoformat()

})

return {

"trace_id": trace_id,

"status": status,

"answer": answer

}

```

这一步完成后，每一次请求都会有完整链路记录。

后续无论是排查错误、统计耗时，还是分析成本，都可以基于这些日志完成。

----

## 八、生成可观测性报告

最后一步是读取指标日志，生成简单报告。

报告包括总请求数、失败数、成功率和平均耗时。这是 AI 系统运行状态的基本画像。

```

def read_metrics():

metrics = []

try:

with open(METRIC_FILE, "r", encoding="utf-8") as file:

for line in file:

metrics.append(json.loads(line))

except FileNotFoundError:

return []

return metrics

def generate_report():

metrics = read_metrics()

summaries = [

item for item in metrics

if item.get("step_name") == "request_summary"

]

total = len(summaries)

failed = len([

item for item in summaries

if item.get("status") == "failed"

])

avg_latency = 0

if total > 0:

avg_latency = round(

sum(item["total_duration_ms"] for item in summaries) / total,

2

)

success_rate = 0

if total > 0:

success_rate = round((total - failed) / total * 100, 2)

return {

"report_name": "AI 可观测性运行报告",

"total_requests": 30657.t.kuaisou.com

"failed_requests": failed,

"success_rate": success_rate,

"avg_latency_ms": avg_latency,

"generate_time": datetime.now().isoformat()

}

if __name__ == "__main__":

questions = [

"什么是 AI 可观测性？",

"RAG 检索为空怎么办？",

"Agent 工具调用失败如何排查？"

]

for question in questions:

result = handle_ai_request(

user_id="user_001",

query=question

)

print(json.dumps(

result,

ensure_ascii=False,

indent=2

))

report = generate_report()

print(json.dumps(

report,

ensure_ascii=False,

indent=2

))

```

----

## 九、趋势判断

从这套流程可以看到，AI 可观测性正在成为大模型应用的重要基础设施。

过去，企业更关心模型效果；现在，企业还需要看清模型运行过程。

响应时间、Token 消耗、RAG 召回数量、Agent 工具调用成功率、错误类型和请求链路，都会成为 AI 系统的重要指标。

未来，大模型应用不会只拼“回答得好不好”，还会拼“运行得稳不稳、成本是否可控、问题能否快速定位”。

谁能更早建立 AI 可观测性体系，谁就更容易把大模型从 Demo 推向生产系统。","createTime":1782746908,"ext":{"closeTextLink":0,"comment_ban":0,"description":"","focusRead":0},"favNum":0,"html":"","isOriginal":0,"likeNum":0,