Agent工程化三重门：编译校验、Schema契约与静态检查

1. 项目概述：从“会说话的玩具”到“能交付的工程师”

“别再让 Agent 光会表演”——这句话我第一次在内部技术复盘会上听到时，台下十几号人集体沉默了三秒。不是因为听不懂，而是太懂了。我们团队上个月刚上线一个面向金融风控场景的PI Agent，它能用自然语言解释模型决策路径、生成合规报告、甚至模拟监管问答。演示视频在管理层会议上放了三遍，掌声很响。结果上线第三天，它在处理一笔跨境支付流水时，把“USD”误识别为“USDA”，触发了错误的反洗钱规则链；第五天，它调用下游服务时传入的 JSON 字段名和接口文档定义的Schema完全对不上，返回 400 错误后直接静默退出，连日志都没打全。没人怪它“没逻辑”，大家心里都清楚：它压根没经过编译期校验，没走 Schema 校验流程，更没接入任何静态检查器。它只是个披着 Agent 外衣的、会流利胡说八道的聊天机器人。

这就是当前绝大多数所谓“AI Agent”项目的现实水位：能对话、能调用、能画图、能写诗，但只要碰上真实生产环境里的编译链路、数据契约、安全红线，立刻原形毕露。标题里说的“过编译、过 Schema、过检查”，不是三个并列动作，而是一条不可绕行的工程化铁律——它定义了“玩具”和“产品”的分水岭。这里的“编译”，不是指把 Python 源码变成字节码那种狭义编译，而是泛指Agent 行为逻辑的可验证性构建过程：你的提示词（Prompt）是否能被结构化解析？你的工具调用链是否能在执行前完成类型推导？你的决策树分支是否满足业务约束？这些，都需要一套类编译器的静态分析能力。“Schema”则直指数据契约的核心：Agent 输入的用户指令、中间状态、输出结果，必须严格遵循预定义的数据结构规范，不能靠运行时 try-catch 去兜底。“检查”是最后一道闸门，涵盖代码规范、安全策略、资源消耗阈值、甚至合规性审计点——比如金融场景下，Agent 生成的每份报告都必须带可追溯的签名与时间戳，这个签名生成逻辑本身，就得被纳入检查范围。

所以，这篇博文不讲怎么用 LangChain 搭个聊天机器人，也不教你怎么微调一个 LLM 让它更“拟人”。我们要拆解的是：一个真正能嵌入现有 CI/CD 流水线、能通过 Jenkins 编译任务、能被 SonarQube 扫描、能和 Spring Boot 服务共享同一套 OpenAPI Schema 的 Agent，它底层的骨架到底长什么样？它的“编译”环节要检查什么？它的 Schema 如何设计才能兼顾灵活性与强约束？它的检查项清单，为什么必须比传统后端服务还长？这些问题的答案，就藏在你每天写的那几行 Prompt、那个看似简单的 Tool Call 配置、以及你忽略掉的那行pydantic.BaseModel定义里。如果你正卡在 Agent 项目从 PoC 走向落地的最后一公里，或者正被“为什么线上总出奇奇怪怪的 500 错误”折磨得睡不着觉，那你接下来读的每一行，都是踩过坑的人递过来的扳手。

2. 内容整体设计与思路拆解：为什么必须把 Agent 当成“程序”来编译？

2.1 传统 Agent 架构的致命软肋：运行时即兴发挥

先看一张我们团队踩坑现场的截图：一个用于自动化生成数据库迁移脚本的 Agent，在测试环境跑得好好的，一上生产就报错sqlalchemy.exc.ArgumentError: Mapper mapped class MigrationPlan->migration_plan could not assemble any primary key columns for mapped table 'migration_plan'。排查三天，最后发现是 Agent 在解析用户需求“给用户表加个邮箱字段”时，生成的 Pydantic Model 定义里漏写了id: int = Field(primary_key=True)。这个错误根本不会在启动时暴露——因为 Model 是在运行时动态eval()出来的。这暴露了当前主流 Agent 框架（包括不少商业 PI Agent 平台）的共性缺陷：它们把“逻辑”和“数据”混在了一起，用字符串拼接代替类型系统，用运行时反射代替编译期校验。

我们来对比两种思路：

• “表演型”Agent 设计：用户输入 → LLM 生成一段包含 SQL 和 Python 代码的 Markdown 字符串 → 正则提取代码块 →exec()执行。整个过程像即兴喜剧，依赖 LLM 的临场发挥，没有语法树，没有 AST，没有符号表。好处是开发快；坏处是：你永远不知道下一次exec()会炸出什么。

