​RAG过时了？电信级AI应用需要「工作流 RAG」双引擎模式​

运营商的每一次“故障→修复”不仅考验网络，还考验组织的流程和信息流。单靠人力排班和经验规则，难以在高并发与复杂依赖下保持稳定；单靠一次性把全部信息丢给模型，也会因为上下文窗、时序依赖、接口幂等等工程问题崩盘。

解决办法不是更强的模型，而是把「智能判断」变成「可控流程」。把模型的判断当作一个节点，把工单、告警、计费当作受控接口，再用工作流把这些节点按规则串起来——这就是本文的核心视角。接下来你会看到清晰的原理、面向运营商的实战样例，以及可直接运行的工程代码，能立刻验证思路并产生业务价值。 本文先讲清楚原理，再给一套面向电信运营商的可运行代码示例（基于 Agently 的工程化工作流）。

一、用招聘合格员工的比喻说明问题

运营商需要什么样的“员工”来处理用户请求？

• 能快速判断问题类型（断网 / 账单 / 套餐变更 / 新装），并用合适话术先安抚用户；

• 能把复杂流程拆解（比如断网 -> 排查 -> 工单 -> 派单 -> 跟进 -> 反馈）并在每一步做清晰记录；

• 能把人工与自动化结合，遇到高风险或需要现场介入的场景自动升级到人工或 NOC（网络运营中心）。

把“模型 + 代码”作为这个员工的智能中枢，工作流就是 SOP（标准操作流程）和流水线。

二、单次请求的局限性

• 上下文窗口受限：用户历史、设备信息、基站状态、上次工单等信息很长，不可能一次性塞进 prompt。

• 思路不可见但不想暴露：直接让模型写 CoT（思维链）会把内部思考“写给用户看”，这既没必要又不专业。

• 时序依赖重：工单生成后会有异步回调、第三方接口、现场派单、收费结算，这些都需要跨请求管理状态。

三、工作流带来的具体好处

• 节点职责清晰：例如“意图识别”“设备定位”“本地化排障规则”“生成工单”“派单给工程师”“客户回访”。

• 可插入外部系统：在节点里直接调用 OSS/BSS 接口、计费系统、工单系统或告警平台。

• 中间态可审计：每个节点产出的结构化数据（如ticket_id、site_id、confidence）可持久化，便于回溯与统计。

• 可控的自动化：对于低风险且可自动修复的问题（远程重启、配置下发），工作流可自动执行；高风险场景自动转人工。

四、面向电信运营商的工作流实战（Agently 示例代码）

下面是一套面向电信运营商客服/工单场景的工作流示例。场景：用户报障（如家庭宽带断网）或咨询（账单、流量异常、套餐变更）。工作流会完成：意图识别 → 快速安抚 → 设备/用户信息查询 → 线下排障规则判断 → 生成或更新工单 → 派单或提示自助操作 → 最终回复用户。

前提：你已在工程里配置好模型 API（ENV 中的deep_seek_url、deep_seek_api_key、deep_seek_default_model），并已能使用 Agently。如无 Agently，可把业务逻辑移植到等效框架中。

