12 · 生产级工程化

本章目标：把 RAG / Agent 从"能跑"带到"能上线"。核心就四件事——评估、可观测性、安全、成本与稳定性。

本章你将学到

1. 用测试集 + LLM-as-judge 建立评估闭环；
2. 追踪（Tracing）、指标、日志三板斧的可观测性实践；
3. 提示注入、工具越权等 Agent 特有的安全风险与对策；
4. 成本控制与稳定性（限流、超时、降级）的落地清单。

前置知识

• 已完成前 11 章任意一条完整路径；本章是"工程化"视角的总纲。

12.1 评估：没有评估，就没有优化

Agent 的输出是概率性的，"感觉变好了"不算数。生产级的前提是：每次改动都能量化"涨了还是跌了"。

12.1.1 评估体系分三层

工程顺序：先搭离线评估（门槛最低、收益最直接），再逐步补在线。

12.1.2 离线评估的测试集长什么样

每条样本 = 输入 + 期望输出 + 可选的相关上下文：

# eval_dataset.py：手工构造 20~50 条即可起步
EVAL_SET = [
    {"question": "航班起飞前4小时内退票，手续费是多少？",
     "answer_contains": ["40%"],            # 期望答案包含的关键信息
     "relevant": ["起飞前4小时内退票"]},      # 期望检索到的主题
    {"question": "婴儿票退票免手续费吗？",
     "answer_contains": ["免"], "relevant": ["婴儿票"]},
    # ... 覆盖：退票/改签/延误/特殊旅客/里程票
]

12.1.3 LLM-as-judge：用模型给模型打分

对开放式回答（"这个回答好不好"）难以写规则断言，主流做法是用另一个模型当裁判：

"""LLM-as-judge：给 Agent 的回答打分（0-5）。"""
import json
import os

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

judge = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)

JUDGE_PROMPT = """你是严格的质量评估员。根据以下维度给回答打分（0-5分，整数），并给出简要理由。

评分维度：
- 忠实度(faithfulness)：是否只依据提供的资料回答，有没有编造
- 完整度(completeness)：用户问题是否被完整回答
- 清晰度(clarity)：是否条理清晰、便于用户理解

【用户问题】{question}
【资料】{context}
【模型回答】{answer}

输出 JSON：{{"score": 0-5, "reason": "一句话理由"}}"""


def judge_answer(question: str, context: str, answer: str) -> dict:
    resp = judge.with_structured_output(
        type("Verdict", (), {"__annotations__": {"score": int, "reason": str}})
    ).invoke(JUDGE_PROMPT.format(question=question, context=context, answer=answer))
    return resp


# 使用：对测试集逐条执行 → 求平均分 → 写入报告
for item in EVAL_SET:
    answer = rag_chain.invoke(item["question"])          # 被测系统
    verdict = judge_answer(item["question"], item["relevant"], answer)
    print(item["question"], "=>", verdict)

要点：

• 裁判也要固定：judge 的模型、temperature、提示词都要版本化，否则"打分标准的波动"会污染对比结果；
• judge 的偏见：有的模型偏爱长回答。重要场景可让 judge 输出结构化理由，人工抽检；
• 检索质量的专门指标（Hit Rate / MRR / faithfulness / context_precision 等）见第 06 章的 Ragas。

12.1.4 评估的工程化

• 测试集与评分脚本挂进 CI：改 Prompt、改切分参数、升级模型前必须跑一遍，diff 看指标升降；
• 把历史指标存库（或 CSV），趋势可查——"上周还 4.2 分，这周怎么 3.8 了"要能定位；
• 留出集：调参时不要反复用同一批数据微调到过拟合（第 3 章提过）。

12.2 可观测性：追踪 + 指标 + 日志

Agent 是多次模型调用 + 工具调用 + 检索的组合，出问题最难定位。三板斧：

12.2.1 追踪（Tracing）：一次问答 = 一条 trace