• “工程师型”Agent 设计：用户输入 → 经过一个轻量级 Parser（如基于 Lark 的 DSL 解析器）→ 生成结构化的 AST（抽象语法树）→ AST 被送入 Type Checker（类型检查器）→ 检查所有变量引用是否在作用域内、所有函数调用参数类型是否匹配、所有 SQL 表名是否存在于元数据 Schema 中 → 通过后，AST 被 Code Generator（代码生成器）翻译成可执行的 Python 字节码或 SQL 语句 → 最终由 Runtime Executor 执行。

提示：这不是在造轮子。Lark 解析器 50 行代码就能搞定一个基础 DSL；Pydantic V2 的model_validate_json()方法配合自定义@field_validator，就是现成的 Schema 校验器；而ast.parse()+ast.NodeVisitor就是你的免费编译器前端。关键在于，你要有意识地把“编译”这个环节显式地、强制地塞进你的 Agent 生命周期里。

2.2 “编译”在 Agent 场景下的三层含义：从 Prompt 到字节码

很多人一听“Agent 编译”，第一反应是“LLM 又不生成 C 代码，编什么译？” 这是对“编译”概念的窄化。在现代软件工程中，“编译”的本质是将高级、模糊、人类友好的描述，转换为低级、精确、机器可验证的表示，并在此过程中捕获所有结构性错误。Agent 的“编译”同样分三层，缺一不可：

这三层编译，共同构成了 Agent 的“可信执行基线”。它意味着：在 Agent 第一行代码真正执行之前，你已经能 100% 确定它不会因为语法错误、类型错配、Schema 不符或权限缺失而崩溃。这才是“能过编译”的真实含义——不是让它跑起来，而是让它“不可能跑不起来”。

2.3 Schema 不是文档，是运行时契约：为什么 OpenAPI 3.0 是 Agent 的新 ABI

很多团队把 Schema 当成一份“参考文档”，放在 Confluence 里吃灰。但在 Agent 世界，Schema 是Agent 与外部世界交互的二进制接口（ABI）。想象一下：你的 Agent 要调用一个银行核心系统的转账接口，该接口要求{"from_account": "string", "to_account": "string", "amount": "number", "currency": "string"}，且currency必须是 ISO 4217 标准三字母代码（如"USD"）。如果 Agent 在运行时才去校验currency == "USD"，那当用户输入"usd"或"U.S. Dollar"时，它只能返回一个模糊的“参数错误”，用户体验极差。

真正的做法，是把这份契约提前固化：

from pydantic import BaseModel, Field, field_validator from typing import Literal class TransferRequest(BaseModel): from_account: str = Field(..., min_length=10, max_length=20, pattern=r'^[A-Z0-9]+$') to_account: str = Field(..., min_length=10, max_length=20, pattern=r'^[A-Z0-9]+$') amount: float = Field(..., gt=0.0, le=1000000.0) currency: Literal["USD", "EUR", "CNY", "JPY"] # 强制枚举，非字符串 @field_validator('from_account', 'to_account') def validate_account_format(cls, v): if not v.startswith('ACC'): raise ValueError('Account must start with "ACC"') return v

这段代码，就是你的 Agent 的“Schema 编译器”的输入。当 LLM 生成一个 JSON 对象时，TransferRequest.model_validate_json(json_str)会瞬间告诉你：

• 如果currency是"usd"→ValidationError: Input should be 'USD', 'EUR', 'CNY' or 'JPY'
• 如果from_account是"12345"→ValidationError: Account must start with "ACC"
• 如果amount是-100.0→ValidationError: Input should be greater than 0

注意：这里用的是 Pydantic V2 的model_validate_json()，它比 V1 的parse_obj()快 3-5 倍，且错误信息更精准。我们实测过，一个包含 20 个字段的复杂 Schema，校验耗时稳定在 0.8ms 以内，完全可以作为 Agent 请求入口的同步校验环节，无需异步化。

更重要的是，这个TransferRequest模型，可以一键导出为 OpenAPI 3.0 Schema：

from fastapi import FastAPI app = FastAPI() @app.post("/transfer") def transfer(req: TransferRequest): pass # 启动后访问 /openapi.json，即可获得标准 OpenAPI 文档