#file:telecom_workflow_agently.pyfromENVimportdeep_seek_url,deep_seek_api_key,deep_seek_default_modelimportAgentlyimporttimeimportjson#创建agent（示例）agent=(Agently.create_agent().set_settings("current_model","OAIClient").set_settings("model.OAIClient.url",deep_seek_url).set_settings("model.OAIClient.auth",{"api_key":deep_seek_api_key}).set_settings("model.OAIClient.options",{"model":deep_seek_default_model}))#假设：有本地函数封装对OSS/BSS/工单系统的简单调用defquery_customer_info(msisdn):#伪代码：实际应调用BSS/CRM接口return{"customer_name":"张先生","address":"上海市浦东区","site_id":"SITE-1001","last_visit":"2025-08-20"}defquery_latest_alarm(site_id):#伪代码：调用告警系统或基站监控#返回最近24小时的关键告警摘要return{"has_recent_alarm":True,"alarm_summary":"OLTlinkflapping"}defcreate_ticket(payload):#伪代码：写入工单系统，返回ticket_idreturnf"TICKET-{int(time.time())}"defassign_to_engineer(ticket_id,skill):#伪代码：派单逻辑return{"assigned":True,"engineer_id":"ENG-237"}#工作流定义workflow=Agently.Workflow()@workflow.chunk()defuser_input(inputs,storage):storage.set("user_input",input("[请输入用户描述或工单ID/MSISDN]:").strip())return@workflow.chunk()defdetect_intent(inputs,storage):#用模型来做意图分类（更精细的规则可以结合本地正则/黑白表）result=(agent.input(storage.get("user_input")).output({"intent""断网|账单|套餐变更|查询工单|其他","判断用户请求属于哪类"),"confidence""float","模型对意图判断的置信度(0-1)")}).start())storage.set("intent",result["intent"])storage.set("intent_confidence",result.get("confidence",0.9))returnresult["intent"]@workflow.chunk()defquick_ack_and_guidance(inputs,storage):#立即给出快速安抚或指引（提升用户体验）intent=storage.get("intent")ifintent=="断网":quick="我们已收到您关于断网的报告，正在为您排查。请先确认路由器电源是否正常并尝试重启。"elifintent=="账单":quick="收到关于账单的咨询，请稍等，我帮您查看最近账单明细。"elifintent=="查询工单":quick="请提供工单号或我们将根据您的号码查询最近工单进展。"else:quick="已收到您的问题，我们会尽快处理。"storage.set("quick_reply",quick)print("[快速回复]:",quick)return@workflow.chunk()defenrich_with_customer_info(inputs,storage):#从输入解析MSISDN或工单号（此处简化）text=storage.get("user_input")msisdn=Noneticket_id=Noneiftext.upper().startswith("TICKET-"):ticket_id=text.strip().upper()else:msisdn=text#假设直接输入手机号storage.set("msisdn",msisdn)storage.set("ticket_id",ticket_id)ifmsisdn:cust=query_customer_info(msisdn)storage.set("customer_info",cust)return@workflow.chunk()defdiagnose_and_route(inputs,storage):intent=storage.get("intent")cust=storage.get("customer_info",{})#结合本地规则与模型建议决定是否立刻生成工单或给出自助指引ifintent=="断网":#查询告警系统site_id=cust.get("site_id")alarm=query_latest_alarm(site_id)ifsite_idelse{"has_recent_alarm":False}storage.set("alarm",alarm)#模型建议（使用模型来补充自然语言诊断与操作建议）model_suggest=agent.input(f"""用户：{storage.get("user_input")}已知信息：{json.dumps(cust)}最近告警：{json.dumps(alarm)}请判断是否需要生成现场工单，或先引导用户做远程排查（如重启ONT/路由器）。输出格式：{{"action":"dispatch|remote_diagnose|ask_more_info","reason":"str","estimated_time_minutes":int}}""").start()#约定模型返回结构化数据（实际需用output/schema强制）#这里假设模型返回JSON字符串try:suggestion=json.loads(model_suggest)exceptException:suggestion={"action":"dispatch","reason":"模型解析失败，默认派单","estimated_time_minutes":60}storage.set("suggestion",suggestion)elifintent=="账单":#直接拉账单数据（伪）storage.set("bill_summary",{"last_amount":98.5,"due_date":"2025-09-05"})returnstorage.get("suggestion",None)@workflow.chunk()defexecute_action(inputs,storage):suggestion=storage.get("suggestion",{})ifsuggestion.get("action")=="dispatch":payload={"msisdn":storage.get("msisdn"),"customer":storage.get("customer_info"),"alarm":storage.get("alarm"),"reason":suggestion.get("reason")}ticket_id=create_ticket(payload)storage.set("ticket_id",ticket_id)assign=assign_to_engineer(ticket_id,skill="OLT/FTTx")storage.set("assignment",assign)storage.set("reply",f"已为您创建工单{ticket_id}，预计处理时间约{suggestion.get('estimated_time_minutes')}分钟，工程师{assign.get('engineer_id')}会跟进。")elifsuggestion.get("action")=="remote_diagnose":#可下发远程命令或给用户自助操作步骤storage.set("reply","请先按以下步骤：1)断电30秒后重启路由器；2)若仍有问题，请告知指示灯状态。")else:storage.set("reply","需要更多信息，请描述您看到的指示灯状态或上次何时能正常使用。")return@workflow.chunk_class()deffinal_reply(inputs,storage):print("[最终回复]:",storage.get("reply"))return(workflow.connect_to("user_input").connect_to("detect_intent").connect_to("quick_ack_and_guidance").connect_to("enrich_with_customer_info").connect_to("diagnose_and_route").connect_to("execute_action").connect_to("final_reply"))if__name__=="__main__":workflow.start()