• LangSmith / Langfuse（自托管可选）：自动记录每次 LLM 调用、工具结果、token 数、耗时，形成完整调用链；
• 开启方式极简（第 8.7 节）：设环境变量即可；
• 排查场景："用户说退款，模型调了 3 次检索还没答" → 看 trace 里每步的工具结果与相似度分。

12.2.2 指标（Metrics）：监控健康度

用 Prometheus + Grafana 或云监控实现；把 thread_id、trace_id 贯通到日志。

12.2.3 日志（Logging）：可解释性

• 记录输入与输出摘要（不要记完整 prompt 里的隐私数据）；
• 记录决策过程：模型调了哪个工具、参数是什么、结果是什么（Agent 可解释性的第一手材料）；
• 结构化日志（JSON）便于检索，把 user_id / thread_id / trace_id 放一起。

12.3 安全：Agent 特有的坑

参考 OWASP LLM Top 10，挑 Agent 系统最容易踩的几条：

三条核心原则：

1. 模型只做决策，不直接执行——权限与校验都在你的工具实现里；
2. 不可信内容（用户输入、检索文档）与可信指令隔离——用分隔符标记、用代码解析而非让模型"理解"；
3. 敏感操作永远要人工确认——interrupt 不是可选项。

12.4 成本与性能

Agent 一次问答可能触发多次模型调用，成本 = 次数 × 每次 token 数 × 单价。从三个方向控制：

12.4.1 省 token

• 上下文精简：检索只拼 Top-K 相关块（而非整库）；历史做窗口/摘要（第 11.2.1 节）；
• Prompt 瘦身：few-shot 示例够用就好，多余的 example 全是成本；
• 输出限制：max_tokens 按需设，避免模型啰嗦。

12.4.2 缓存

• LLM 结果缓存：相同问题直接返回历史答案（Redis 按 hash(问题+上下文) 做 Key）；
• Embedding 缓存：相同文本不重复调 Embedding API（知识库增量更新时尤其省钱）；
• 语义缓存：语义相近的问题复用答案（第 6 章检索同一套向量思路，成本敏感时再用）。

12.4.3 模型分级（Routing）

不同任务用不同价位的模型：

简单/高频（意图判断、分类、闲聊）→ 便宜的小模型（如 gpt-4o-mini / deepseek-chat）
复杂/关键（推理、长文档、高风险决策）→ 贵的大模型

用一个路由规则（关键字/意图分类/兜底）分发即可，能省 50% 以上成本。

12.4.4 稳定性三件套：限流、超时、降级

12.5 部署架构：Python Agent 服务怎么融入 Java 体系

你的生产系统大概率是 Java。两条路：

12.5.1 方案 A：Agent 独立服务（推荐）

[Java 业务服务] --(HTTP/gRPC)--> [Agent Service(Python/FastAPI + LangGraph)]
                                        │
                                        ├── LLM API
                                        ├── 向量库 / 知识库
                                        └── 内部业务接口（航班查询等）

• Agent 是"计算密集 + 外部调用密集"的独立领域，单独部署、独立扩容；
• Java 侧只需一个客户端（Feign / HTTP），把用户消息与 thread_id 传过去，拿流式结果；
• 适合：复杂 LangGraph 编排、RAG、多模型路由。

# FastAPI 包装 LangGraph（骨架）
from fastapi import FastAPI
from pydantic import BaseModel

from app.graph import build_agent

app = FastAPI()
agent = build_agent()   # 编译一次，全局复用（编译是重活，别放请求里）

class ChatRequest(BaseModel):
    thread_id: str
    message: str

@app.post("/chat")
def chat(req: ChatRequest):
    config = {"configurable": {"thread_id": req.thread_id}}
    result = agent.invoke({"messages": [("user", req.message)]}, config=config)
    return {"reply": result["messages"][-1].content}

12.5.2 方案 B：全 Java（LangChain4j / Spring AI）

