2026年，大模型应用正迈入全新阶段：核心关注点从“功能是否可用”转向“运行是否稳定”。
回顾过往，大家对大模型的注意力基本集中在模型效果本身——回答准确度如何、生成速度快慢、能否对接知识库、是否支持多轮对话。这些固然是基础能力，但当模型真正嵌入客服、办公、研发、运维、数据分析等核心业务场景后，新的挑战随即浮现。
例如，模型偶尔响应变慢了怎么办？一次请求为什么消耗了异常多的Token？RAG检索未能召回正确资料，问题出在哪里？Agent调用工具失败后，系统能否精准定位到具体出错步骤？
这些问题共同指向一个明确方向：AI可观测性。
传统可观测性主要关注服务器、容器、接口、数据库和中间件——CPU、内存、请求耗时、错误率、日志与调用链。但大模型应用的运行链路远比传统架构复杂。它不仅仅是普通的API接口，还涵盖Prompt、模型调用、Token消耗、向量检索、工具调用、Agent执行以及最终结果生成。
因此，AI可观测性绝不仅仅是“查看日志”那么简单，而是需要将每一次AI请求拆解为可追踪、可分析、可优化的完整运行链路。
为什么AI可观测性现在变得至关重要？
当AI应用还只是一个Demo时，偶尔失败或卡顿影响有限。但一旦进入真实业务系统，稳定性问题立刻凸显。
具体来说：
- 客服机器人不能长时间无响应；
- 代码助手不能频繁生成失败；
- RAG问答不能经常召回空内容；
- Agent在工具调用失败后，不能没有任何记录；
- 企业需要清楚了解每次模型调用消耗了多少Token。
这些问题如果没有有效监控，排查起来无异于大海捞针。
因此，大模型应用需要一套全新的指标体系：请求耗时、模型耗时、输入Token、输出Token、工具调用成功率、RAG召回数量、错误类型、单次请求状态。下面，我们用Python演示一个简化版的AI可观测性系统。
基础配置：定义指标日志文件
第一步，先定义日志文件和基础工具函数。
这里使用JSONL文件记录每一步的指标数据。JSONL的优势在于每行一条数据，便于后续导入日志系统或分析平台。
```python 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" ) ```这段代码相当于一个最小化的指标采集层。在实际生产环境中，这些数据可以写入日志服务、监控平台、时序数据库或数据仓库。
链路追踪：记录每个步骤耗时
第二步，为关键函数增加追踪能力。
这里使用装饰器记录函数的开始时间、结束时间、耗时、执行状态及错误信息。这样，每一步都会留下可追溯的运行记录。
```python 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系统不仅需要记录检索耗时，还要关注召回数量。若召回结果为空，通常意味着知识库、查询语义或向量索引可能存在异常。
```python @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以及模型延迟。在实际系统中，这些字段通常来自模型接口的返回结果。
```python 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工具调用。
在智能体系统中，模型经常需要调用搜索、数据库、文件、代码执行等工具。每次工具调用都应记录成功率和错误信息。
```python @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组装和模型生成。这里用一个函数把这些步骤串联起来。
```python 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": trace_id, "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系统运行状态的基本画像。
```python 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": total, "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顺利推向真实生产系统。