这意味着：你的 Agent 的输入 Schema，和你的 FastAPI 后端的 API Schema，是同一份源码。前端、后端、Agent，三方共享同一个契约。当银行系统升级，新增了transfer_reason字段，你只需要改一行transfer_reason: str，然后重新生成 OpenAPI，所有依赖方（包括 Agent 的校验逻辑）自动同步更新。这才是 Schema 的终极价值：消灭契约漂移，让变更可预测、可追溯、可自动化。

3. 核心细节解析与实操要点：手把手构建 Agent 的三重校验流水线

3.1 L1 Prompt 编译：用 DSL 解析器驯服 LLM 的“自由发挥”

LLM 的强大在于其泛化能力，但这也正是它在工程化场景中最危险的地方。你永远无法 100% 保证它下次生成的 JSON 字符串，格式和上次完全一致。解决方案不是禁用 LLM，而是给它戴上“语法镣铐”——用领域特定语言（DSL）约束它的输出空间。

我们以一个真实的电商客服 Agent 为例。它的核心任务是解析用户投诉：“我昨天买的 iPhone 15，屏幕有划痕，申请退货，订单号是 ORD-2024-789012”。传统做法是让 LLM 直接输出 JSON：

{ "intent": "return_request", "product": "iPhone 15", "issue": "screen_scratch", "order_id": "ORD-2024-789012" }

但 LLM 可能某次输出：

{ "intent": "return", "item": "iPhone 15", "problem": "scratched screen", "order": "ORD-2024-789012" }

字段名变了，值的格式也松散了。这时，你需要一个 DSL 解析器，强制 LLM 输出符合你定义的语法：

// 定义一个极简的投诉解析 DSL %import common.WS %import common.INT %import common.ESCAPED_STRING start: intent WS order_id WS product WS issue intent: "return_request" | "refund_request" | "exchange_request" order_id: "ORD-" INT "-" INT product: ESCAPED_STRING issue: "screen_scratch" | "battery_drain" | "camera_not_working" | "other" %ignore WS

这段 Lark 语法，定义了 LLM 必须输出的字符串格式，例如：return_request ORD-2024-789012 "iPhone 15" screen_scratch。它比 JSON 更紧凑，更难被 LLM “自由发挥”篡改。我们的实践是：在 System Prompt 里明确告诉 LLM：“你只能输出符合以下语法规则的字符串，不要加任何额外字符，不要加 JSON 大括号，不要加引号，只输出纯文本”。

然后，用 Lark 解析器进行编译：

from lark import Lark from lark.tree import Tree # 加载上面定义的 DSL 语法 parser = Lark(open("complaint_grammar.lark").read(), parser='lalr') def compile_prompt(prompt_text: str) -> dict: try: # LLM 输出的纯文本 tree = parser.parse(prompt_text.strip()) # 将 AST 转为字典 result = {} for child in tree.children: if isinstance(child, Tree) and child.data == 'intent': result['intent'] = child.children[0].value elif isinstance(child, Tree) and child.data == 'order_id': result['order_id'] = child.children[0].value # ... 其他字段解析 return result except Exception as e: raise ValueError(f"Prompt compilation failed: {e}") # 使用示例 raw_output = 'return_request ORD-2024-789012 "iPhone 15" screen_scratch' compiled = compile_prompt(raw_output) # {'intent': 'return_request', 'order_id': 'ORD-2024-789012', ...}

实操心得：DSL 语法越简单越好。我们最初设计了支持嵌套、数组的复杂语法，结果 LLM 错误率飙升。后来砍掉所有花哨功能，只保留平铺的键值对，错误率从 12% 降到 0.3%。记住，目标不是让 LLM 学会编程，而是让它学会填空。

3.2 L2 Tool Chain 编译：用调用图分析杜绝“死循环”和“类型雪崩”

Agent 的“工具调用”（Tool Calling）常被简化为一个for循环：LLM 说“调用 A”，就执行 A；A 返回结果，LLM 说“再调用 B”，就执行 B…… 这种线性思维在简单场景 OK，但在复杂业务流中，极易陷入“调用地狱”。我们曾遇到一个 Agent，为处理一个保险理赔请求，需要依次调用：get_policy_info→get_claim_history→calculate_payout→check_fraud_risk→get_policy_info（再次！）→send_notification。它在第 5 步又调用了get_policy_info，而这个调用的参数，竟然是上一步check_fraud_risk的返回值里一个不存在的字段policy_id_hash。结果就是无限重试，直到超时。

解决之道，是把 Tool Chain 当成一个有向图来编译和分析：