简单场景（单模型 + 工具 + RAG）用 Java 生态直接落地，少一个服务。详见第 13 章。复杂编排（Multi-Agent、人工介入）目前 Python/LangGraph 更成熟，仍建议方案 A。

12.5.3 有状态问题

LangGraph 的记忆依赖 Checkpointer（thread_id）。部署要点：

• 会话亲和：同一 thread_id 的请求路由到同一副本（或用共享 Checkpointer，如 Postgres/Redis 实现），否则多副本下"记忆"串副本就丢了；
• 长会话超时：长期不活跃的会话清理 Checkpointer 记录。

12.6 上线清单（Checklist）

• [ ] 有 ≥20 条离线测试集，CI 跑评估，指标有历史趋势
• [ ] 全链路 Tracing 已接入，trace 里有工具调用与检索详情
• [ ] 记录 QPS/延迟/成功率/成本/工具失败率指标
• [ ] 高危工具已加 interrupt 人工审批 + 审计日志
• [ ] 检索按用户/部门做了权限过滤
• [ ] 提示注入防护：System 声明 + 输入/文档与指令隔离
• [ ] LLM 调用设了超时与重试；工具幂等性已确认
• [ ] 有降级路径（模型挂了怎么办、检索挂了怎么办）
• [ ] 会话亲和性/共享 Checkpointer 已确认
• [ ] 成本监控：每天能看到 token 账单
• [ ] 隐私合规：日志脱敏、数据最小化

12.7 小结

• 评估先行：测试集 + LLM-as-judge 挂进 CI，让"改进"可量化；
• 可观测三件套：追踪（一次问答一条 trace）、指标（健康与成本）、日志（决策过程可解释）；
• 安全底线：模型只决策不执行、不可信内容隔离、敏感操作人工审批；
• 成本 = 次数 × token × 单价：从上下文精简、缓存、模型分级三处下手；
• 稳定性：限流、超时、重试、降级缺一不可；
• 部署：复杂 Agent 独立成服务，Java 侧走 HTTP 调用；别忘了会话亲和。

下一章，回到你的主场——用 Java 生态落地：第 13 章。

练习

1. 给你的 RAG 链写 20 条测试集并跑通 12.1 的 judge 打分，把平均分和最低分样本记录下来。
2. 给 8.4 的客服助手加一个"工具失败率"指标埋点。
3. 设计一个高危工具（如 refund_ticket）的审批与审计方案：审批接口长什么样、日志记哪些字段。
4. 给 LLM 调用封装一个带超时+重试+降级的小工具类（参考 12.4.4）。

参考答案

<details>
<summary>练习1（20 条测试集 + judge 打分）</summary>

构造测试集并跑 judge：

TEST_SET = [
    {"question": "航班起飞前4小时退票手续费多少？", "ground_truth": "收取票面价 40%"},
    # ... 共 20 条，覆盖政策/航班/边界/知识库外问题
]

def run_eval():
    scores, low_samples = [], []
    for case in TEST_SET:
        answer = rag_chain.invoke(case["question"])
        score = judge_answer(case["question"], case["ground_truth"], answer)   # 12.1.3
        scores.append(score)
        if score < 3:                              # 记录最低分样本
            low_samples.append((case, answer, score))
    print("平均分：", sum(scores) / len(scores))
    for case, answer, score in low_samples:
        print(f"[{score}] {case['question']} -> {answer[:50]}...")

记录重点：平均分 + 最低分样本清单。对每个最低分样本做归因——是检索漏了（context_recall 问题）、上下文带噪（context_precision 问题）、还是生成不忠实（faithfulness 问题），对应第 6 章的修复手段。20 条的规模只够做冒烟回归，正式评估要上百条且持续扩充。
</details>

<details>
<summary>练习2（工具失败率埋点）</summary>

给 8.4 客服助手加指标。最简单的方式是在工具执行处打日志，再用 Prometheus 汇总：

# 工具执行统一入口加埋点
class ToolMetrics:
    def __init__(self):
        self.total = 0
        self.fail = 0

