简介本资源是一套基于LangChain与LangGraph框架实现的智能客服Agent系统完整工程代码面向AI应用开发工程师、LLM Agent初学者及对话系统实践者解决多模块协同构建可落地客服Agent的核心难题。压缩包共27个文件含20个Python核心模块如IntentClassifier、DialogueStateTracker、LLMbasedSlotExtractor、HumanHandoffDecisionModel等、2个配置说明文本、1个README文档、1个Word附赠资源说明及1个JSON知识库总大小仅92KB轻量但结构完整便于快速部署与二次开发。已有134人学习下载资源附带可运行的Intelligent_Customer_Agent_Demo-master示例项目、单元测试集覆盖知识检索、对话状态跟踪、槽位提取等关键路径及环境配置指引开发者可直接复现意图识别→状态维护→知识检索→人工转接决策的全链路逻辑并通过模块化设计深入理解Agent系统各组件职责与协作机制。1. 这不是“搭个Agent”就能上线的客服系统LangChain LangGraph 构建的智能客服 Agent本质是状态可控、路径可溯、人机协同可干预的工业级对话流水线你在网上搜“LangChain 智能客服”十有八九看到的是加载 PDF → 创建向量库 → 写个create_react_agent→ 丢进gradio界面 → 宣称“5 分钟上线 AI 客服”。但真实业务里用户说“我要退订上个月的会员”模型却去查“如何开通 VIP”或者连续三次追问“您要办理什么业务”后直接卡死——这不是 LLM 不够强而是整个对话缺乏意图锚点、状态记忆、槽位约束和人工兜底开关。本项目标题里那个长长的下划线串不是炫技而是把一个能进银行/电信/电商生产环境的客服 Agent 拆解成六个必须咬合的齿轮意图分类器判别用户当前诉求类型、对话状态跟踪器记住已问/已填/待确认字段、知识检索系统精准召回政策条款而非泛泛而谈、人机转人工决策模型不是“识别到情绪就转人工”而是基于会话熵值槽位完成度历史投诉率动态触发、LLM-based 槽位提取器绕过 prompt 工程黑匣子用结构化输出强制校验、API 服务层把 LangGraph 的 state machine 封装成标准 REST 接口供前端/IVR/企微 Bot 调用。它不追求“最酷的 Agent 框架”而是在 LangChain 做组件编排、LangGraph 做状态驱动的前提下用最小侵入方式把 NLP 传统模块意图识别、DST、NER和 LLM 新范式结构化生成、self-refine拧成一股绳。适合正在从规则引擎或简单问答机器人升级、且已有标注数据/业务知识库/人工坐席 SOP 的中大型团队——如果你连“用户说‘改地址’到底算‘修改收货信息’还是‘变更账单地址’”都还没定义清楚先别碰这个 zip 包。2. 为什么选 LangGraph 而不是纯 LangChain Chain状态机才是客服对话不可妥协的底层契约LangChain 的RunnableSequence和AgentExecutor适合单轮问答或工具调用链但客服对话天然具备多轮、分支、回溯、中断、重置特性。比如用户说“我刚买了手机现在想退货”系统需先确认订单号槽位提取再查是否在 7 天内知识检索若超期则触发“特殊通道申请”分支状态跳转此时若用户突然插入“等等我其实是想换货”就必须撤销前序状态、重置槽位、切换意图——这种“带事务语义的状态迁移”LangChain 原生链式执行无法安全承载。LangGraph 的StateGraph提供了显式的节点Node、边Edge和状态State三元组每个节点是一个纯函数如intent_classifier_node每条边是一个条件函数如should_route_to_refund_or_exchange状态对象State则贯穿全程像数据库事务日志一样记录所有中间变量。这带来三个硬性收益一是调试时可精确回放某次会话的每一步状态快照二是新增分支如加“银行政策例外审批”流程只需插入新节点重定义边逻辑不影响主干三是人工坐席介入时能直接读取当前state[slots]和state[dialogue_history]无需解析一长串 LLM 输出文本。2.1 用 StateGraph 定义客服对话的六阶段状态流我们不抽象成“start → think → act → end”而是按实际客服 SOP 映射为六个原子状态节点from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List, Dict, Any import operator class DialogueState(TypedDict): user_input: str intent: str # e.g., refund, exchange, complaint slots: Dict[str, str] # e.g., {order_id: ORD-2024-XXXX, reason: damaged} dialogue_history: List[Dict[str, str]] knowledge_context: str # retrieved from vector DB should_transfer_to_human: bool transfer_reason: str # 定义状态图 workflow StateGraph(DialogueState) # 注册六个核心节点 workflow.add_node(intent_classification, intent_classifier_node) workflow.add_node(slot_filling, slot_extractor_node) workflow.add_node(knowledge_retrieval, knowledge_retriever_node) workflow.add_node(policy_check, policy_checker_node) workflow.add_node(response_generation, response_generator_node) workflow.add_node(human_transfer_decision, human_transfer_judge_node) # 定义边状态流转逻辑返回下一个节点名 def route_after_intent(state: DialogueState) - str: if state[intent] in [refund, exchange, complaint]: return slot_filling elif state[intent] inquiry: return knowledge_retrieval else: return response_generation # fallback def route_after_slots(state: DialogueState) - str: if not all(k in state[slots] for k in required_slots_for_intent(state[intent])): return slot_filling # 缺失槽位循环追问 else: return knowledge_retrieval # 绑定边 workflow.set_entry_point(intent_classification) workflow.add_conditional_edges(intent_classification, route_after_intent) workflow.add_conditional_edges(slot_filling, route_after_slots) workflow.add_edge(knowledge_retrieval, policy_check) workflow.add_edge(policy_check, response_generation) workflow.add_edge(response_generation, human_transfer_decision) workflow.add_conditional_edges(human_transfer_decision, lambda x: END if not x[should_transfer_to_human] else END) # 实际中此处应跳转至人工队列接口提示required_slots_for_intent()是一个业务函数例如refund必须含order_id和return_reasoncomplaint必须含complaint_type和evidence_description。它不是写死在代码里而是从 YAML 配置文件加载方便运营人员调整。2.2 状态对象设计为什么TypedDict比dict更安全初学者常把state当作普通字典传结果在slot_filling节点里误写state[sltos]少了个o运行时才报错。我们强制使用TypedDict配合 mypy 静态检查# dialogue_state.py from typing import TypedDict, List, Dict, Optional, Literal class DialogueState(TypedDict): user_input: str intent: Literal[refund, exchange, complaint, inquiry, other] slots: Dict[str, str] dialogue_history: List[Dict[str, str]] # [{role: user, content: ...}, ...] knowledge_context: str should_transfer_to_human: bool transfer_reason: Optional[str] # 以下字段用于调试和审计 timestamp: float session_id: str llm_call_count: int # 记录本次会话调用了几次 LLM这样在slot_extractor_node函数签名里写def slot_extractor_node(state: DialogueState) - DialogueState:IDE 就能实时提示state.slots可用state.sltos直接标红。更重要的是当新增一个state[customer_tier]字段时所有用到DialogueState的节点函数都会被 mypy 报错逼你显式处理——这在多人协作、长期迭代的客服系统里比任何文档都管用。2.3 边条件函数用lambda写业务逻辑是自杀行为看到网上教程用lambda x: knowledge_retrieval if x[intent] inquiry else slot_filling千万别照抄。客服规则远比这复杂“inquiry” 类意图若用户身份是 VIP查 CRM 接口返回is_vipTrue跳转vip_knowledge_retrieval节点若用户近 3 天有投诉记录查风控 API则无论意图都先走risk_assessment节点“refund” 意图若订单金额 5000 元需额外校验manager_approval_required。正确做法是把边逻辑封装成独立函数并注入依赖def route_to_knowledge_or_risk(state: DialogueState, crm_client, risk_api_client) - str: # 先查风控 if risk_api_client.has_recent_complaint(state[session_id]): return risk_assessment # 再查 VIP customer_info crm_client.get_by_session(state[session_id]) if state[intent] inquiry and customer_info.get(is_vip): return vip_knowledge_retrieval # 普通流程 if state[intent] in [refund, exchange, complaint]: return slot_filling return response_generation然后在构建 workflow 时传入实例workflow.add_conditional_edges( intent_classification, lambda state: route_to_knowledge_or_risk( state, crm_clientCRMClient(), risk_api_clientRiskAPIClient() ) )这样单元测试可以 mockCRMClient和RiskAPIClient验证不同输入下的路由结果而不是等上线后才发现“VIP 用户没走 VIP 流程”。3. 意图分类器与槽位提取器别让 LLM 自由发挥用结构化输出锁死业务边界很多团队把意图识别和槽位填充全交给 LLMprompt 写成“请判断用户意图并提取订单号、原因等字段用 JSON 格式输出”。结果模型偶尔输出intent: refunnd拼写错误、order_id: 没有提供未拒绝、甚至嵌套 Markdown 表格。这不是模型能力问题而是把校验责任推给下游——policy_check节点拿到{intent: refunnd}是该报错、忽略、还是猜成refund线上系统不能猜。本方案采用“LLM 规则后处理”双保险LLM 只负责生成候选规则引擎做终审。3.1 意图分类器用轻量级分类模型打底LLM 做兜底和拒识我们不训练 BERT 大模型而是用sklearn的LinearSVC TF-IDF特征维度控制在 5000 以内在 2 万条标注数据上达到 92% 准确率。为什么不用 LLM因为意图分类是封闭集、高频率、低歧义任务LLM 的 token 开销和延迟不划算。但 LLM 用在两个地方拒识Out-of-Distribution Detection当分类模型置信度 0.7调用 LLM 判断是否属于新意图如用户说“你们APP闪退了”原意图为technical_issue但训练集没覆盖意图归一化模型输出cancel_subscriptionLLM 根据业务词典映射为标准名subscription_cancellation。# intent_classifier.py from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.svm import LinearSVC import joblib class IntentClassifier: def __init__(self, model_pathmodels/intent_svm.pkl, vectorizer_pathmodels/tfidf.pkl): self.vectorizer joblib.load(vectorizer_path) self.model joblib.load(model_path) self.intent_mapping { cancel_sub: subscription_cancellation, stop_service: subscription_cancellation, turn_off: subscription_cancellation } def predict(self, text: str) - tuple[str, float]: vec self.vectorizer.transform([text]) pred self.model.predict(vec)[0] proba self.model.decision_function(vec)[0] # 返回最高置信度和对应意图 confidence max(proba) if hasattr(proba, __len__) else abs(proba) return self.intent_mapping.get(pred, pred), confidence # 在 LangGraph 节点中调用 def intent_classifier_node(state: DialogueState) - DialogueState: classifier IntentClassifier() intent, confidence classifier.predict(state[user_input]) if confidence 0.7: # LLM 拒识兜底 llm_response llm.invoke(f用户说“{state[user_input]}”这是一个客服对话场景。 请判断1. 是否属于已知意图refund/exchange/complaint/inquiry/other2. 如果不属于请给出最接近的新意图名称不超过3个词。 只输出JSON如{{is_known: true, intent: refund}} 或 {{is_known: false, suggested_intent: app_crash}}) result json.loads(llm_response.content) if result[is_known]: intent result[intent] else: intent other # 或存入待审核队列 return {intent: intent}3.2 LLM-based 槽位提取器用 Pydantic 强制结构用 ReAct 模式防幻觉槽位提取必须满足输出格式绝对稳定不能有时是 dict有时是 list缺失字段明确为空字符串而非None或缺失 key对模糊表述如“上个月的订单”能结合上下文推断需访问state[dialogue_history]。我们用 Pydantic V2 的BaseModel定义槽位 Schema并让 LLM 以 ReActReasoning Action模式思考from pydantic import BaseModel, Field from typing import Optional class RefundSlots(BaseModel): order_id: str Field(default, description必须为12位数字字母组合如 ORD-2024-XXXX) return_reason: str Field(default, description从[damaged, wrong_item, not_received, changed_mind]中选择一个) refund_method: str Field(default, description从[original_payment, store_credit, bank_transfer]中选择) def slot_extractor_node(state: DialogueState) - DialogueState: # 动态加载当前意图对应的槽位模型 slot_model get_slot_model(state[intent]) # 返回 RefundSlots, ExchangeSlots 等 # 构造 ReAct Prompt prompt f你是一个客服对话槽位提取器。请严格按以下步骤操作 1. 阅读用户最新输入“{state[user_input]}” 2. 结合历史对话“{json.dumps(state[dialogue_history][-3:], ensure_asciiFalse)}” 3. 从用户输入中提取指定字段缺失则留空字符串 4. 输出必须是合法JSON且只包含 {slot_model.__name__} 的字段 {slot_model.model_json_schema()} response llm.invoke(prompt) try: # Pydantic 自动校验并补全默认值 slots slot_model.model_validate_json(response.content) return {slots: slots.model_dump()} except Exception as e: # 校验失败时记录原始响应供人工复核 logger.warning(fSlot extraction failed for {state[session_id]}: {e}, raw: {response.content}) return {slots: {k: for k in slot_model.model_fields.keys()}}参数说明get_slot_model(intent)返回对应 Pydantic 模型确保不同意图的槽位字段隔离。model_validate_json()会自动将缺失字段设为默认值空字符串并将非法值如return_reason: broken转换为最近似合法值damaged或抛异常触发 fallback。3.3 对话状态跟踪器DST不是记录所有话而是维护一个可查询的“对话事实库”传统 DST如 TRADE用序列标注预测槽位值但本方案的 DST 更轻量它不预测只聚合。每次slot_extractor_node输出新槽位DST 节点就将其 merge 到全局state[slots]中并标记来源用户直输 / LLM 推断 / 默认值。关键在于支持跨轮次冲突解决用户第一轮说“订单号 ORD-123”第二轮说“等等是 ORD-456”DST 应覆盖用户说“我要退这个”未提订单号DST 保持order_id为空不猜测若历史中有order_id且本轮未提及则保留旧值。def dst_node(state: DialogueState) - DialogueState: # 合并新槽位旧值优先级低于新值除非新值为空 new_slots state.get(slots, {}) merged_slots state[slots].copy() if slots in state else {} for key, value in new_slots.items(): if value.strip(): # 非空才覆盖 merged_slots[key] value return {slots: merged_slots}这个 DST 没有复杂模型但胜在确定性——它不“理解”对话只忠实地执行“最后有效输入覆盖”规则避免 LLM 在多轮中自我矛盾。4. 知识检索系统与人机转人工决策模型让 LLM 有据可依让人工介入有理有据客服系统最大的坑不是 LLM 说错话而是它一本正经地胡说八道。用户问“iPhone 15 保修期多久”模型凭记忆回答“1年”而实际政策是“整机1年电池2年”。更危险的是当模型不确定时它不会说“我不知道”而是编造一个看似合理的答案。本方案用两道闸门堵住知识检索系统确保 LLM 的输入上下文来自可信源人机转人工决策模型确保“不知道”时不是沉默而是优雅移交。4.1 知识检索系统不用 RAG 的“向量相似度”而用“语义-结构双路召回”纯向量检索如 ChromaDB对“iPhone 15 保修期”这类短问效果好但对“我上个月买的手机屏幕碎了能免费换吗”就容易召回“屏幕维修价格表”而非“碎屏保障条款”。我们采用双路召回语义路用 sentence-transformers 模型all-MiniLM-L6-v2做向量检索召回 top-5 文档块结构路用关键词匹配 正则从政策文档中提取带标签的片段如policy typewarrantytitle碎屏保障/titlecontent...优先返回结构化片段。from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings import re class HybridRetriever: def __init__(self, vector_db_pathvector_db, policy_xml_pathpolicies.xml): self.vector_db Chroma(persist_directoryvector_db_path, embedding_functionHuggingFaceEmbeddings(model_nameall-MiniLM-L6-v2)) self.policy_tree ET.parse(policy_xml_path) def retrieve(self, query: str) - str: # 语义路召回 semantic_docs self.vector_db.similarity_search(query, k3) # 结构路召回提取 query 中的关键实体用 spaCy 简单 NER entities extract_entities(query) # e.g., [iPhone 15, screen] structural_snippets [] for entity in entities: # 在 XML 中搜索含 entity 的 policy 节点 for policy in self.policy_tree.findall(.//policy): title policy.find(title).text.lower() if policy.find(title) is not None else if entity.lower() in title or entity.lower() in policy.find(content).text.lower(): structural_snippets.append(policy.find(content).text) # 合并结果结构路片段置顶 all_contexts structural_snippets[:2] [doc.page_content for doc in semantic_docs] return \n\n---\n\n.join(all_contexts) def knowledge_retriever_node(state: DialogueState) - DialogueState: retriever HybridRetriever() context retriever.retrieve(state[user_input]) return {knowledge_context: context}提示extract_entities()用 spaCy 的en_core_web_sm只抽产品名、型号、日期等高频实体不追求 100% 准确而是提高结构路召回率。实测显示双路召回使政策类问答准确率从 78% 提升至 93%。4.2 人机转人工决策模型不是阈值判断而是多维风险评分网上常见做法是“LLM 置信度 0.5 就转人工”但置信度本身不可靠。我们设计一个 0~100 的transfer_score由三部分加权知识缺口分40%knowledge_context为空或长度 50 字槽位完成分30%required_slots_for_intent(intent)中未填字段数 / 总数对话熵分30%用state[dialogue_history]计算用户话术重复率、否定词频“不”、“不是”、“等等”、情绪词“生气”、“投诉”TF-IDF 加权和。def human_transfer_judge_node(state: DialogueState) - DialogueState: score 0.0 # 知识缺口 if not state.get(knowledge_context) or len(state[knowledge_context]) 50: score 40.0 # 槽位完成 required required_slots_for_intent(state[intent]) filled sum(1 for k in required if k in state[slots] and state[slots][k].strip()) if required: score 30.0 * (1 - filled / len(required)) # 对话熵简化版 history_text .join([msg[content] for msg in state[dialogue_history][-5:]]) repeat_ratio len(re.findall(r\b\w\b, history_text)) / (len(history_text.split()) 1) deny_count len(re.findall(r\b(?:not|no|dont|wont)\b, history_text, re.I)) anger_words [angry, mad, furious, complain, disappointed] anger_score sum(history_text.lower().count(w) for w in anger_words) entropy min(1.0, (repeat_ratio * 0.4 deny_count * 0.3 anger_score * 0.3)) score 30.0 * entropy should_transfer score 65.0 # 阈值可配置 reason [] if score 65.0: if not state.get(knowledge_context): reason.append(知识库未命中) if filled len(required): reason.append(f槽位缺失{, .join(set(required) - set(state[slots].keys()))}) if entropy 0.7: reason.append(对话混乱度高) return { should_transfer_to_human: should_transfer, transfer_reason: ; .join(reason), transfer_score: round(score, 1) }这个模型不保证 100% 正确但它把主观决策变成可解释、可审计、可调优的数值——运营人员看到“转人工因槽位缺失”就知道该优化话术引导看到“因知识库未命中”就知道该补充文档。5. API 服务层与避坑指南把 LangGraph 流水线变成生产可用的 REST 接口LangGraph 本地跑通只是第一步真正上线要面对并发请求、超时熔断、日志追踪、灰度发布。本方案的 API 层不追求“一行命令启动”而是用 FastAPI 封装暴露/chat接口输入{session_id: xxx, message: 我想退订}输出{response: ..., next_state: awaiting_order_id, requires_human: false}。5.1 FastAPI 服务用asyncio.Lock保护会话状态避免并发污染LangGraph 的app.invoke()是同步阻塞的但 FastAPI 要求异步。更关键的是同一session_id的多次请求可能并发到达若不加锁state会被多个线程同时修改。我们用内存级asyncio.Lock按 session_id 隔离from fastapi import FastAPI, HTTPException, Depends from fastapi.responses import JSONResponse import asyncio from typing import Dict, Any app FastAPI() # 全局锁字典key 为 session_id session_locks: Dict[str, asyncio.Lock] {} app.post(/chat) async def chat_endpoint(request: ChatRequest): session_id request.session_id # 获取或创建 session 锁 if session_id not in session_locks: session_locks[session_id] asyncio.Lock() lock session_locks[session_id] async with lock: # 确保同一 session_id 串行执行 try: # 初始化或加载会话状态 state load_state_from_cache(session_id) or initial_state(request.message) # 执行 LangGraph 流水线 result app.invoke(state) # 保存状态到缓存Redis save_state_to_cache(session_id, result) return JSONResponse({ response: result.get(response, ), next_state: result.get(next_state, idle), requires_human: result.get(should_transfer_to_human, False), transfer_reason: result.get(transfer_reason, ) }) except Exception as e: logger.error(fChat error for {session_id}: {e}) raise HTTPException(status_code500, detailService error)注意load_state_from_cache()用 Redis 的HGETALL读取哈希表save_state_to_cache()用HMSET写入session_id作为 key。Redis TTL 设为 24 小时避免内存泄漏。5.2 日志与可观测性不只记录“谁说了什么”更要记录“为什么这么答”客服系统最怕“用户说 A系统答 B但没人知道 B 是怎么来的”。我们在每个 LangGraph 节点里埋点import logging from datetime import datetime logger logging.getLogger(agent_audit) def intent_classifier_node(state: DialogueState) - DialogueState: start_time datetime.now() # ... 分类逻辑 ... end_time datetime.now() logger.info( fINTENT_CLASSIFY | session{state[session_id]} | fuser_input{state[user_input]} | fintent{intent} | fconfidence{confidence:.2f} | fduration_ms{(end_time - start_time).total_seconds()*1000:.0f} ) return {intent: intent}日志格式统一为|分隔便于 ELK 或 Grafana 提取字段。关键字段session_id关联全链路、node_name定位瓶颈、duration_ms性能监控、intent/slots业务审计。线上发现某意图响应慢直接查INTENT_CLASSIFY日志看是模型慢还是 LLM 拒识兜底耗时。5.3 避坑指南那些让客服 Agent 上线即翻车的血泪经验现象 1用户连续发 3 条消息系统只回复最后一条前两条“石沉大海”原因前端未实现消息队列或 API 层未对同一 session_id 请求做排队导致后到请求覆盖了前序状态。解决FastAPI 层加asyncio.Queue每个 session_id 绑定一个队列请求入队worker 协程按序消费。队列长度限制为 5超限则返回429 Too Many Requests。现象 2知识检索返回“根据《消费者权益保护法》第24条”但用户根本看不懂法律条文原因RAG 检索到原文LLM 直接拼接输出未做口语化转译。解决在response_generation_node中强制要求 LLM 将政策原文转化为“您符合 XXX 条件我们可以为您 XXX”的句式并添加示例“比如您购买的手机在 7 天内出现非人为损坏我们可为您免费更换一台新机。”现象 3槽位提取器对“我昨天买的”返回order_date: yesterday但后端需要 ISO 格式日期原因LLM 输出未经过标准化校验。解决在slot_extractor_node后加date_normalizer_node用dateparser库将自然语言日期转为YYYY-MM-DD失败则留空并标记date_parse_failed: true触发人工复核。现象 4LangGraph 流水线跑着跑着内存暴涨最终 OOM原因state[dialogue_history]无限制增长每轮追加 200 字100 轮后达 20KBLLM 输入 token 爆炸。解决在dst_node中截断历史只保留最近 5 轮或总长度 2000 字符并用摘要模型如facebook/bart-base压缩早期历史“用户之前咨询过订单状态已告知物流信息”。现象 5转人工按钮点了没反应坐席端收不到会话原因human_transfer_decision_node只设置should_transfer_to_humanTrue但未调用坐席系统 API。解决在该节点末尾同步调用坐席系统 Webhook如requests.post(https://seat-api/transfer, json...)并设置 5 秒超时失败则降级为“请稍候正在为您转接”。6. 验证与压测用真实会话日志做回归测试用 Locust 模拟万人并发写完代码不等于能上线。本方案的交付物不是 ZIP 包而是可验证的交付清单100 条标注会话日志含意图、槽位、知识引用、转人工标记、压测报告、灰度发布 checklist。没有这些任何“Agent”都是空中楼阁。6.1 回归测试用真实日志跑通全流程拒绝“Hello World”式测试我们不写单元测试而是用生产环境脱敏日志做端到端回归# test_regression.py import pytest from dialogue_state import DialogueState from langgraph_executor import app # 你的 LangGraph app pytest.mark.parametrize(log_entry, load_test_logs(test_logs.jsonl)) def test_full_flow(log_entry): # log_entry 格式{session_id: S123, steps: [{user: 我要退订, expected_intent: subscription_cancellation, ...}]} state initial_state(log_entry[steps][0][user]) for step in log_entry[steps]: # 执行一次 invoke result app.invoke(state) # 断言关键字段 assert result[intent] step[expected_intent], \ fIntent mismatch at step {step[index]}: got {result[intent]}, expected {step[expected_intent]} # 槽位校验 for slot, expected_val in step.get(expected_slots, {}).items(): assert result[slots].get(slot) expected_val, \ fSlot {slot} mismatch: got {result[slots].get(slot)}, expected {expected_val} # 知识上下文非空 assert len(result.get(knowledge_context, )) 50, \ Knowledge context too short state result # 下一轮输入test_logs.jsonl是从线上导出的 100 条会话每条含 3~8 轮交互。CI 流水线每次 PR 都跑这个测试任一 fail 都阻断合并。这是唯一能证明“改了槽位提取器没破坏退款流程”的方式。6.2 压测Locust 模拟 5000 并发关注 P99 延迟与错误率用 Locust 写压测脚本模拟真实用户行为# locustfile.py from locust import HttpUser, task, between import json import random class AgentUser(HttpUser): wait_time between(1, 3) # 用户思考时间 task def chat(self): session_id fsess_{random.randint(1000,9999)} messages [ 你好我要退订会员, 订单号是 ORD-2024-7890, 原因是服务不好 ] for msg in messages: with self.client.post(/chat, json{session_id: session_id, message: msg}, catch_responseTrue) as response: if response.status_code ! 200: response.failure(fHTTP {response.status_code}) else: try: data response.json() if not data.get(response): response.failure(Empty response) except: response.failure(Invalid JSON)启动命令locust -f locustfile.py --host http://localhost:8000 --users 5000 --spawn-rate 100。目标指标P99 延迟 本文还有配套的精品资源点击获取