import networkx as nx from typing import Dict, List, Any # 定义所有可用工具及其签名 TOOLS = { "get_policy_info": { "params": {"policy_id": "str"}, "returns": {"policy_number": "str", "insured_name": "str", "coverage": "float"} }, "get_claim_history": { "params": {"policy_id": "str"}, "returns": {"claims": "list[dict]"} }, "calculate_payout": { "params": {"claim_id": "str", "policy_coverage": "float"}, "returns": {"payout_amount": "float", "currency": "str"} } } def compile_tool_chain(intent_ast: dict) -> nx.DiGraph: """ 根据用户意图 AST，生成一个调用图（DiGraph） 节点：工具名 边：数据流向（从上游工具的返回字段，到下游工具的输入参数） """ G = nx.DiGraph() # 1. 添加节点 for tool_name in intent_ast.get("required_tools", []): G.add_node(tool_name) # 2. 添加边（数据依赖） for step in intent_ast.get("execution_plan", []): tool_name = step["tool"] for param_name, source in step["params"].items(): if source.startswith("output."): # 来自上游工具输出 upstream_tool = source.split(".")[1] G.add_edge(upstream_tool, tool_name, param=param_name, source_field=source.split(".")[2] if len(source.split(".")) > 2 else None) # 3. 静态检查 if nx.is_directed_acyclic_graph(G) is False: raise ValueError("Tool chain contains cycle! Cannot execute.") # 4. 类型检查：检查每条边的 source_field 类型，是否匹配 target_param 类型 for u, v, data in G.edges(data=True): upstream_returns = TOOLS[u]["returns"] downstream_params = TOOLS[v]["params"] if data["source_field"] not in upstream_returns: raise ValueError(f"Field {data['source_field']} not found in output of {u}") if upstream_returns[data["source_field"]] != downstream_params[data["param"]]: raise ValueError(f"Type mismatch: {u}.{data['source_field']} ({upstream_returns[data['source_field']]}) " f"-> {v}.{data['param']} ({downstream_params[data['param']]})") return G # 使用示例 intent_ast = { "required_tools": ["get_policy_info", "calculate_payout"], "execution_plan": [ {"tool": "get_policy_info", "params": {"policy_id": "ORD-2024-789012"}}, {"tool": "calculate_payout", "params": {"policy_coverage": "output.get_policy_info.coverage"}} ] } graph = compile_tool_chain(intent_ast) # 成功返回图，且已通过循环和类型检查

这个compile_tool_chain函数，就是你的 Agent 的“链接器”（Linker）。它在执行前，就完成了：

• 循环检测：确保调用图是无环的（DAG），杜绝死循环；
• 字段存在性检查：确保output.get_policy_info.coverage这个字段，真的在get_policy_info的返回 Schema 里；
• 类型一致性检查：确保coverage字段的类型（float）和calculate_payout的policy_coverage参数类型（float）完全一致。

注意事项：networkx的is_directed_acyclic_graph()检查非常快，即使图有 100 个节点，耗时也在 0.1ms 级别。我们把它放在 Agent 的plan()阶段，作为run()的前置条件。所有失败都抛出ValueError，并附带清晰的错误信息，方便运维定位。

3.3 L3 Runtime 编译：用 SQLAlchemy Core 生成“永不 SQL 注入”的查询

当 Agent 需要操作数据库时，“拼 SQL 字符串”是最高危的操作。哪怕你用了f-string+escape_string()，也防不住 LLM 生成的'; DROP TABLE users; --。真正的解决方案，是让 Agent 的“SQL 生成”也经过编译——用 SQLAlchemy Core 这样的 ORM 底层，把查询逻辑编译成参数化查询（Parameterized Query）。

假设你的 Agent 需要根据用户指令“查一下张三最近三个月的订单”，生成 SQL。传统做法：

# 危险！绝对不要这样写 user_name = "张三" time_range = "3 months" sql = f"SELECT * FROM orders WHERE customer_name = '{user_name}' AND created_at > NOW() - INTERVAL '{time_range}'"

正确做法，是定义一个“查询编译器”：