metrics = ToolMetrics()

def execute_tool(name: str, args: dict) -> str:
    metrics.total += 1
    try:
        result = TOOL_MAP[name].invoke(args)
        log_trace(name, args, result, ok=True)       # 写 trace（12.2 节）
        return result
    except Exception as e:
        metrics.fail += 1
        log_trace(name, args, str(e), ok=False)
        return f"工具执行失败：{e}"

# 暴露指标端点（Prometheus 格式，或打点后由 Agent 服务暴露）
def metrics_endpoint():
    rate = metrics.fail / metrics.total if metrics.total else 0
    return {"tool_total": metrics.total, "tool_fail": metrics.fail,
            "tool_fail_rate": rate}

为什么看失败率：工具失败率异常升高，通常意味着"上游接口挂了"或"模型在反复调用同一工具（死循环前兆）"。配合 12.2 的 trace，一条失败工具调用可以展开看参数、报错、后续行为，是排障的入口。指标按工具名分维度（tool_name 标签），能发现"只有某个工具老失败"。
</details>

<details>
<summary>练习3（高危工具审批与审计方案）</summary>

以 refund_ticket（退票）为例：

审批接口（人工工作台侧）：

POST /approval/tasks
  请求：{task_id, tool, params, applicant_user, amount, reason_summary}
响应：{task_id, status: pending|approved|rejected, approver, approved_at}

// 审批动作：
POST /approval/tasks/{task_id}/decide   {decision: "approved"|"rejected", approver, remark}

Agent 侧流程：模型请求退票 → 不直接执行，调用"申请审批"工具（而非退票工具）→ 生成审批任务 → interrupt 挂起 → 人工审批 → 通过后 Command(resume=...) 恢复并真正执行 refund_ticket（第 10.3 节模式）。

审计日志必记字段：

红线：审计日志只增不删（append-only，或写入独立审计库）；审批动作必须有操作人记录，不能是系统自动批准（否则审计失去意义）。
</details>

<details>
<summary>练习4（LLM 调用封装：超时+重试+降级）</summary>

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

# 1) 超时：模型调用设上限（网络慢、端点卡住不能无限等）
def llm_invoke(prompt: str, timeout: float = 20.0) -> str:
    return model.invoke(prompt, timeout=timeout).content

# 2) 重试：429 限流 / 5xx 用指数退避重试（别把限流也硬重试到爆）
def should_retry(exc: Exception) -> bool:
    return isinstance(exc, (TimeoutError, ConnectionError, RateLimitError))

@retry(retry=retry_if_exception_type((TimeoutError, ConnectionError, RateLimitError)),
       stop=stop_after_attempt(3),
       wait=wait_exponential(multiplier=1, max=8))
def llm_invoke_retry(prompt: str) -> str:
    return llm_invoke(prompt)

# 3) 降级：重试耗尽 → 返回预设兜底，而不是让上层异常
def safe_llm_invoke(prompt: str) -> str:
    try:
        return llm_invoke_retry(prompt)
    except Exception as e:
        log_error(e, prompt=prompt[:200])        # 记日志便于排查
        return "抱歉，服务暂时不可用，请稍后再试。"    # 用户侧兜底话术

# 使用：业务代码只调 safe_llm_invoke，把稳定性问题全封装住
reply = safe_llm_invoke(user_message)

要点：三个参数分开配——超时（单次上限）、重试次数（总预算）、降级策略（失败后行为）。生产还常叠加"熔断"（连续失败 N 次后快速失败一段时间），防止对已挂的端点继续打（第 12.4 节完整版）。这个封装等价于你在 Java 里用 Resilience4j/RetryTemplate 做的事，只是搬到 Python。
</details>

延伸阅读

• 第 06 章：Ragas 离线评估
• 第 13 章：Java 生态落地
• OWASP LLM Top 10（官方文档）
• LangSmith / Langfuse 官方文档（自托管 Langfuse 可选）