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

# file: telecom_workflow_agently.pyfrom ENV import deep_seek_url, deep_seek_api_key, deep_seek_default_modelimport Agentlyimport timeimport json# 创建 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/工单系统的简单调用def query_customer_info(msisdn):    # 伪代码：实际应调用 BSS/CRM 接口    return {"customer_name": "张先生", "address": "上海市浦东区", "site_id": "SITE-1001", "last_visit": "2025-08-20"}def query_latest_alarm(site_id):    # 伪代码：调用告警系统或基站监控    # 返回最近 24 小时的关键告警摘要    return {"has_recent_alarm": True, "alarm_summary": "OLT link flapping"}def create_ticket(payload):    # 伪代码：写入工单系统，返回 ticket_id    return f"TICKET-{int(time.time())}"def assign_to_engineer(ticket_id, skill):    # 伪代码：派单逻辑    return {"assigned": True, "engineer_id": "ENG-237"}# 工作流定义workflow = Agently.Workflow()@workflow.chunk()def user_input(inputs, storage):    storage.set("user_input", input("[请输入用户描述或工单ID/MSISDN]: ").strip())    [email protected]()def detect_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))    return result["intent"]@workflow.chunk()def quick_ack_and_guidance(inputs, storage):    # 立即给出快速安抚或指引（提升用户体验）    intent = storage.get("intent")    if intent == "断网":        quick = "我们已收到您关于断网的报告，正在为您排查。请先确认路由器电源是否正常并尝试重启。"    elif intent == "账单":        quick = "收到关于账单的咨询，请稍等，我帮您查看最近账单明细。"    elif intent == "查询工单":        quick = "请提供工单号或我们将根据您的号码查询最近工单进展。"    else:        quick = "已收到您的问题，我们会尽快处理。"    storage.set("quick_reply", quick)    print("[快速回复]:", quick)    [email protected]()def enrich_with_customer_info(inputs, storage):    # 从输入解析 MSISDN 或工单号（此处简化）    text = storage.get("user_input")    msisdn = None    ticket_id = None    if text.upper().startswith("TICKET-"):        ticket_id = text.strip().upper()    else:        msisdn = text  # 假设直接输入手机号    storage.set("msisdn", msisdn)    storage.set("ticket_id", ticket_id)    if msisdn:        cust = query_customer_info(msisdn)        storage.set("customer_info", cust)    [email protected]()def diagnose_and_route(inputs, storage):    intent = storage.get("intent")    cust = storage.get("customer_info", {})    # 结合本地规则与模型建议决定是否立刻生成工单或给出自助指引    if intent == "断网":        # 查询告警系统        site_id = cust.get("site_id")        alarm = query_latest_alarm(site_id) if site_id else {"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)        except Exception:            suggestion = {"action": "dispatch", "reason": "模型解析失败，默认派单", "estimated_time_minutes": 60}        storage.set("suggestion", suggestion)    elif intent == "账单":        # 直接拉账单数据（伪）        storage.set("bill_summary", {"last_amount": 98.5, "due_date": "2025-09-05"})    return storage.get("suggestion", None)@workflow.chunk()def execute_action(inputs, storage):    suggestion = storage.get("suggestion", {})    if suggestion.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')} 会跟进。")    elif suggestion.get("action") == "remote_diagnose":        # 可下发远程命令或给用户自助操作步骤        storage.set("reply", "请先按以下步骤：1) 断电30秒后重启路由器；2) 若仍有问题，请告知指示灯状态。")    else:        storage.set("reply", "需要更多信息，请描述您看到的指示灯状态或上次何时能正常使用。")    [email protected]_class()def final_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()