from sqlalchemy import select, text, func from sqlalchemy.dialects.postgresql import INTERVAL class QueryCompiler: def __init__(self, metadata): self.metadata = metadata # SQLAlchemy MetaData 对象，包含所有表结构 def compile_order_query(self, customer_name: str, months: int) -> str: """ 将高层语义编译为安全的 SQL 字符串 返回的是已参数化的 SQL，可直接传给 execute() """ orders_table = self.metadata.tables['orders'] stmt = select(orders_table).where( orders_table.c.customer_name == customer_name ).where( orders_table.c.created_at > func.now() - text(f"INTERVAL '{months} MONTH'") ) # 关键：compile() 会生成带占位符的 SQL，并返回参数字典 compiled = stmt.compile(compile_kwargs={"literal_binds": True}) # 注意：literal_binds=True 是为了演示，生产环境应使用参数化绑定 # 实际应返回 (str, dict) 元组，str 是带 ? 占位符的 SQL，dict 是参数映射 return str(compiled) # 使用 compiler = QueryCompiler(metadata) safe_sql = compiler.compile_order_query("张三", 3) # 输出：SELECT orders.* FROM orders WHERE orders.customer_name = '张三' AND orders.created_at > now() - INTERVAL '3 MONTH'

这个QueryCompiler的核心价值在于：它把“用户意图”（查张三的订单）和“数据库 Schema”（orders 表的结构、字段类型、索引）牢牢绑定在一起。如果orders表里根本没有customer_name字段，self.metadata.tables['orders'].c.customer_name这行代码在编译期（Agent 启动时）就会报错，根本不会等到运行时。

实操心得：我们把QueryCompiler的所有方法，都注册为 Agent 的@tool。当 LLM 说“调用 query_orders”，它传入的参数是{"customer_name": "张三", "months": 3}，@tool装饰器会自动调用compile_order_query()，生成安全 SQL，再交给execute()执行。整个过程，LLM 无需知道 SQL 是什么，它只负责“说人话”，编译器负责“写安全代码”。

4. 实操过程与核心环节实现：一个可落地的 Agent 编译流水线

4.1 从零搭建：Agent 编译流水线的 5 个核心组件

一个完整的、能嵌入 CI/CD 的 Agent 编译流水线，由以下 5 个核心组件构成。它们不是黑盒框架，而是你可以用 200 行以内 Python 代码搭出来的模块：

我们以一个最简化的电商退货 Agent 为例，展示这 5 个组件如何串联：

# agent_compiler.py from lark import Lark from pydantic import BaseModel, ValidationError import networkx as nx from sqlalchemy import select, text from typing import Dict, Any, List # 1. DSL Parser (Lark) complaint_parser = Lark(r""" start: intent WS order_id WS product WS issue intent: "return_request" | "refund_request" order_id: "ORD-" INT "-" INT product: ESCAPED_STRING issue: "screen_scratch" | "battery_drain" | "other" %import common.WS %import common.INT %import common.ESCAPED_STRING %ignore WS """) # 2. Schema Validator (Pydantic) class ComplaintSchema(BaseModel): intent: str order_id: str product: str issue: str # 3. Tool Graph Builder (NetworkX) def build_and_validate_graph(ast_dict: Dict[str, Any]) -> nx.DiGraph: G = nx.DiGraph() G.add_node("parse_complaint") G.add_node("validate_order") G.add_node("initiate_return") G.add_edge("parse_complaint", "validate_order") G.add_edge("validate_order", "initiate_return") if not nx.is_directed_acyclic_graph(G): raise ValueError("Cycle detected in tool graph") return G # 4. Query Compiler (SQLAlchemy) def compile_return_query(order_id: str) -> str: # 这里应连接到你的 metadata return f"SELECT * FROM orders WHERE order_id = '{order_id}'" # 简化示意 # 5. Check Runner (自定义) def run_pre_execution_checks(ast_dict: Dict[str, Any]) -> List[str]: errors = [] # 检查1：订单ID格式 if not ast_dict["order_id"].startswith("ORD-"): errors.append("Order ID must start with 'ORD-'") # 检查2：产品名长度 if len(ast_dict["product"]) < 3: errors.append("Product name too short") return errors # 主编译函数 def compile_agent_input(raw_llm_output: str) -> Dict[str, Any]: """Agent 的‘编译器’主入口""" # Step 1: Parse try: tree = complaint_parser.parse(raw_llm_output.strip()) # 手动提取 AST（实际项目用 Transformer） ast_dict = { "intent": tree.children[0].children[0].value, "order_id": tree.children[1].children[0].value, "product": tree.children[2].children[0].value, "issue": tree.children[3].children[0].value } except Exception as e: raise ValueError(f"Parse failed: {e}") # Step 2: Validate Schema try: validated = ComplaintSchema(**ast_dict) ast_dict = validated.model_dump() except ValidationError as e: raise ValueError(f"Schema validation failed: {e}") # Step 3: Build & Validate Tool Graph try: graph = build_and_validate_graph(ast_dict) except ValueError as e: raise ValueError(f"Tool graph validation failed: {e}") # Step 4: Compile Queries (此处简化) ast_dict["sql_query"] = compile_return_query(ast_dict["order_id"]) # Step 5: Run Custom Checks check_errors = run_pre_execution_checks(ast_dict) if check_errors: raise ValueError(f"Pre-execution checks failed: {check_errors}") return ast_dict # 使用 if __name__ == "__main__": # 模拟 LLM 输出 llm_output = 'return_request ORD-2024-789012 "iPhone 15" screen_scratch' try: compiled = compile_agent_input(llm_output) print("✅ Compilation successful!") print("Generated SQL:", compiled["sql_query"]) except ValueError as e: print("❌ Compilation failed:", e)

