企业官网建设流程全解析

大模型算法应用与 提示词工程：上线后怎样观察真实使用情况

早上9点接收的客诉：明明 Prompt 没变，回答质量却滑坡了

周一刚到工位，监控群里就叠了几十条客诉转接单。业务方反映，客服 Agent 在处理退换货约束时，开始频繁给出模糊甚至矛盾的推滚策略。排查第一反应是去查 Git 提交记录，但 Prompt 模板近两周没有任何变更，底座模型 API 调用的参数也维持在原始配置。这种情况在基于大模型的应用落地中屡见不鲜。基座模型提供方可能在后台进行了无声的版本小微调，或者线上用户的输入分布随营销活动发生了偏移，导致原本在测试集上表现良好的系统出现了性能衰减。

传统软件工程中，依靠固定输入和输出的单元测试可以覆盖大部分逻辑断言。但在 Prompt 工程落地后，输出具备了天然的随机性与非确定性。如果仅仅依赖线上用户的偶发投诉来发现质量滑坡，说明整体评估体系处于盲区状态。我们需要一套能够持续监测线上真实请求输出质量的打分与告警机制，把被动排障转化为主动观察。

影子链路与语义离散度采样：抓出不可见的输出漂移

为了在不增加线上请求响应延迟的前提下完成观察，应当将监控逻辑从主链路中剥离出来。使用代理网关或者消息队列异步抓取全量或按比例采样的请求-响应对，构建一条影子评估链路。影子链路接收到输入与输出后，首先解决的不是复杂的大模型打分，而是快速衡量当前输出在语义空间中的离散程度。

针对特定任务，通过预训练轻量级 Embedding 模型将输出转换为高维向量，并计算其与基准测试集中已知优良输出分布的余弦相似度距离。如果近期采样的输出向量聚类中心发生了显著偏移，或者方差突然变大，往往意味着模型在表达形式或逻辑链条上出现了漂移。在无监督阶段，语义离散度是感知输出异常极其敏锐的指标。

自动化断言与 LLM-as-a-Judge 评估流架构

仅凭向量距离无法定位具体的错误类型，应当引入分层校验机制。整体评估架构分为两层：硬性规则断言层与软性语义评估层。

硬性规则断言层负责捕获语法级别的崩溃。包括 JSON 格式是否完备、必填字段是否缺失、回答长度是否突破上下限，以及是否触碰了预设的关键词敏感黑名单。这部分校验计算成本极低，且执行速度在毫秒级，能够第一时间拦截严重的格式毁损故障。

软性语义评估层则引入轻量级或定制化的 LLM-as-a-Judge。将原始 Prompt、用户输入、模型输出以及评价 Rubric（评分标准）拼接为评测 Prompt，投递给评估模型。为了防止评估模型自身产生非确定性，评测 Prompt 应当约束评估模型仅输出结构化的 JSON 评分与简短依据，并采用温度值为 0 的采样策略。

面向生产环境的离线与在线异步打分流水线实现

下面是用 Python 实现的在线异步采样与分层评估流水线核心代码。代码包含了确定性 Schema 校验、语义离散度计算以及异步 LLM-as-a-Judge 调用逻辑。

