Posts

引言 AI应用的质量保证（QA）远比传统软件复杂。传统软件QA关注"功能是否正常"，而AI应用QA还需要关注"输出是否准确"、“行为是否安全”、“体验是否良好"等多个维度。2026年，AI质量保证已经发展成为一个完整的学科。本文将系统介绍AI质量保证体系。 AI质量保证框架 框架组成 AI质量保证体系 ├── 开发阶段QA │ ├── 提示测试 │ ├── 模型评估 │ └── 安全审查 ├── 发布阶段QA │ ├── 回归测试 │ ├── A/B测试 │ └── 灰度发布 ├── 运行阶段QA │ ├── 实时监控 │ ├── 用户反馈 │ └── 异常检测 └── 治理阶段QA ├── 合规审查 ├── 审计日志 └── 持续改进 质量维度 维度 说明 评估方法 准确性 输出信息是否正确 事实核查、专家评估 完整性 是否完整回答了问题 人工评估、LLM评估 一致性 相似输入的输出是否一致 一致性测试 安全性 是否拒绝有害请求 红队测试 公平性 是否存在偏见 偏见检测 延迟 响应时间是否可接受 性能监控 可靠性 系统是否稳定运行 可用性监控 成本 单次调用成本是否合理 成本监控 开发阶段QA 提示质量审查 class PromptQA: def review(self, prompt, test_cases): """ 提示质量审查 """ report = { "accuracy": self.test_accuracy(prompt, test_cases), "format": self.test_format(prompt, test_cases), "robustness": self.test_robustness(prompt), "safety": self.test_safety(prompt), "cost": self.estimate_cost(prompt) } report["overall_score"] = self.calculate_overall(report) report["recommendation"] = self.recommend(report) return report 安全审查清单 ### AI安全审查清单 □ 提示注入防御 - 是否有输入隔离？ - 是否有指令强化？ - 是否有输出过滤？ □ 有害内容防御 - 是否能拒绝暴力内容请求？ - 是否能拒绝违法内容请求？ - 是否有内容过滤器？ □ 隐私保护 - 是否会泄露用户数据？ - 是否会泄露系统信息？ - 是否有数据脱敏？ □ 公平性 - 是否存在性别偏见？ - 是否存在种族偏见？ - 是否存在年龄偏见？ □ 可靠性 - 高负载下是否稳定？ - 模型API故障时是否有兜底？ - 是否有超时处理？ 发布阶段QA 灰度发布流程 class CanaryRelease: def __init__(self, config): self.stages = [ {"name": "internal", "traffic": 0.0, "duration": "1d"}, {"name": "canary_1", "traffic": 0.01, "duration": "1d"}, {"name": "canary_5", "traffic": 0.05, "duration": "2d"}, {"name": "canary_20", "traffic": 0.20, "duration": "2d"}, {"name": "full", "traffic": 1.0, "duration": "permanent"} ] def evaluate_stage(self, stage, metrics): """ 评估灰度阶段是否可以推进 """ checks = { "error_rate_ok": metrics["error_rate"] < 0.01, "latency_ok": metrics["p95_latency"] < 3000, "satisfaction_ok": metrics["satisfaction"] > 4.0, "cost_ok": metrics["cost_per_request"] < 0.05, "no_safety_incidents": metrics["safety_incidents"] == 0 } return all(checks.values()) 运行阶段QA 实时监控系统 class AIQualityMonitor: def __init__(self): self.metrics = { "accuracy": RollingMetric(window=1000), "latency": PercentileMetric(), "error_rate": RateMetric(window=60), "user_satisfaction": RollingMetric(window=500), "cost": CostTracker(), "safety": SafetyMonitor() } def record_request(self, request, response, user_feedback=None): """记录每次请求""" self.metrics["latency"].record(response.latency) self.metrics["error_rate"].record(response.error) self.metrics["cost"].record(response.token_cost) if user_feedback: self.metrics["user_satisfaction"].record(user_feedback) # 异步分析准确性和安全性 asyncio.create_task(self.analyze_async(request, response)) def check_alerts(self): """检查告警""" alerts = [] if self.metrics["error_rate"].current > 0.05: alerts.append("错误率过高") if self.metrics["latency"].p95 > 5000: alerts.append("延迟超标") if self.metrics["user_satisfaction"].current < 3.5: alerts.append("用户满意度下降") if self.metrics["safety"].has_incident(): alerts.append("安全事件") return alerts 用户反馈收集 class FeedbackCollector: def collect(self, user_id, response_id, feedback_type, content): """ 收集用户反馈 """ feedback = { "user_id": user_id, "response_id": response_id, "type": feedback_type, # "thumbs_up", "thumbs_down", "rating", "text" "content": content, "timestamp": datetime.now() } # 存储 self.store(feedback) # 如果是负面反馈，触发分析 if feedback_type == "thumbs_down": asyncio.create_task(self.analyze_negative_feedback(feedback)) 异常检测 class AnomalyDetector: def detect(self, recent_outputs, baseline): """ 检测输出异常 """ anomalies = [] # 长度异常 recent_lengths = [len(o) for o in recent_outputs] if mean(recent_lengths) > baseline["length_mean"] * 1.5: anomalies.append("输出长度异常增加") # 拒绝率异常 recent_refusals = sum(1 for o in recent_outputs if "无法" in o) if recent_refusals / len(recent_outputs) > baseline["refusal_rate"] * 2: anomalies.append("拒绝率异常升高") # 重复率异常 if self.compute_diversity(recent_outputs) < 0.3: anomalies.append("输出多样性下降") return anomalies 质量治理 审计日志 class AuditLogger: def log(self, event_type, details): """ 记录审计日志 """ log_entry = { "timestamp": datetime.now().isoformat(), "event_type": event_type, "details": details, "version": self.current_version, "hash": self.compute_hash(details) } # 写入不可变日志 self.immutable_store.append(log_entry) 合规检查 ### AI合规检查清单 □ 数据合规 - 用户数据是否加密存储？ - 是否有数据保留策略？ - 是否满足GDPR/个人信息保护法？ □ 算法合规 - 是否有算法备案？ - 是否有安全评估报告？ - 是否满足深度合成管理规定？ □ 内容合规 - 是否有内容审核机制？ - 是否有违法内容过滤？ - 是否有未成年人保护？ □ 透明度 - 是否告知用户在使用AI？ - 是否提供反馈渠道？ - 是否有人工替代方案？ 持续改进 PDCA循环 Plan（计划）： - 设定质量目标 - 制定改进计划 Do（执行）： - 实施改进 - 收集数据 Check（检查）： - 分析数据 - 评估效果 Act（行动）： - 标准化成功经验 - 修正不成功的尝试 质量仪表板 class QualityDashboard: def generate(self): return { "overall_health": "green", # green/yellow/red "metrics": { "accuracy": {"current": 0.92, "trend": "↑", "target": 0.90}, "latency_p95": {"current": 1200, "trend": "→", "target": 2000}, "satisfaction": {"current": 4.3, "trend": "↑", "target": 4.0}, "error_rate": {"current": 0.003, "trend": "↓", "target": 0.01}, "cost_per_request": {"current": 0.02, "trend": "→", "target": 0.05} }, "recent_incidents": [...], "improvement_actions": [...] } 2026年新趋势 1. AI自监控 AI系统自己监控自己的输出质量，自动发现问题。 ...

为什么需要流式响应？ LLM生成一个完整回答可能需要5-30秒。如果等待完整响应再返回，用户体验极差。流式响应（Streaming）让用户看到"逐字打印"的效果，大幅降低感知延迟。 SSE（Server-Sent Events）方案 服务端实现 from fastapi import FastAPI from fastapi.responses import StreamingResponse import json app = FastAPI() @app.post("/chat/stream") async def chat_stream(request: dict): async def event_stream(): # 调用LLM的流式接口 async for chunk in llm.astream( messages=[{"role": "user", "content": request["message"]}] ): data = json.dumps({ "content": chunk.content, "role": "assistant" }, ensure_ascii=False) yield f"data: {data}\n\n" # 发送结束标记 yield f"data: [DONE]\n\n" return StreamingResponse( event_stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", # Nginx不缓冲 } ) 客户端实现 // 浏览器端SSE const eventSource = new EventSource('/chat/stream'); eventSource.onmessage = (event) => { if (event.data === '[DONE]') { eventSource.close(); return; } const data = JSON.parse(event.data); document.getElementById('output').innerHTML += data.content; }; eventSource.onerror = (error) => { console.error('SSE error:', error); eventSource.close(); }; Python客户端 import httpx import json async def stream_chat(url, message): async with httpx.AsyncClient() as client: async with client.stream( "POST", f"{url}/chat/stream", json={"message": message}, timeout=60.0 ) as response: async for line in response.aiter_lines(): if line.startswith("data: "): data = line[6:] if data == "[DONE]": break chunk = json.loads(data) print(chunk["content"], end="", flush=True) WebSocket方案 服务端 from fastapi import FastAPI, WebSocket app = FastAPI() @app.websocket("/ws/chat") async def websocket_chat(ws: WebSocket): await ws.accept() try: while True: message = await ws.receive_text() # 流式生成 async for chunk in llm.astream( messages=[{"role": "user", "content": message}] ): await ws.send_json({ "type": "token", "content": chunk.content }) await ws.send_json({"type": "done"}) except WebSocketDisconnect: print("Client disconnected") 客户端 const ws = new WebSocket('ws://localhost:8000/ws/chat'); ws.onopen = () => { ws.send(JSON.stringify({message: '你好'})); }; ws.onmessage = (event) => { const data = JSON.parse(event.data); if (data.type === 'token') { output.innerHTML += data.content; } else if (data.type === 'done') { console.log('Generation complete'); } }; 高级特性 打字机效果 async def typewriter_stream(text, delay=0.03): """模拟打字机效果的流式输出""" for char in text: yield char await asyncio.sleep(delay) 流式中断 class CancellableStream: def __init__(self): self.cancelled = False def cancel(self): self.cancelled = True async def stream(self, llm, messages): async for chunk in llm.astream(messages): if self.cancelled: break yield chunk 并行流式响应 async def parallel_stream(prompts, llm): """同时流式生成多个响应""" async def stream_one(prompt, index): result = [] async for chunk in llm.astream([{"role": "user", "content": prompt}]): result.append({"index": index, "content": chunk.content}) return result tasks = [stream_one(p, i) for i, p in enumerate(prompts)] results = await asyncio.gather(*tasks) # 交错输出 for i in range(max(len(r) for r in results)): for result in results: if i < len(result): yield result[i] 性能优化 缓冲控制 # Nginx配置：禁用缓冲 location /chat/stream { proxy_pass http://backend; proxy_buffering off; # 关闭代理缓冲 proxy_cache off; # 关闭缓存 chunked_transfer_encoding on; proxy_read_timeout 300s; } Token级vs字符级 # Token级流式（推荐）：每个token一个chunk async for token in llm.astream(messages): yield token # 字符级流式：将token拆分为字符 async for token in llm.astream(messages): for char in token.content: yield char await asyncio.sleep(0.01) # 添加微小延迟 结语 流式响应是LLM应用的标配功能。SSE适合单向流式输出，WebSocket适合需要双向交互的场景。关键点：禁用所有层级的缓冲、正确处理中断、保持连接稳定性。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 LLM应用的一个独特挑战是：即使代码没变，模型提供商更新模型版本也可能导致输出变化。同样，提示的微小修改可能在某些场景下引入退化。2026年，LLM回归测试已经成为AI应用的标配。本文将系统介绍回归测试策略。 什么是LLM回归测试 与传统回归测试的区别 维度 传统软件 LLM应用 变化来源 代码修改 代码+模型版本+提示修改 输出确定性 确定性 不确定（同输入可能不同输出） 测试方法 精确匹配 语义匹配/范围匹配 回归原因 代码bug 模型行为变化 LLM回归的场景 模型升级：从GPT-4升级到GPT-5 提示修改：优化提示时可能影响其他场景 配置变更：调整temperature、max_tokens等 依赖更新：嵌入模型、向量数据库等更新 模型版本：提供商静默更新模型 回归测试策略 策略一：黄金测试集 维护一个经过验证的"黄金"测试集： class GoldenTestSuite: def __init__(self): self.golden_cases = [ { "id": "gold_001", "input": "解释什么是递归", "expected_keywords": ["函数", "自身", "终止条件"], "expected_min_length": 100, "expected_max_length": 500, "must_not_contain": ["错误代码示例"], "category": "concept_explanation", "last_verified": "2026-06-15", "verified_by": "expert_001" }, # ... 更多黄金测试用例 ] def run(self, model_config): results = [] for case in self.golden_cases: response = call_llm(case["input"], **model_config) result = self.verify(response, case) results.append(result) return results def verify(self, response, case): checks = { "keywords_present": all(kw in response for kw in case["expected_keywords"]), "length_ok": case["expected_min_length"] <= len(response) <= case["expected_max_length"], "no_forbidden": not any(bad in response for bad in case.get("must_not_contain", [])) } checks["passed"] = all(checks.values()) return {"case_id": case["id"], "response": response, "checks": checks} 策略二：语义回归检测 不只检查精确匹配，还检查语义是否一致： def semantic_regression_check(old_response, new_response, threshold=0.85): """ 检查新旧响应的语义相似度 """ # 使用嵌入模型计算语义相似度 old_embedding = embed(old_response) new_embedding = embed(new_response) similarity = cosine_similarity(old_embedding, new_embedding) if similarity < threshold: return { "status": "potential_regression", "similarity": similarity, "old_response": old_response, "new_response": new_response } return {"status": "ok", "similarity": similarity} 策略三：多维回归检测 def multi_dimensional_regression(old_outputs, new_outputs): """ 多维度回归检测 """ dimensions = { "format": check_format_consistency, # 格式一致性 "length": check_length_distribution, # 长度分布 "sentiment": check_sentiment_shift, # 情感偏移 "quality": check_quality_degradation, # 质量退化 "safety": check_safety_regression, # 安全性退化 } results = {} for dim, checker in dimensions.items(): results[dim] = checker(old_outputs, new_outputs) return results 策略四：分布回归检测 检查输出分布是否发生变化： ...

为什么结构化输出如此重要？ LLM默认输出自由文本，但实际应用中我们通常需要结构化数据——JSON对象、表格、特定格式。不可靠的结构化输出会导致下游解析失败，是LLM应用从原型到生产的主要障碍之一。 方案一：Prompt工程 基础方案 STRUCTURED_PROMPT = """ 请按以下JSON格式输出，不要包含其他内容： { "intent": "用户意图分类", "confidence": 0.0-1.0之间的数值, "entities": [ {"type": "实体类型", "value": "实体值"} ], "response": "回复内容" } 用户输入：{input} """ Few-shot增强 FEW_SHOT_PROMPT = """ 请按JSON格式输出意图分析结果。 示例1： 输入：我想订一张明天去北京的机票 输出：{"intent": "book_flight", "confidence": 0.95, "entities": [{"type": "destination", "value": "北京"}, {"type": "date", "value": "明天"}], "response": "好的，我来帮您查询明天去北京的航班。"} 示例2： 输入：今天天气怎么样 输出：{"intent": "weather_query", "confidence": 0.9, "entities": [{"type": "date", "value": "今天"}], "response": "让我为您查询今天的天气。"} 现在请分析： 输入：{input} 输出： """ 优缺点 优点：简单通用，任何LLM都支持 缺点：不可靠，模型可能输出多余文本、格式错误、字段缺失 方案二：JSON Mode from openai import OpenAI client = OpenAI() response = client.chat.completions.create( model="qwen3-32b", messages=[{"role": "user", "content": "分析以下文本的情感，返回JSON"}], response_format={"type": "json_object"} # 强制JSON输出 ) result = json.loads(response.choices[0].message.content) JSON Mode保证输出是合法的JSON，但不保证包含特定字段。 ...

引言 A/B测试是验证AI系统效果最可靠的方法。与其依赖基准分数，不如在真实用户中进行对照实验。2026年，A/B测试已经成为AI产品迭代的标配流程。本文将系统介绍AI系统A/B测试的实践方法。 A/B测试基础 什么是A/B测试 将用户随机分为两组：A组使用版本A（如GPT-5），B组使用版本B（如Claude 4），比较两组的关键指标。 AI系统A/B测试的独特性 输出不确定性：同一输入可能产生不同输出 延迟变化：不同模型的响应速度不同 成本差异：不同模型的API成本可能差10倍 多维评估：不仅看准确率，还要看用户满意度、延迟、成本 实验设计 步骤一：定义假设 假设：使用Claude 4替代GPT-5作为客服机器人后端， 用户满意度将提升5%以上，且API成本降低30%以上。 步骤二：选择指标 指标类型 具体指标 说明 主要指标 用户满意度评分 核心评估指标 次要指标 任务完成率、首次解决率 辅助评估 护栏指标 延迟、错误率、成本 确保不恶化 业务指标 留存率、转化率 最终业务价值 步骤三：计算样本量 from scipy import stats def calculate_sample_size(baseline_rate, mde, alpha=0.05, power=0.8): """ 计算所需样本量 baseline_rate: 基线指标值 mde: 最小可检测效应 alpha: 显著性水平 power: 统计功效 """ z_alpha = stats.norm.ppf(1 - alpha/2) z_beta = stats.norm.ppf(power) p1 = baseline_rate p2 = baseline_rate + mde p_avg = (p1 + p2) / 2 n = ((z_alpha * sqrt(2 * p_avg * (1 - p_avg)) + z_beta * sqrt(p1 * (1 - p1) + p2 * (1 - p2))) ** 2) / (p2 - p1) ** 2 return ceil(n) 步骤四：流量分配 方案一：50/50均分 - A组：50%流量 - B组：50%流量 - 优点：最快达到统计显著 - 缺点：风险较高（B可能更差） 方案二：90/10渐增 - A组：90%流量（对照组） - B组：10%流量（实验组） - 优点：风险可控 - 缺点：需要更长时间 推荐：先10%灰度，确认无问题后扩到50% 实验执行 流量路由 class ABTestRouter: def __init__(self, experiment_id, variants): self.experiment_id = experiment_id self.variants = variants # {"A": 0.5, "B": 0.5} def assign(self, user_id): """ 根据用户ID分配实验组 """ # 使用一致性哈希，确保同一用户始终在同一组 hash_value = hash(f"{self.experiment_id}:{user_id}") bucket = hash_value % 100 / 100 cumulative = 0 for variant, ratio in self.variants.items(): cumulative += ratio if bucket < cumulative: return variant return list(self.variants.keys())[-1] 实验配置 experiment_config = { "id": "exp_2026_07_gpt5_vs_claude4", "name": "GPT-5 vs Claude 4 客服对比", "start_date": "2026-07-01", "end_date": "2026-07-14", "variants": { "A": { "model": "gpt-5", "system_prompt": "v1.0", "temperature": 0.7 }, "B": { "model": "claude-4-opus", "system_prompt": "v1.0", "temperature": 0.7 } }, "allocation": {"A": 0.5, "B": 0.5}, "primary_metric": "user_satisfaction", "guardrail_metrics": ["latency_p95", "error_rate", "cost_per_session"], "sample_size": 10000 # 每组 } 数据收集 def log_experiment_event(user_id, variant, event_type, event_data): """ 记录实验事件 """ event = { "timestamp": datetime.now().isoformat(), "experiment_id": "exp_2026_07_gpt5_vs_claude4", "user_id": user_id, "variant": variant, "event_type": event_type, # "request", "response", "feedback" "event_data": event_data } # 写入数据仓库 data_warehouse.insert("ab_test_events", event) 结果分析 统计显著性检验 def analyze_experiment(results_a, results_b, metric="satisfaction"): """ 分析实验结果 """ # 描述性统计 stats_a = { "mean": mean(results_a[metric]), "std": std(results_a[metric]), "n": len(results_a) } stats_b = { "mean": mean(results_b[metric]), "std": std(results_b[metric]), "n": len(results_b) } # t检验 t_stat, p_value = stats.ttest_ind(results_a[metric], results_b[metric]) # 效应量 pooled_std = sqrt((stats_a["std"]**2 + stats_b["std"]**2) / 2) cohens_d = (stats_b["mean"] - stats_a["mean"]) / pooled_std # 置信区间 diff = stats_b["mean"] - stats_a["mean"] se = sqrt(stats_a["std"]**2/stats_a["n"] + stats_b["std"]**2/stats_b["n"]) ci_lower = diff - 1.96 * se ci_upper = diff + 1.96 * se return { "stats_a": stats_a, "stats_b": stats_b, "difference": diff, "p_value": p_value, "significant": p_value < 0.05, "effect_size": cohens_d, "confidence_interval": (ci_lower, ci_upper) } 护栏指标检查 def check_guardrails(results_a, results_b, thresholds): """ 检查护栏指标 """ alerts = [] # 延迟检查 if results_b["latency_p95"] > thresholds["latency_p95"]: alerts.append(f"延迟超标：B组P95={results_b['latency_p95']}ms") # 错误率检查 if results_b["error_rate"] > thresholds["error_rate"]: alerts.append(f"错误率超标：B组={results_b['error_rate']}") # 成本检查 if results_b["cost_per_session"] > thresholds["cost_per_session"]: alerts.append(f"成本超标：B组={results_b['cost_per_session']}") return alerts 常见陷阱 陷阱一：提前停止 实验还没达到所需样本量就因为"B看起来更好"而停止。 ...

Function Calling的核心价值 Function Calling让LLM能够调用外部工具和API，从"对话助手"升级为"行动助手"。但实践中的挑战在于：如何让模型可靠地选择正确的工具、生成正确的参数、处理工具执行失败的情况。 工具定义最佳实践 清晰的工具描述 tools = [ { "type": "function", "function": { "name": "search_knowledge_base", "description": ( "搜索企业知识库中的文档。当用户询问公司政策、产品信息、" "技术文档等内部知识时使用此工具。\n" "注意：不要用于搜索公开互联网信息。" ), "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索关键词，使用自然语言描述要查找的内容" }, "department": { "type": "string", "enum": ["engineering", "sales", "hr", "finance"], "description": "限定搜索的部门范围，不指定则搜索全部" }, "limit": { "type": "integer", "description": "返回结果数量，默认5", "default": 5, "minimum": 1, "maximum": 20 } }, "required": ["query"] } } } ] 工具选择的引导 # 好的描述：明确什么场景用，什么场景不用 "description": "获取当前天气信息。当用户询问天气状况、温度、降水概率时使用。不要用于历史天气查询。" # 坏的描述：模糊不清 "description": "天气工具" # 模型不知道何时使用 参数验证 from pydantic import BaseModel, Field, validator class SearchParams(BaseModel): query: str = Field(..., min_length=2, max_length=500) department: str = Field("all", pattern="^(engineering|sales|hr|finance|all)$") limit: int = Field(5, ge=1, le=20) @validator('query') def sanitize_query(cls, v): # 移除潜在注入 v = v.replace('\n', ' ').replace('\r', ' ') return v.strip() def validate_tool_args(func_name, args): """验证工具参数""" schema_map = { "search_knowledge_base": SearchParams, } if func_name in schema_map: return schema_map[func_name](**args) return args 执行与错误处理 class ToolExecutor: def __init__(self, tools_dict, timeout=30): self.tools = tools_dict self.timeout = timeout async def execute(self, tool_name, arguments): # 验证工具存在 if tool_name not in self.tools: return {"error": f"Unknown tool: {tool_name}"} # 验证参数 try: validated = validate_tool_args(tool_name, arguments) except Exception as e: return {"error": f"Invalid arguments: {e}"} # 带超时执行 try: result = await asyncio.wait_for( self.tools[tool_name](**validated.dict()), timeout=self.timeout ) return {"result": result} except asyncio.TimeoutError: return {"error": f"Tool execution timed out after {self.timeout}s"} except Exception as e: return {"error": f"Tool execution failed: {str(e)}"} async def function_calling_loop(llm, messages, tools, executor, max_rounds=5): """Function Calling循环""" for round_idx in range(max_rounds): response = await llm.achat_completion( messages=messages, tools=tools, tool_choice="auto" ) msg = response.choices[0].message messages.append(msg) # 如果模型没有调用工具，返回最终回答 if not msg.tool_calls: return msg.content # 执行所有工具调用 for tool_call in msg.tool_calls: result = await executor.execute( tool_call.function.name, json.loads(tool_call.function.arguments) ) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False) }) return "达到最大工具调用轮次限制。" 并行工具调用 async def parallel_tool_calls(llm, messages, tools, executor): """支持并行工具调用""" response = await llm.achat_completion( messages=messages, tools=tools, tool_choice="auto" ) msg = response.choices[0].message messages.append(msg) if msg.tool_calls: # 并行执行所有工具调用 tasks = [] for tool_call in msg.tool_calls: task = executor.execute( tool_call.function.name, json.loads(tool_call.function.arguments) ) tasks.append((tool_call.id, task)) results = await asyncio.gather(*[t for _, t in tasks]) for (tool_call_id, _), result in zip(tasks, results): messages.append({ "role": "tool", "tool_call_id": tool_call_id, "content": json.dumps(result, ensure_ascii=False) }) return messages 工具调用日志 import structlog logger = structlog.get_logger() class LoggingToolExecutor: def __init__(self, executor): self.executor = executor async def execute(self, tool_name, arguments): log = logger.bind(tool=tool_name) log.info("tool_call_start", args=arguments) start = time.time() result = await self.executor.execute(tool_name, arguments) duration = time.time() - start if "error" in result: log.error("tool_call_error", error=result["error"], duration_ms=duration*1000) else: log.info("tool_call_success", duration_ms=duration*1000) return result 结语 Function Calling的可靠性来自清晰的工具描述、严格的参数验证、完善的错误处理和详尽的日志记录。将这些实践标准化，可以显著提升LLM Agent在生产环境中的稳定性。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 LLM应用的测试与传统软件测试截然不同：输入空间无限大、输出不确定、质量主观。2026年，LLM自动化测试已经形成了一套完整的方法论和工具链。本文将系统介绍如何构建LLM自动化测试体系。 LLM测试的挑战 挑战一：输入空间无限 传统软件的输入是结构化的（如表单字段），而LLM的输入是自由文本，可能的组合无限多。 挑战二：输出不确定 同一个输入，LLM可能给出不同的输出（即使temperature=0，不同版本也可能不同）。 挑战三：质量主观 “好的回答"的定义因任务、用户、上下文而异。 挑战四：成本高 每次测试都需要调用LLM API，大规模测试成本可观。 测试分层架构 ┌─────────────────────────────────┐ │ 端到端测试 (E2E) │ ← 模拟真实用户场景 ├─────────────────────────────────┤ │ 集成测试 │ ← 测试模块间协作 ├─────────────────────────────────┤ │ 提示测试 │ ← 测试提示效果 ├─────────────────────────────────┤ │ 单元测试 │ ← 测试单个功能 └─────────────────────────────────┘ 第一层：单元测试 测试LLM的基本功能： class TestLLMUnit: def test_json_output(self): """测试JSON输出格式""" response = llm.generate("输出JSON格式：{\"name\": \"test\"}") assert is_valid_json(response) def test_language_detection(self): """测试语言识别""" response = llm.generate("用中文回复：Hello") assert is_chinese(response) def test_length_constraint(self): """测试长度约束""" response = llm.generate("用50字以内回答：什么是AI？") assert len(response) <= 100 # 中文字符数 第二层：提示测试 测试提示在不同输入上的表现： class TestPrompt: def test_sentiment_analysis(self): """测试情感分析提示""" test_cases = [ {"input": "太棒了！", "expected": "positive"}, {"input": "太差了！", "expected": "negative"}, {"input": "还行吧。", "expected": "neutral"}, ] for case in test_cases: response = llm.generate(prompt.format(input=case["input"])) assert case["expected"] in response.lower() def test_robustness(self): """测试提示鲁棒性""" # 测试不同表述方式 inputs = [ "这部电影怎么样？", "你觉得这部电影如何？", "评价一下这部电影", ] responses = [llm.generate(prompt.format(input=i)) for i in inputs] # 检查输出是否一致（语义相似度） for i in range(len(responses)-1): similarity = compute_similarity(responses[i], responses[i+1]) assert similarity > 0.7 第三层：集成测试 测试多个模块协作： ...

RAG的核心挑战 RAG（检索增强生成）让LLM能够基于私有知识回答问题。但一个简单的RAG原型——文档切块→向量化→相似度检索→拼接到prompt——在实际使用中往往效果不佳。2026年的RAG优化需要从文档处理、检索质量、上下文管理和生成控制四个维度系统提升。 文档处理优化 智能分块 from langchain.text_splitter import RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter class SmartChunker: def __init__(self, chunk_size=512, chunk_overlap=64): self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap # 通用分割器 self.general_splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""] ) # Markdown分割器 self.md_splitter = MarkdownHeaderTextSplitter( headers_to_split_on=[ ("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3"), ] ) def chunk(self, text, doc_type="auto"): if doc_type == "auto": doc_type = self.detect_type(text) if doc_type == "markdown": # 先按标题分块，再按大小细分 md_chunks = self.md_splitter.split_text(text) final_chunks = [] for chunk in md_chunks: if len(chunk.page_content) > self.chunk_size * 2: sub_chunks = self.general_splitter.split_text(chunk.page_content) for sc in sub_chunks: sc.metadata = chunk.metadata final_chunks.append(sc) else: final_chunks.append(chunk) return final_chunks else: return self.general_splitter.split_text(text) def detect_type(self, text): md_indicators = ["# ", "## ", "- [", "```", "| "] if any(ind in text[:500] for ind in md_indicators): return "markdown" return "text" 表格与代码处理 class TableAwareChunker: """避免在表格中间切分""" def chunk(self, text): # 识别表格区域 table_pattern = r'(\|[^\n]+\|\n)+' chunks = [] last_end = 0 for match in re.finditer(table_pattern, text): # 表格前的文本正常切分 pre_text = text[last_end:match.start()] if pre_text.strip(): chunks.extend(self.split_text(pre_text)) # 表格作为完整块 chunks.append(text[match.start():match.end()]) last_end = match.end() # 剩余文本 if last_end < len(text): chunks.extend(self.split_text(text[last_end:])) return chunks 向量化优化 多向量表示 from sentence_transformers import SentenceTransformer class MultiVectorEncoder: def __init__(self): self.dense_model = SentenceTransformer('BAAI/bge-large-zh-v1.5') # 稀疏向量用于关键词匹配 self.sparse_model = None # 使用BM25或SPLADE def encode(self, text): # 密集向量：语义相似度 dense_vec = self.dense_model.encode(text, normalize_embeddings=True) # 稀疏向量：精确匹配 sparse_vec = self.sparse_model.encode(text) if self.sparse_model else None return { "dense": dense_vec, "sparse": sparse_vec, "text": text } 查询扩展 class QueryExpander: def __init__(self, llm): self.llm = llm async def expand(self, query): """使用LLM扩展查询""" prompt = f"""将以下查询改写为3个不同角度的表述，用于检索： 原始查询：{query} 输出格式： 1. [改写1] 2. [改写2] 3. [改写3]""" response = await self.llm.ainvoke(prompt) expansions = self.parse_expansions(response) expansions.append(query) # 保留原始查询 return expansions def parse_expansions(self, text): lines = text.strip().split('\n') return [line.split('.', 1)[1].strip() for line in lines if '.' in line] 检索优化 混合检索 import numpy as np class HybridRetriever: def __init__(self, vector_store, bm25_store, alpha=0.7): self.vector_store = vector_store # 密集向量检索 self.bm25_store = bm25_store # BM25稀疏检索 self.alpha = alpha # 混合权重 async def retrieve(self, query, k=5): # 密集检索 dense_results = await self.vector_store.asimilarity_search_with_score( query, k=k*2 ) # 稀疏检索 sparse_results = self.bm25_store.search(query, k=k*2) # 分数归一化 dense_scores = self.normalize_scores([s for _, s in dense_results]) sparse_scores = self.normalize_scores([s for _, s in sparse_results]) # 混合排序 combined = {} for (doc, _), score in zip(dense_results, dense_scores): doc_id = doc.metadata["id"] combined[doc_id] = combined.get(doc_id, 0) + self.alpha * score combined.setdefault(f"{doc_id}_doc", doc) for (doc, _), score in zip(sparse_results, sparse_scores): doc_id = doc.metadata["id"] combined[doc_id] = combined.get(doc_id, 0) + (1 - self.alpha) * score combined.setdefault(f"{doc_id}_doc", doc) # 排序返回top-k sorted_ids = sorted( [k for k in combined if not k.endswith("_doc")], key=lambda x: combined[x], reverse=True )[:k] return [combined[f"{did}_doc"] for did in sorted_ids] def normalize_scores(self, scores): if not scores: return [] scores = np.array(scores) if scores.max() > scores.min(): return (scores - scores.min()) / (scores.max() - scores.min()) return np.ones_like(scores) 重排序 from sentence_transformers import CrossEncoder class Reranker: def __init__(self, model_name="BAAI/bge-reranker-large"): self.model = CrossEncoder(model_name) def rerank(self, query, documents, top_k=5): # 构建query-document对 pairs = [(query, doc.page_content) for doc in documents] # 计算相关性分数 scores = self.model.predict(pairs) # 排序 ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True) return [doc for doc, _ in ranked[:top_k]] 上下文管理 动态上下文窗口 class ContextManager: def __init__(self, max_tokens=4096): self.max_tokens = max_tokens def build_context(self, query, retrieved_docs, conversation_history=None): # 预留生成空间 context_budget = self.max_tokens - 512 # 分配：历史对话30%，检索文档70% history_budget = int(context_budget * 0.3) docs_budget = context_budget - history_budget # 构建历史上下文 history_text = self.build_history(conversation_history, history_budget) # 构建文档上下文（按相关性排序） docs_text = self.build_docs(retrieved_docs, docs_budget) # 组装最终prompt context = f""" ## 历史对话 {history_text} ## 相关知识 {docs_text} ## 用户问题 {query} """ return context def build_docs(self, docs, budget): result = [] current_tokens = 0 for i, doc in enumerate(docs): doc_text = f"[{i+1}] {doc.page_content}\n" doc_tokens = len(doc_text) // 4 # 粗略估计 if current_tokens + doc_tokens > budget: break result.append(doc_text) current_tokens += doc_tokens return "\n".join(result) 引用标注 class CitationGenerator: def __init__(self, llm): self.llm = llm async def generate_with_citations(self, query, retrieved_docs): # 为每个文档分配编号 docs_context = "\n\n".join([ f"[{i+1}] {doc.page_content}" for i, doc in enumerate(retrieved_docs) ]) prompt = f"""基于以下参考资料回答问题。在回答中使用 [编号] 标注信息来源。 参考资料： {docs_context} 问题：{query} 要求： 1. 只使用参考资料中的信息 2. 用 [编号] 标注每条信息的来源 3. 如果资料中没有相关信息，说明"根据现有资料无法回答" """ response = await self.llm.ainvoke(prompt) return response 评估与迭代 RAG评估指标 class RAGEvaluator: def __init__(self, llm): self.llm = llm async def evaluate(self, query, response, retrieved_docs, ground_truth=None): metrics = {} # 1. 检索相关性 metrics["retrieval_relevance"] = await self.eval_retrieval( query, retrieved_docs ) # 2. 回答忠实度（是否基于检索内容） metrics["faithfulness"] = await self.eval_faithfulness( response, retrieved_docs ) # 3. 回答完整性 if ground_truth: metrics["completeness"] = await self.eval_completeness( response, ground_truth ) return metrics async def eval_faithfulness(self, response, docs): """评估回答是否忠实于检索内容""" prompt = f"""判断以下回答是否完全基于给定的参考资料。 参考资料：{' '.join(d.page_content[:200] for d in docs)} 回答：{response} 请输出： 1. 忠实/不忠实 2. 不忠实的部分（如有） """ result = await self.llm.ainvoke(prompt) return "忠实" in result 结语 RAG管线优化是一个系统性工程，从文档分块到检索策略、从上下文管理到生成控制，每个环节都需要精心设计。2026年的RAG最佳实践强调混合检索、重排序、动态上下文管理和引用标注——这些技术组合使用可以显著提升RAG系统的准确性和可靠性。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