这段代码，就是你 Agent 的“编译器”。把它放进你的requirements.txt，在 CI 流水线里加一行python agent_compiler.py --input "$LLM_OUTPUT"，它就能在每次部署前，自动校验所有 Agent 的输入是否合法。它不依赖任何大模型 API，不联网，纯本地执行，毫秒级响应，100% 可控。

4.2 CI/CD 集成：让 Agent 编译成为 Jenkins 的一个构建步骤

很多团队觉得“Agent 编译”是个玄学概念，没法放进 Jenkins。其实，它比编译一个 Java 项目还简单。我们团队的 Jenkinsfile 片段如下：

pipeline { agent any stages { stage('Checkout') { steps { checkout scm } } stage('Compile Agent Logic') { steps { script { // 1. 安装依赖 sh 'pip install -r requirements-agent.txt' // 2. 运行编译器，传入测试用的 LLM 输出样本 // 这些样本来自 ./test_samples/ 目录，覆盖所有常见错误场景 sh ''' for sample in ./test_samples/*.txt; do echo "Compiling $sample..." python agent_compiler.py --input "$(cat $sample)" || exit 1 done ''' } } } stage('Run Unit Tests') { steps { sh 'pytest tests/agent_tests.py' } } stage('Deploy') { steps { sh 'kubectl apply -f k8s/agent-deployment.yaml' } } } }

关键点在于./test_samples/目录。它不是随便放几个例子，而是我们精心构造的“编译边界测试集”：

• valid_order.txt:return_request ORD-2024-789012 "iPhone 15" screen_scratch（正常用例）
• invalid_intent.txt:refund_request ORD-2024-789012 "iPhone 15" screen_scratch（非法 intent）
• missing_order_id.txt:return_request "iPhone 15" screen_scratch（缺少 order_id）
• sql_injection.txt:return_request ORD-2024-789012 "iPhone 15'; DROP TABLE orders; --" screen_scratch（SQL 注入尝试）

Jenkins 每次构建，都会运行这 4 个样本。只要有一个样本编译失败，整个构建就标红，阻止发布。这比等上线后被用户骂醒，成本低一万倍。

注意事项：agent_compiler.py必须是幂等的（Idempotent）。它不能修改任何外部状态，不能发网络请求，不能读写数据库。它就是一个纯函数：输入字符串，输出结构化对象，或抛出异常。这是它能稳定运行在 CI 环境里的前提。

4.3 生产环境监控：如何在 K8s 里观测 Agent 的“编译健康度”

编译通过，不代表万事大吉。你还需要在生产环境里，实时观测 Agent 的“编译健康度”。我们在 K8s 的 Deployment 中，添加了以下 Prometheus 指标：

# metrics.py from prometheus_client import Counter, Histogram # 编译成功率（按意图类型分） COMPILATION_SUCCESS = Counter( 'agent_compilation_success_total', 'Total number of successful compilations', ['intent', 'error_type'] # error_type 为空表示成功 ) # 编译耗时（P95, P99） COMPILATION_DURATION = Histogram( 'agent_compilation_duration_seconds', 'Compilation duration in seconds', buckets=[0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0] ) # 在 compile_agent_input 函数末尾添加 def compile_agent_input(raw_llm_output: str) -> Dict[str, Any]: start_time = time.time() try: # ... 原有编译逻辑 ... COMPILATION_SUCCESS.labels(intent=ast_dict["intent"], error_type="").inc() return ast_dict except ValueError as e: # 解析错误、Schema 错误、图错误等，都归类到 error_type error_type = "parse_error" if "Parse failed" in str(e) else \ "schema_error" if "Schema validation failed" in str(e) else \ "graph_error" COMPILATION_SUCCESS.labels(intent="unknown",