import json import logging import asyncio from typing import Dict, Any, Optional, List import numpy as np logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") class PromptEffectivenessObserver: def __init__(self, judge_client: Any, baseline_embeddings: Optional[np.ndarray] = None): self.judge_client = judge_client self.baseline_embeddings = baseline_embeddings self.sample_queue: asyncio.Queue = asyncio.Queue() def check_deterministic_rules(self, response_text: str, expected_schema: Dict[str, Any]) -> tuple[bool, str]: """第一层：确定性规则与 JSON Schema 校验""" if not response_text or len(response_text.strip()) == 0: return False, "EMPTY_RESPONSE" try: data = json.loads(response_text) except json.JSONDecodeError: return False, "INVALID_JSON_FORMAT" for required_key in expected_schema.get("required", []): if required_key not in data: return False, f"MISSING_KEY_{required_key}" return True, "PASSED" def compute_embedding_drift(self, current_embedding: np.ndarray) -> float: """第二层：计算当前输出向量与基线分布的平均相似度漂移""" if self.baseline_embeddings is None or len(self.baseline_embeddings) == 0: return 0.0 # 计算与基线数据集的 Cosine 相似度均值 norm_current = current_embedding / (np.linalg.norm(current_embedding) + 1e-8) norm_baseline = self.baseline_embeddings / (np.linalg.norm(self.baseline_embeddings, axis=1, keepdims=True) + 1e-8) similarities = np.dot(norm_baseline, norm_current) avg_similarity = float(np.mean(similarities)) return 1.0 - avg_similarity # 漂移度定义为 1 - 相似度 async def evaluate_with_llm_judge(self, prompt: str, response: str, rubric: str) -> Dict[str, Any]: """第三层：异步 LLM-as-a-Judge 评分""" judge_prompt = f""" 你是一名严谨的 AI 算法效果评估员。请根据以下标准评估输出质量。 【原始 Prompt/输入】: {prompt} 【模型输出】: {response} 【评估标准】: {rubric} 请以 JSON 格式输出评估结果，必须包含 score (1-5分的整数) 和 reason (字符串)。 """ try: # 模拟异步调用大模型 API 打分 judge_output = await self.judge_client.generate(judge_prompt, temperature=0.0) result = json.loads(judge_output) return { "score": int(result.get("score", 1)), "reason": result.get("reason", "No reason provided"), "status": "SUCCESS" } except Exception as e: logging.error(f"LLM-as-a-Judge 执行失败: {str(e)}") return {"score": 0, "reason": str(e), "status": "ERROR"} async def process_shadow_sample(self, payload: Dict[str, Any]) -> Dict[str, Any]: """异步处理单条影子采样数据""" prompt = payload.get("prompt", "") response = payload.get("response", "") schema = payload.get("schema", {}) rubric = payload.get("rubric", "") current_emb = payload.get("embedding", None) # 1. 执行规则校验 is_valid, rule_msg = self.check_deterministic_rules(response, schema) if not is_valid: logging.warning(f"采样 ID {payload.get('id')} 硬规则拦截: {rule_msg}") return {"id": payload.get("id"), "passed": False, "stage": "RULE", "error": rule_msg} # 2. 计算语义漂移 drift_score = 0.0 if current_emb is not None: drift_score = self.compute_embedding_drift(np.array(current_emb)) if drift_score > 0.45: logging.warning(f"采样 ID {payload.get('id')} 语义漂移过高: {drift_score:.3f}") # 3. 异步 LLM 打分 judge_result = await self.evaluate_with_llm_judge(prompt, response, rubric) return { "id": payload.get("id"), "passed": judge_result.get("score", 0) >= 4, "rule_status": rule_msg, "drift_score": drift_score, "judge_score": judge_result.get("score"), "judge_reason": judge_result.get("reason") } class MockLLMClient: async def generate(self, prompt: str, temperature: float = 0.0) -> str: await asyncio.sleep(0.1) # 模拟网络延迟 return json.dumps({"score": 4, "reason": "回答符合约束条件，逻辑清晰"}) async def main(): observer = PromptEffectivenessObserver(judge_client=MockLLMClient()) sample_data = { "id": "req_99812", "prompt": "用户询问退货规则，请输出结构化方案", "response": '{"action": "RETURN", "policy_id": 102, "details": "支持7天无理由"}', "schema": {"required": ["action", "policy_id"]}, "rubric": "必须准确说明退货类型和政策ID", "embedding": [0.12, -0.43, 0.88, 0.05] } result = await observer.process_shadow_sample(sample_data) print("影子链路打分结果:", json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == "__main__": asyncio.run(main())

告警阈值设置与回滚闸门：避免假阳性炸群

线上评估流水线运转起来后，最大的风险在于告警噪音过大。如果评估模型偶尔打出低分就触发紧急响铃，不仅会引发团队的排障疲劳，还可能导致不必要的人工干预。

告警机制需要按时间窗或批次进行滑动统计。通常以 5 分钟或 100 条采样为一个计算窗口。当窗口内的 JSON Schema 校验失败率突破 2%，或者 LLM-as-a-Judge 平均评分跌破 3.8 分时，才触发二级告警。对于持续滑动窗口均低评的高危情况，系统应当联动配置中心，自动将 Prompt 版本回滚至上一个固化的 Stable 标记版本，或者将流量切流至规则兜底分支。通过这类自动化回滚闸门，可以将 LLM 效果不确定性控制在可承受的工程边界之内。