代码说明（关键点）

1. 快速回复（quick_ack_and_guidance）：在后台进行复杂判断前，先给用户即时反馈，提升体验并减少重复催促。

2. enrich_with_customer_info：把 OSS/BSS/CRM 的真实数据接入工作流，供后续节点用。

3. diagnose_and_route：把“模型推理 + 本地规则（告警、黑白表）”结合起来决策，既利用模型的泛化，又保留工程可控性。

4. execute_action：把最终动作（下发远程命令 / 生成工单 / 派单）封装成幂等的 API 调用，并把ticket_id等关键信息存入storage，便于后续查询。

5. 可扩展点：把create_ticket、assign_to_engineer替换为公司真实工单平台 API，并在节点前后加上 schema 验证与异常重试。

五、落地工程注意事项

1. 幂等性不能忽视：派单、计费等操作必须保证幂等（用业务键如msisdn + alarm_hash防重复）。

2. 告警与工单的去重：同一故障可能触发多条告警，工单系统需要做聚合策略（eg. 同站点 5 分钟内同类告警只产生一个工单）。

3. SLA 驱动的分级：对高价值客户或 SLA 要求高的业务（企业专线）设置不同的工作流分支（优先派单、专员跟进）。

4. 审计与回溯：存储每个节点的输入输出（脱敏），并保留版本号，便于事后追责和模型/规则调整。

5. 灰度策略：先在小范围（某城市、某类故障）跑自动化，观测误判率与 NPS，再逐步放量。

6. 人机协作界面：为人工客服/工程师提供“操作建议 + 证据链”（如模型的诊断理由、相关告警快照），让人工更快决策而不是全部依赖模型。

7. 安全与隐私：手机号、地址、账单金额等敏感信息在日志中掩码；模型返回可能包含敏感推断时应触发人工复核。

六、实战示例（典型对话与工作流走向）

• 用户输入：家里宽带突然断线了，路由器指示灯只有 PON 亮

• detect_intent -> 断网

• quick_reply -> “我们收到断网报告，请先重启设备...”

• enrich -> 拉到 site_id 与最近告警（发现 OLT link flapping）

• diagnose -> 模型建议派单

• execute -> 创建工单、派单给具有 OLT 经验的工程师

• final_reply -> 给用户工单号与预计时长

• 用户输入：我的上月账单异常，多扣了流量

• detect_intent -> 账单

• quick_reply -> “收到账单咨询，正在核实...”

• enrich -> 拉账单摘要

• diagnose -> 若金额异常且小额可自动退款，则执行自动退款流程；否则转人工审核。

七、指标与持续改进

• 关键指标：自动处理率、误判率（误派工单）、平均修复时长（MTTR）、NPS/用户满意度、人工接入率。

• 持续改进：定期把误判样本回流训练或优化 prompt/规则，按工单类型设置微调优先级（高频问题优先）。

八、总结

工作流把“人类的分工、工程化思路和智能能力”结合起来，是运营商把“智能”变成“稳定服务”的关键路径。把模型能力视为“判断与建议”，把核心的写操作（工单、计费、派单）视为受控的接口和节点，你就能既获得自动化效率又保证工程可控性。

当工作流把“判断—决策—执行”拆成一串可观测、可回溯的节点时，智能便从“猜测”变成“能力”：你可以衡量、可以改进、可以用数据证明它带来的收益。对运营商而言，这意味着更少的误派工单、更短的平均修复时间（MTTR）、更高的自动化通过率和更少的客户流失。

把模型当成“建议引擎”，把工作流当成“执行引擎”，你就能把一次又一次的客户投诉，变成可复制、可量化的服务改进——这是把智能变成运营竞争力的实际路径。