从原型到生产的鸿沟 LangChain让构建LLM Agent原型变得非常简单——几十行代码就能实现一个能调用工具、检索知识的智能体。但从原型到生产环境，需要解决可靠性、性能、可观测性、成本控制等一系列工程问题。 基础架构 Agent框架选择 from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_openai import ChatOpenAI from langchain.tools import Tool from langchain.memory import ConversationBufferWindowMemory # 使用工具调用Agent（比ReFi更可靠） llm = ChatOpenAI( model="qwen3-32b", base_url="http://localhost:8000/v1", temperature=0.1, # 生产环境低温度 max_retries=3, timeout=30 ) # 工具定义 tools = [ Tool( name="search", func=search_function, description="搜索知识库中的信息" ), Tool( name="calculator", func=calculator_function, description="数学计算" ), ] # 记忆管理 memory = ConversationBufferWindowMemory( memory_key="chat_history", k=10, # 只保留最近10轮对话 return_messages=True ) agent = create_tool_calling_agent(llm, tools, prompt) executor = AgentExecutor( agent=agent, tools=tools, memory=memory, max_iterations=5, # 限制迭代次数 max_execution_time=60, # 超时60秒 early_stopping_method="generate", verbose=True ) 可靠性工程 错误处理与重试 from tenacity import retry, stop_after_attempt, wait_exponential import logging logger = logging.getLogger(__name__) class RobustAgentExecutor: def __init__(self, agent_executor, fallback_response="抱歉，我暂时无法处理这个请求。"): self.executor = agent_executor self.fallback = fallback_response @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10), retry_error_callback=lambda _: None ) async def invoke(self, input_data): try: result = await self.executor.ainvoke(input_data) # 验证结果 if not result or "output" not in result: raise ValueError("Invalid agent output") return result except TimeoutError: logger.warning(f"Agent timeout for input: {input_data}") return {"output": self.fallback} except Exception as e: logger.error(f"Agent error: {e}", exc_info=True) raise async def safe_invoke(self, input_data): """不抛异常的调用""" result = await self.invoke(input_data) return result or {"output": self.fallback} 工具调用验证 from pydantic import BaseModel, validator class SearchInput(BaseModel): query: str max_results: int = 5 @validator('query') def query_must_be_valid(cls, v): if not v or len(v.strip()) < 2: raise ValueError("Query too short") if len(v) > 500: raise ValueError("Query too long") return v.strip() @validator('max_results') def max_results_range(cls, v): if v < 1 or v > 20: raise ValueError("max_results must be 1-20") return v class ValidatedTool: def __init__(self, func, input_schema): self.func = func self.input_schema = input_schema async def __call__(self, **kwargs): # 验证输入 validated = self.input_schema(**kwargs) try: result = await self.func(**validated.dict()) # 验证输出 if not result: return "No results found" return result except Exception as e: logger.error(f"Tool error: {e}") return f"Tool execution failed: {str(e)}" 速率限制 import asyncio from datetime import datetime, timedelta class RateLimiter: def __init__(self, max_calls=10, window_seconds=60): self.max_calls = max_calls self.window = window_seconds self.calls = [] self.lock = asyncio.Lock() async def acquire(self): async with self.lock: now = datetime.now() # 清理过期记录 self.calls = [t for t in self.calls if now - t < timedelta(seconds=self.window)] if len(self.calls) >= self.max_calls: wait_time = self.window - (now - self.calls[0]).total_seconds() await asyncio.sleep(wait_time) self.calls.append(now) class RateLimitedAgent: def __init__(self, executor, rate_limiter): self.executor = executor self.limiter = rate_limiter async def invoke(self, input_data): await self.limiter.acquire() return await self.executor.ainvoke(input_data) 可观测性 链路追踪 from langchain.callbacks import BaseCallbackHandler import json import time class TracingCallbackHandler(BaseCallbackHandler): def __init__(self): self.traces = [] self.current_trace = None def on_chain_start(self, serialized, inputs, **kwargs): self.current_trace = { "chain": serialized.get("name", "unknown"), "start_time": time.time(), "inputs": str(inputs)[:500], "steps": [] } def on_llm_start(self, serialized, prompts, **kwargs): if self.current_trace: self.current_trace["steps"].append({ "type": "llm", "model": serialized.get("name", "unknown"), "start_time": time.time() }) def on_llm_end(self, response, **kwargs): if self.current_trace and self.current_trace["steps"]: step = self.current_trace["steps"][-1] step["end_time"] = time.time() step["duration"] = step["end_time"] - step["start_time"] step["tokens"] = response.llm_output.get("token_usage", {}) def on_tool_start(self, serialized, input_str, **kwargs): if self.current_trace: self.current_trace["steps"].append({ "type": "tool", "tool": serialized.get("name", "unknown"), "input": input_str[:200], "start_time": time.time() }) def on_tool_end(self, output, **kwargs): if self.current_trace and self.current_trace["steps"]: step = self.current_trace["steps"][-1] step["end_time"] = time.time() step["duration"] = step["end_time"] - step["start_time"] step["output"] = str(output)[:500] def on_chain_end(self, outputs, **kwargs): if self.current_trace: self.current_trace["end_time"] = time.time() self.current_trace["duration"] = ( self.current_trace["end_time"] - self.current_trace["start_time"] ) self.current_trace["output"] = str(outputs)[:500] self.traces.append(self.current_trace) self.current_trace = None # 使用 tracing = TracingCallbackHandler() result = executor.invoke( {"input": "What is the weather?"}, config={"callbacks": [tracing]} ) 结构化日志 import structlog logger = structlog.get_logger() class LoggingMiddleware: async def log_request(self, request_data, response_data, duration): logger.info( "agent_request", input_length=len(str(request_data)), output_length=len(str(response_data)), duration_ms=duration * 1000, agent_version="1.0.0", timestamp=datetime.now().isoformat() ) 成本控制 Token预算管理 class TokenBudget: def __init__(self, daily_budget=1000000): self.daily_budget = daily_budget self.used = 0 self.date = datetime.now().date() def consume(self, tokens): # 跨天重置 if datetime.now().date() != self.date: self.used = 0 self.date = datetime.now().date() self.used += tokens if self.used > self.daily_budget: raise BudgetExceededError( f"Daily budget exceeded: {self.used}/{self.daily_budget}" ) def remaining(self): return self.daily_budget - self.used class BudgetAwareAgent: def __init__(self, executor, budget): self.executor = executor self.budget = budget async def invoke(self, input_data): # 预估输入token estimated_input = len(str(input_data)) // 4 if self.budget.remaining() < estimated_input + 1000: return {"output": "今日配额已用尽，请明天再试。"} result = await self.executor.ainvoke(input_data) # 记录实际消耗 if "intermediate_steps" in result: total_tokens = sum( step.get("token_usage", {}).get("total_tokens", 0) for step in result["intermediate_steps"] ) self.budget.consume(total_tokens) return result 部署架构 FastAPI服务 from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI(title="LLM Agent API") class ChatRequest(BaseModel): message: str session_id: str = None max_tokens: int = 2048 class ChatResponse(BaseModel): response: str session_id: str latency_ms: float @app.post("/chat", response_model=ChatResponse) async def chat(request: ChatRequest): start_time = time.time() try: result = await agent.safe_invoke({ "input": request.message, "session_id": request.session_id }) latency = (time.time() - start_time) * 1000 return ChatResponse( response=result["output"], session_id=request.session_id, latency_ms=latency ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) 结语 LangChain Agent的生产化需要从可靠性、可观测性、成本控制三个维度系统性地构建基础设施。通过错误处理、速率限制、链路追踪和预算管理，可以将原型级别的Agent转变为可靠的生产服务。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 尽管自动评估和LLM-as-Judge在2026年已经非常成熟，但人工评估仍然是LLM评估的"黄金标准"。特别是对于创意写作、对话质量、文化敏感性等主观维度，人工评估不可替代。本文将系统介绍LLM人工评估的方法论。 为什么需要人工评估 自动评估的局限 创意任务：BLEU等指标无法衡量创意质量 文化敏感：自动指标无法检测文化不当内容 用户体验：用户满意度需要人类主观感受 安全边界：某些安全问题的判断需要人类常识 新颖性：自动指标倾向于奖励"常见"回答 人工评估的优势 可以评估主观维度 可以发现细微错误 可以判断文化适宜性 可以评估用户体验 可以提供改进建议 人工评估流程 步骤一：明确评估目标 评估目标: 评估对象: GLM-5 vs GPT-5 评估维度: - 回答准确性 - 表达流畅性 - 用户体验 - 安全性 评估任务: - 问答对话 (100条) - 创意写作 (50条) - 代码生成 (50条) - 多轮对话 (30组) 步骤二：设计评估规范 评分标准 ### 回答准确性（1-5分） 5分：完全正确，信息准确 4分：基本正确，有小瑕疵 3分：部分正确，有遗漏 2分：大部分错误 1分：完全错误 ### 表达流畅性（1-5分） 5分：表达自然流畅，无语法错误 4分：表达流畅，偶有小问题 3分：表达尚可，有一些不自然 2分：表达困难，语法错误多 1分：无法理解 ### 用户体验（1-5分） 5分：回答贴心、有帮助、超出预期 4分：回答有用、友好 3分：回答可以接受 2分：回答不太有用 1分：回答完全没用或令人不快 标注指南 ### 标注指南 1. 独立评估 - 请独立完成评估，不要参考其他评估者的评分 2. 不确定处理 - 如果不确定，请选择较低的分数 - 在"comment"字段说明不确定的原因 3. 安全性问题 - 如果回答包含以下内容，请标记为"unsafe"： - 歧视性言论 - 暴力内容 - 违法建议 - 隐私泄露 4. 文化敏感性 - 如果回答可能冒犯特定文化群体，请标记为"culturally_insensitive" 5. 对比评估 - 在对比两个模型时，请评估哪个"更好"，而不是分别打分 - "更好"的判断应该综合考虑所有维度 步骤三：选择评估者 评估者类型 类型 适用任务 成本 质量 领域专家 专业领域评估 高 最高 通用标注员 通用任务评估 中 中等 众包工人 简单分类任务 低 较低 用户群体 用户体验评估 中 高 评估者数量 每个样本建议3-5个评估者 计算评估者间一致性（Inter-Annotator Agreement） 如果一致性低（Krippendorff’s α < 0.6），需要修订标注规范 步骤四：执行评估 class HumanEvaluation: def __init__(self, config): self.config = config self.evaluators = self.recruit_evaluators(config) def create_evaluation_task(self, model_outputs): """ 创建评估任务 """ tasks = [] for output in model_outputs: task = { "id": output["id"], "input": output["input"], "output_a": output["model_a_output"], # 模型A输出 "output_b": output["model_b_output"], # 模型B输出 "dimensions": self.config["dimensions"], "guidelines": self.config["guidelines"] } tasks.append(task) return tasks def collect_results(self, tasks): """ 收集评估结果 """ results = [] for task in tasks: # 分发给多个评估者 evaluations = [] for evaluator in self.evaluators: eval_result = evaluator.evaluate(task) evaluations.append(eval_result) # 计算一致性 agreement = self.calculate_agreement(evaluations) # 如果一致性低，重新评估 if agreement < 0.6: evaluations = self.re_evaluate(task) results.append({ "task": task, "evaluations": evaluations, "agreement": agreement, "final_score": self.aggregate_scores(evaluations) }) return results 步骤五：分析结果 def analyze_evaluation_results(results): """ 分析人工评估结果 """ analysis = {} # 总体对比 model_a_wins = sum(1 for r in results if r["final_score"]["a"] > r["final_score"]["b"]) model_b_wins = sum(1 for r in results if r["final_score"]["b"] > r["final_score"]["a"]) ties = len(results) - model_a_wins - model_b_wins analysis["overall"] = { "model_a_win_rate": model_a_wins / len(results), "model_b_win_rate": model_b_wins / len(results), "tie_rate": ties / len(results) } # 按维度分析 for dim in ["accuracy", "fluency", "user_experience", "safety"]: dim_scores_a = [r["final_score"]["a"][dim] for r in results] dim_scores_b = [r["final_score"]["b"][dim] for r in results] analysis[dim] = { "model_a_avg": mean(dim_scores_a), "model_b_avg": mean(dim_scores_b), "p_value": t_test(dim_scores_a, dim_scores_b) } # 评估者一致性 analysis["agreement"] = { "avg_krippendorff_alpha": mean([r["agreement"] for r in results]), "low_agreement_cases": [r for r in results if r["agreement"] < 0.6] } return analysis 评估方法 方法一：绝对评分 让评估者对每个输出独立打分： ...