1692 lines
81 KiB
Python
1692 lines
81 KiB
Python
"""
|
||
故障诊断工作流 checkpointer 版本
|
||
"""
|
||
from typing import TypedDict, Optional, Dict, Any, List
|
||
from langgraph.graph import StateGraph, START, END
|
||
from langgraph.types import interrupt, Command
|
||
from tools.function_tool import graph_rag_search, fault_graph_rag_search
|
||
from tools.rag_tools import rag_search
|
||
from tools.graph_tools import atlas_retrieval, format_atlas_results, extract_entity_names_from_history, normalize_device_name_by_ship_graph
|
||
from modelsAPI.model_api import OpenaiAPI
|
||
from utils.function_tracker import get_all_callbacks
|
||
from utils.image_utils import extract_image_paths, normalize_markdown_images, get_image_prompt_guidance
|
||
from workflows.workflow_baike import run_baike_workflow
|
||
from workflows.workflow_utils import (
|
||
safe_json_extract, parse_history, normalize_latex_spacing,
|
||
remove_duplicate_content, convert_rag_result, calculate_info_completeness,
|
||
emit_callback_event, build_history_str, build_detail_info,
|
||
append_atlas_section, stream_format_and_postprocess,
|
||
)
|
||
from utils.word_generator import generate_report_word
|
||
from utils.intent_matcher import is_confirmation_intent
|
||
from config import KB_TREE_CONFIG, SERVER_CONFIG
|
||
from prompts import FAULT_DIAGNOSIS_PROMPTS, COMMON_PROMPTS
|
||
from tools.fault_record_db import record_fault_from_state
|
||
import asyncio
|
||
import json
|
||
import re
|
||
import random
|
||
import os
|
||
import time
|
||
|
||
class FaultDiagnosisState(TypedDict):
|
||
"""故障诊断工作流状态"""
|
||
extracted_text: str
|
||
combined_query: str
|
||
history_messages: List[Dict[str, Any]]
|
||
|
||
ship_number: str
|
||
device_name: str
|
||
fault: str
|
||
additional_info: str
|
||
fault_code: str
|
||
fault_time: str
|
||
fault_frequency: str
|
||
tried_measures: str
|
||
running_condition: str
|
||
|
||
rag_search_result: Optional[List[Dict[str, Any]]]
|
||
graph_search_result: Optional[str]
|
||
deep_rag_results: Optional[Dict[str, Any]]
|
||
|
||
response: str
|
||
suggestedReplies: List[Dict[str, Any]]
|
||
actions: List[Dict[str, Any]]
|
||
error_message: str
|
||
|
||
agent_decision: str
|
||
agent_reasoning: str
|
||
tools_to_call: List[str]
|
||
missing_info: List[str]
|
||
info_completeness: float
|
||
|
||
matched_kb_id: str
|
||
matched_kb_name: str
|
||
|
||
iteration_count: int
|
||
max_iterations: int
|
||
scheme_generated: bool
|
||
waiting_for_supplement: bool
|
||
|
||
extracted_info: Dict[str, Any]
|
||
has_rag_result: bool
|
||
has_graph_result: bool
|
||
ask_round: int
|
||
user_confirmed_info: bool
|
||
|
||
need_model_confirm: bool
|
||
user_confirmed_model: bool
|
||
|
||
user_feedback: str
|
||
waiting_report_confirm: bool
|
||
report_confirmed: bool
|
||
|
||
initialized: bool
|
||
current_node: str
|
||
|
||
# 意图分类相关字段
|
||
user_intent: str # 用户初始意图: '维修', '百科'
|
||
|
||
# 后续处理相关字段
|
||
follow_up_intent: str # 用户后续意图: 'not_solved', 'related_question', 'has_error', 'modify_info', 'other'
|
||
user_feedback_for_regenerate: str # 用户用于重新生成方案的反馈
|
||
is_regenerating: bool # 是否正在重新生成方案
|
||
last_generated_scheme: str # 最后一次生成的方案（用于基于反馈修改）
|
||
scheme_type: str # 方案类型：'normal' 普通方案，'deep' 深度方案
|
||
deep_analysis_points: Optional[List[str]] # 深度分析点（用于针对性检索）
|
||
atlas_results: Optional[Dict[str, Any]] # 图册检索结果
|
||
last_combined_query: str # 上一轮生成方案时的用户输入
|
||
|
||
def _preprocess_scheme_for_word(content: str) -> str:
|
||
"""
|
||
预处理方案内容：将相对路径图片URL转为绝对URL
|
||
图片格式解析和公式渲染由 word_generator.py 直接处理
|
||
"""
|
||
if not content:
|
||
return content
|
||
|
||
base_url = SERVER_CONFIG.get("base_url", "").rstrip("/")
|
||
|
||
kb_base_url = KB_TREE_CONFIG.get("url", "").rstrip("/")
|
||
kb_host = ""
|
||
if kb_base_url:
|
||
from urllib.parse import urlparse
|
||
kb_parsed = urlparse(kb_base_url)
|
||
kb_host = f"{kb_parsed.scheme}://{kb_parsed.netloc}"
|
||
|
||
# 知识库图片由知识库服务提供
|
||
if kb_host:
|
||
content = re.sub(
|
||
r'(/api/v1/knowledge/files/[^)\s]+)',
|
||
lambda m: f"{kb_host}{m.group(1)}" if not m.group(1).startswith("http") else m.group(1),
|
||
content
|
||
)
|
||
elif base_url:
|
||
content = re.sub(
|
||
r'(/api/v1/knowledge/files/[^)\s]+)',
|
||
lambda m: f"{base_url}{m.group(1)}" if not m.group(1).startswith("http") else m.group(1),
|
||
content
|
||
)
|
||
|
||
# /picture/ 图片由本应用提供
|
||
if base_url:
|
||
content = re.sub(
|
||
r'(/picture/[^)\s]+)',
|
||
lambda m: f"{base_url}{m.group(1)}" if not m.group(1).startswith("http") else m.group(1),
|
||
content
|
||
)
|
||
|
||
return content
|
||
|
||
|
||
def _generate_report_from_scheme(scheme_content: str, ship_number: str = "", device_name: str = "", fault: str = "") -> List[Dict[str, Any]]:
|
||
"""
|
||
从方案内容直接生成报告并返回下载按钮
|
||
|
||
Args:
|
||
scheme_content: 方案内容（Markdown 格式）
|
||
ship_number: 舷号
|
||
device_name: 设备名称
|
||
fault: 故障现象
|
||
|
||
Returns:
|
||
actions 列表，包含下载按钮（如果生成成功）
|
||
"""
|
||
try:
|
||
# 构建报告标题
|
||
report_title = "维修方案报告"
|
||
if ship_number:
|
||
report_title = f"{ship_number} - {report_title}"
|
||
if device_name:
|
||
report_title = f"{report_title} - {device_name}"
|
||
|
||
# 预处理方案内容：解析图片URL和公式格式
|
||
processed_content = _preprocess_scheme_for_word(scheme_content)
|
||
|
||
# 准备报告内容 - 直接使用方案内容，添加标题信息
|
||
random_letters = ''.join(random.choices('ABCDEFGHIJKLMNOPQRSTUVWXYZ', k=2))
|
||
report_number = time.strftime(f"{random_letters}-%Y-%m-%d-%H%M")
|
||
compile_time = time.strftime("%Y-%m-%d %H:%M")
|
||
|
||
# 构建报告内容
|
||
report_content = f"""## **{report_title}**
|
||
|
||
报告编号：{report_number}
|
||
编制时间：{compile_time}
|
||
|
||
---
|
||
|
||
{processed_content}
|
||
"""
|
||
|
||
# 确保 created_word 目录存在
|
||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||
parent_dir = os.path.dirname(current_dir)
|
||
word_dir = os.path.join(parent_dir, "created_word")
|
||
os.makedirs(word_dir, exist_ok=True)
|
||
|
||
# 生成 Word 文档
|
||
timestamp = time.strftime("%Y%m%d_%H%M%S")
|
||
safe_title = re.sub(r'[\\/:*?"<>|]', '_', report_title)
|
||
file_name = f"{safe_title}_{timestamp}.docx"
|
||
output_path = os.path.join(word_dir, file_name)
|
||
|
||
word_result = generate_report_word(markdown=report_content,title=report_title,output=output_path)
|
||
|
||
# 构建下载按钮
|
||
actions = []
|
||
if word_result and os.path.exists(word_result):
|
||
base_url = SERVER_CONFIG.get("report", "")
|
||
download_url = f"{base_url}/api/download/{file_name}"
|
||
|
||
actions.append({"type": "download","label": f"{report_title}.docx","url": download_url,"fileName": file_name,"content": report_content})
|
||
|
||
print(f"[_generate_report_from_scheme] 报告生成成功: {output_path}")
|
||
|
||
return actions
|
||
|
||
except Exception as e:
|
||
print(f"[_generate_report_from_scheme] 生成报告失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return []
|
||
|
||
FAULT_INFO_WEIGHTS = {"ship_number": 0.20, "device_name": 0.20, "fault": 0.25, "fault_code": 0.08, "fault_time": 0.06, "fault_frequency": 0.06, "tried_measures": 0.06, "running_condition": 0.05, "additional_info": 0.04}
|
||
|
||
def _calculate_info_completeness(ship_number: str = "", device_name: str = "", fault: str = "", fault_code: str = "", fault_time: str = "", fault_frequency: str = "", tried_measures: str = "", running_condition: str = "", additional_info: str = "") -> float:
|
||
values = {"ship_number": ship_number, "device_name": device_name, "fault": fault, "fault_code": fault_code, "fault_time": fault_time, "fault_frequency": fault_frequency, "tried_measures": tried_measures, "running_condition": running_condition, "additional_info": additional_info}
|
||
return calculate_info_completeness(FAULT_INFO_WEIGHTS, values)
|
||
|
||
async def classify_intent(state: FaultDiagnosisState) -> Dict[str, Any]:
|
||
"""
|
||
意图分类节点
|
||
"""
|
||
combined_query = state.get("combined_query", "") or ""
|
||
extracted_info = state.get("extracted_info", {}) or {}
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
|
||
# 如果已经提取过维修信息（有设备名或故障现象），说明已在维修流程中，继续维修流程
|
||
if extracted_info or device_name or fault or ship_number:
|
||
print("[classify_intent] 已在维修流程中，跳过意图分类")
|
||
return {"user_intent": "维修","current_node": "extract_info",}
|
||
|
||
system_prompt = FAULT_DIAGNOSIS_PROMPTS["classify_intent_system"]
|
||
|
||
prompt = COMMON_PROMPTS["classify_intent_user"].format(combined_query=combined_query)
|
||
|
||
try:
|
||
response_text = await OpenaiAPI.open_api_chat_without_thinking(query=prompt,model=None,json_output=True,system_prompt=system_prompt,messages=[])
|
||
|
||
parsed = safe_json_extract(response_text)
|
||
intent = "维修"
|
||
if isinstance(parsed, dict):
|
||
intent = parsed.get("intent", "维修")
|
||
|
||
print(f"[意图分类] 用户意图: {intent}")
|
||
|
||
if intent == "维修":
|
||
return {"user_intent": intent,"scheme_generated": False,"current_node": "extract_info",}
|
||
else:
|
||
return {"user_intent": intent,"current_node": "follow_up_qa",}
|
||
|
||
except Exception as e:
|
||
print(f"[意图分类失败: {str(e)}，默认按维修处理")
|
||
return {"user_intent": "维修","scheme_generated": False,"current_node": "extract_info",}
|
||
|
||
def _route_after_classify(state: FaultDiagnosisState) -> str:
|
||
"""意图分类后的路由"""
|
||
intent = state.get("user_intent", "维修")
|
||
if intent == "百科":
|
||
return "follow_up_qa"
|
||
return "extract_info"
|
||
|
||
async def extract_info(state: FaultDiagnosisState) -> Dict[str, Any]:
|
||
"""
|
||
信息提取节点
|
||
"""
|
||
if state.get("initialized", False):
|
||
print("[extract_info] 已初始化，跳过")
|
||
return {"current_node": "agent_think"}
|
||
|
||
combined_query = state.get("combined_query", "") or ""
|
||
history_messages = state.get("history_messages", []) or []
|
||
|
||
history_str = build_history_str(history_messages, recent_count=4)
|
||
|
||
existing_info = []
|
||
existing_ship_number = state.get("ship_number", "") or ""
|
||
existing_device_name = state.get("device_name", "") or ""
|
||
existing_fault = state.get("fault", "") or ""
|
||
existing_fault_code = state.get("fault_code", "") or ""
|
||
existing_fault_time = state.get("fault_time", "") or ""
|
||
existing_fault_frequency = state.get("fault_frequency", "") or ""
|
||
existing_tried_measures = state.get("tried_measures", "") or ""
|
||
existing_running_condition = state.get("running_condition", "") or ""
|
||
existing_additional_info = state.get("additional_info", "") or ""
|
||
|
||
if existing_ship_number:
|
||
existing_info.append(f"舷号: {existing_ship_number}")
|
||
if existing_device_name:
|
||
existing_info.append(f"设备名称: {existing_device_name}")
|
||
if existing_fault:
|
||
existing_info.append(f"故障现象: {existing_fault}")
|
||
if existing_fault_code:
|
||
existing_info.append(f"故障码: {existing_fault_code}")
|
||
if existing_fault_time:
|
||
existing_info.append(f"故障时间: {existing_fault_time}")
|
||
if existing_fault_frequency:
|
||
existing_info.append(f"故障频率: {existing_fault_frequency}")
|
||
if existing_tried_measures:
|
||
existing_info.append(f"已尝试措施: {existing_tried_measures}")
|
||
if existing_running_condition:
|
||
existing_info.append(f"运行工况: {existing_running_condition}")
|
||
if existing_additional_info:
|
||
existing_info.append(f"其他信息: {existing_additional_info}")
|
||
|
||
existing_info_str = "\n".join(existing_info) if existing_info else "（无已提取信息）"
|
||
|
||
system_prompt = FAULT_DIAGNOSIS_PROMPTS["extract_info_system"]
|
||
|
||
prompt = COMMON_PROMPTS["extract_info_user"].format(
|
||
biz_label="维修",
|
||
existing_info_str=existing_info_str,
|
||
history_str=history_str or '（无历史对话）',
|
||
combined_query=combined_query
|
||
)
|
||
|
||
try:
|
||
response_text = await OpenaiAPI.open_api_chat_without_thinking(query=prompt,model=None,json_output=True,system_prompt=system_prompt,messages=[])
|
||
|
||
parsed = safe_json_extract(response_text)
|
||
if not isinstance(parsed, dict):
|
||
parsed = {}
|
||
|
||
new_ship_number = parsed.get("ship_number", "") or existing_ship_number
|
||
new_device_name = parsed.get("device_name", "") or existing_device_name
|
||
new_fault = parsed.get("fault", "") or existing_fault
|
||
new_fault_code = parsed.get("fault_code", "") or existing_fault_code
|
||
new_fault_time = parsed.get("fault_time", "") or existing_fault_time
|
||
new_fault_frequency = parsed.get("fault_frequency", "") or existing_fault_frequency
|
||
new_tried_measures = parsed.get("tried_measures", "") or existing_tried_measures
|
||
new_running_condition = parsed.get("running_condition", "") or existing_running_condition
|
||
new_additional_info = parsed.get("additional_info", "") or existing_additional_info
|
||
|
||
extracted_info = {"ship_number": new_ship_number,"device_name": new_device_name,"fault": new_fault,"fault_code": new_fault_code,"fault_time": new_fault_time,"fault_frequency": new_fault_frequency,"tried_measures": new_tried_measures,"running_condition": new_running_condition,"additional_info": new_additional_info,}
|
||
|
||
print(f"[信息提取] 舷号={extracted_info['ship_number']}, 设备={extracted_info['device_name']}, 故障={extracted_info['fault']}")
|
||
|
||
return {**extracted_info,"extracted_info": extracted_info,"initialized": True,"current_node": "agent_think",}
|
||
|
||
except Exception as e:
|
||
print(f"信息提取失败: {str(e)}")
|
||
return {"initialized": True,"current_node": "agent_think",}
|
||
|
||
async def agent_think(state: FaultDiagnosisState) -> Dict[str, Any]:
|
||
"""
|
||
Agent 思考节点：基于当前状态决定下一步动作
|
||
"""
|
||
combined_query = state.get("combined_query", "") or ""
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
fault_code = state.get("fault_code", "") or ""
|
||
fault_time = state.get("fault_time", "") or ""
|
||
fault_frequency = state.get("fault_frequency", "") or ""
|
||
tried_measures = state.get("tried_measures", "") or ""
|
||
running_condition = state.get("running_condition", "") or ""
|
||
additional_info = state.get("additional_info", "") or ""
|
||
history_messages = state.get("history_messages", []) or []
|
||
|
||
scheme_generated = state.get("scheme_generated", False)
|
||
has_rag_result = state.get("has_rag_result", False)
|
||
has_graph_result = state.get("has_graph_result", False)
|
||
|
||
rag_search_result = state.get("rag_search_result")
|
||
graph_search_result = state.get("graph_search_result")
|
||
iteration_count = state.get("iteration_count", 0)
|
||
|
||
user_confirmed_info = state.get("user_confirmed_info", False)
|
||
waiting_feedback = state.get("waiting_feedback", False)
|
||
|
||
has_ship = bool(ship_number and ship_number.strip())
|
||
has_device = bool(device_name and device_name.strip())
|
||
has_fault = bool(fault and fault.strip())
|
||
has_rag = (rag_search_result is not None and len(rag_search_result) > 0) or has_rag_result
|
||
has_graph = (graph_search_result is not None and len(graph_search_result) > 100) or has_graph_result
|
||
|
||
info_completeness = _calculate_info_completeness(ship_number=ship_number,device_name=device_name,fault=fault,fault_code=fault_code,fault_time=fault_time,fault_frequency=fault_frequency,tried_measures=tried_measures,running_condition=running_condition,additional_info=additional_info)
|
||
|
||
MIN_COMPLETENESS_FOR_GENERATE = 0.25
|
||
|
||
if scheme_generated:
|
||
_re_diagnose_keywords = ["开始诊断", "信息已足够", "信息已足够，开始诊断", "开始", "诊断"]
|
||
_is_re_diagnose = any(kw in combined_query for kw in _re_diagnose_keywords)
|
||
if _is_re_diagnose:
|
||
print(f"[重新诊断] 检测到重新诊断意图: '{combined_query}'，重置状态直接调用工具生成方案")
|
||
tools_to_call = ["rag_search"]
|
||
if device_name and fault:
|
||
tools_to_call.append("graph_rag_search")
|
||
return {"scheme_generated": False,"rag_search_result": None,"graph_search_result": None,"deep_rag_results": None,"has_rag_result": False,"has_graph_result": False,"is_regenerating": False,"user_feedback_for_regenerate": "","follow_up_intent": "","user_confirmed_info": True,"initialized": True,"iteration_count": 0,"ask_round": 0,"agent_decision": "call_tools","tools_to_call": tools_to_call,"current_node": "call_tools",}
|
||
|
||
history_str = build_history_str(history_messages, recent_count=5, assistant_truncate=300)
|
||
|
||
intent_system_prompt = FAULT_DIAGNOSIS_PROMPTS["agent_think_intent_system"]
|
||
|
||
intent_prompt = FAULT_DIAGNOSIS_PROMPTS["agent_think_intent_user"].format(
|
||
ship_number=ship_number or '未提供',
|
||
device_name=device_name or '未提供',
|
||
fault=fault or '未提供',
|
||
history_str=history_str or '无',
|
||
combined_query=combined_query
|
||
)
|
||
|
||
try:
|
||
intent_response = await OpenaiAPI.open_api_chat_without_thinking(
|
||
query=intent_prompt,
|
||
model=None,
|
||
json_output=True,
|
||
system_prompt=intent_system_prompt,
|
||
messages=[]
|
||
)
|
||
|
||
parsed_intent = safe_json_extract(intent_response)
|
||
intent = "other"
|
||
if isinstance(parsed_intent, dict):
|
||
intent = parsed_intent.get("intent", "other")
|
||
|
||
print(f"[Agent 意图分析] 用户意图: {intent}")
|
||
|
||
# 根据大模型判断的意图来决定下一步
|
||
if intent == "generate_report":
|
||
decision = "generate_report"
|
||
reasoning = "用户请求生成报告"
|
||
print(f"[Agent 决策] decision={decision}, reasoning={reasoning}")
|
||
return {"agent_decision": decision,"agent_reasoning": reasoning,"info_completeness": info_completeness,"iteration_count": iteration_count + 1,"current_node": decision,}
|
||
elif intent == "modify_info":
|
||
# 修改信息，跳转到 confirm_info，重置状态
|
||
decision = "confirm_info"
|
||
reasoning = "用户需要修改或补充信息"
|
||
return {"agent_decision": decision,"agent_reasoning": reasoning,"rag_search_result": None,"graph_search_result": None,"deep_rag_results": None,"has_rag_result": False,"has_graph_result": False,"scheme_generated": False,"is_regenerating": False,"user_feedback_for_regenerate": "","follow_up_intent": "","user_confirmed_info": False,"initialized": False,"info_completeness": info_completeness,"iteration_count": iteration_count + 1,"current_node": decision,}
|
||
elif intent == "has_error":
|
||
decision = "handle_feedback"
|
||
reasoning = "用户反馈方案有错误"
|
||
print(f"[Agent 决策] decision={decision}, reasoning={reasoning}")
|
||
return {"agent_decision": decision,"agent_reasoning": reasoning,"info_completeness": info_completeness,"iteration_count": iteration_count + 1,"current_node": decision,}
|
||
elif intent == "not_solved":
|
||
decision = "analyze_follow_up_intent"
|
||
reasoning = "已生成方案，分析用户后续意图"
|
||
# 设置 follow_up_intent，让 analyze_follow_up_intent 直接知道
|
||
return {"agent_decision": decision,"agent_reasoning": reasoning,"follow_up_intent": "not_solved","info_completeness": info_completeness,"iteration_count": iteration_count + 1,"current_node": decision,}
|
||
elif intent == "related_question":
|
||
decision = "analyze_follow_up_intent"
|
||
reasoning = "已生成方案，分析用户后续意图"
|
||
return {"agent_decision": decision,"agent_reasoning": reasoning,"follow_up_intent": "related_question","info_completeness": info_completeness,"iteration_count": iteration_count + 1,"current_node": decision,}
|
||
else:
|
||
# other 情况，直接跳转到 follow_up_qa，不需要经过 analyze_follow_up_intent
|
||
decision = "follow_up_qa"
|
||
reasoning = "其他意图，直接调用 baike"
|
||
print(f"[Agent 决策] decision={decision}, reasoning={reasoning}")
|
||
return {"agent_decision": decision,"agent_reasoning": reasoning,"info_completeness": info_completeness,"iteration_count": iteration_count + 1,"current_node": decision,}
|
||
except Exception as e:
|
||
print(f"[Agent 意图分析失败: {str(e)}")
|
||
# 大模型判断失败，降级到原来的逻辑
|
||
decision = "analyze_follow_up_intent"
|
||
reasoning = "已生成方案，分析用户后续意图"
|
||
print(f"[Agent 决策] decision={decision}, reasoning={reasoning} (降级)")
|
||
return {"agent_decision": decision,"agent_reasoning": reasoning,"info_completeness": info_completeness,"iteration_count": iteration_count + 1,"current_node": decision,}
|
||
|
||
if not has_device or not has_fault:
|
||
decision = "ask_user"
|
||
missing_info = []
|
||
if not has_device:
|
||
missing_info.append("设备名称")
|
||
if not has_fault:
|
||
missing_info.append("故障现象")
|
||
reasoning = f"缺少基本信息: {', '.join(missing_info)}"
|
||
print(f"[Agent 决策] decision={decision}, reasoning={reasoning}")
|
||
return {"agent_decision": decision,"agent_reasoning": reasoning,"missing_info": missing_info,"info_completeness": info_completeness,"iteration_count": iteration_count + 1,"current_node": decision,}
|
||
|
||
if info_completeness < MIN_COMPLETENESS_FOR_GENERATE:
|
||
decision = "ask_user"
|
||
reasoning = f"信息完整性评分 {info_completeness} 低于最低要求 {MIN_COMPLETENESS_FOR_GENERATE}"
|
||
missing_info = []
|
||
if not ship_number:
|
||
missing_info.append("舷号（可选）")
|
||
if not fault_code:
|
||
missing_info.append("故障码")
|
||
if not fault_time:
|
||
missing_info.append("故障时间")
|
||
if not fault_frequency:
|
||
missing_info.append("故障频率")
|
||
print(f"[Agent 决策] decision={decision}, reasoning={reasoning}")
|
||
return {"agent_decision": decision,"agent_reasoning": reasoning,"missing_info": missing_info or ["更多详细信息"],"info_completeness": info_completeness,"iteration_count": iteration_count + 1,"current_node": decision,}
|
||
|
||
if has_rag or has_graph:
|
||
decision = "generate_response"
|
||
reasoning = "信息充足，已有检索结果，直接生成回复"
|
||
print(f"[Agent 决策] decision={decision}, reasoning={reasoning}")
|
||
return {"agent_decision": decision,"agent_reasoning": reasoning,"info_completeness": info_completeness,"iteration_count": iteration_count + 1,"current_node": decision,}
|
||
|
||
if not user_confirmed_info:
|
||
decision = "confirm_info"
|
||
reasoning = "信息充足，需要用户确认后再检索知识库"
|
||
print(f"[Agent 决策] decision={decision}, reasoning={reasoning}")
|
||
return {"agent_decision": decision,"agent_reasoning": reasoning,"info_completeness": info_completeness,"iteration_count": iteration_count + 1,"current_node": decision,}
|
||
|
||
decision = "call_tools"
|
||
reasoning = "信息充足，用户已确认，开始检索知识库"
|
||
tools_to_call = ["rag_search"]
|
||
if has_device and has_fault:
|
||
tools_to_call.append("graph_rag_search")
|
||
|
||
print(f"[Agent 决策] decision={decision}, reasoning={reasoning}")
|
||
print(f"[Agent 提取] 舷号={ship_number}, 设备={device_name}, 故障={fault}")
|
||
|
||
emit_callback_event(["🤖 Agent 思考中", "🤖 智能分析", "🤖 决策判断"], f"决策: {decision} | 理由: {reasoning[:50]}...")
|
||
|
||
return {"agent_decision": decision,"agent_reasoning": reasoning,"tools_to_call": tools_to_call,"info_completeness": info_completeness,"iteration_count": iteration_count + 1,"current_node": decision,}
|
||
|
||
async def ask_user(state: FaultDiagnosisState) -> Command:
|
||
"""
|
||
询问用户节点：使用 interrupt 暂停等待用户输入
|
||
"""
|
||
missing_info = state.get("missing_info", [])
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
fault_code = state.get("fault_code", "") or ""
|
||
fault_time = state.get("fault_time", "") or ""
|
||
fault_frequency = state.get("fault_frequency", "") or ""
|
||
tried_measures = state.get("tried_measures", "") or ""
|
||
running_condition = state.get("running_condition", "") or ""
|
||
additional_info = state.get("additional_info", "") or ""
|
||
combined_query = state.get("combined_query", "") or ""
|
||
info_completeness = state.get("info_completeness", 0)
|
||
ask_round = state.get("ask_round", 0)
|
||
|
||
MAX_ASK_ROUNDS = 4
|
||
if ask_round >= MAX_ASK_ROUNDS:
|
||
print(f"[ask_user] 已达到最大询问轮次 {MAX_ASK_ROUNDS}，强制进入下一步")
|
||
tools_to_call = ["rag_search"]
|
||
if device_name and fault:
|
||
tools_to_call.append("graph_rag_search")
|
||
return Command(
|
||
update={"ask_round": ask_round + 1,"current_node": "call_tools","tools_to_call": tools_to_call,},
|
||
goto="call_tools"
|
||
)
|
||
|
||
existing_parts = []
|
||
if ship_number:
|
||
existing_parts.append(f"舷号: {ship_number}")
|
||
if device_name:
|
||
existing_parts.append(f"设备名称: {device_name}")
|
||
if fault:
|
||
existing_parts.append(f"故障现象: {fault}")
|
||
if fault_code:
|
||
existing_parts.append(f"故障码: {fault_code}")
|
||
if fault_time:
|
||
existing_parts.append(f"发生时间: {fault_time}")
|
||
if fault_frequency:
|
||
existing_parts.append(f"故障频率: {fault_frequency}")
|
||
if tried_measures:
|
||
existing_parts.append(f"已尝试措施: {tried_measures}")
|
||
if running_condition:
|
||
existing_parts.append(f"运行工况: {running_condition}")
|
||
if additional_info:
|
||
existing_parts.append(f"其他信息: {additional_info}")
|
||
|
||
existing_str = "；".join(existing_parts) if existing_parts else "暂无"
|
||
missing_str = "、".join(missing_info) if missing_info else "相关信息"
|
||
|
||
system_prompt = COMMON_PROMPTS["ask_user_system"].format(biz_label="故障诊断", biz_label_short="诊断")
|
||
|
||
prompt = COMMON_PROMPTS["ask_user_prompt"].format(
|
||
info_completeness=info_completeness,
|
||
existing_str=existing_str,
|
||
missing_str=missing_str,
|
||
combined_query=combined_query
|
||
)
|
||
|
||
response = ""
|
||
try:
|
||
response = await OpenaiAPI.open_api_chat_without_thinking(query=prompt,model=None,system_prompt=system_prompt,messages=[])
|
||
response = response.strip() if response else ""
|
||
except Exception as e:
|
||
print(f"[ask_user] API调用失败: {str(e)}")
|
||
response = ""
|
||
|
||
if not response:
|
||
response = f"**【需要更多信息】**\n\n请补充：{missing_str}\n\n提供这些信息后，我可以更准确地为您诊断故障。"
|
||
|
||
if existing_parts:
|
||
existing_info_section = f"\n**📋 已获取的信息：**\n\n" + "\n".join(
|
||
f"- {part}" for part in existing_parts) + "\n\n---\n"
|
||
response = existing_info_section + response
|
||
|
||
suggested_replies = []
|
||
if "舷号" in missing_info:
|
||
suggested_replies.append({"title": "补充舷号", "content": "舷号："})
|
||
if "设备名称" in missing_info:
|
||
suggested_replies.append({"title": "补充设备名称", "content": "设备名称："})
|
||
if "故障现象" in missing_info:
|
||
suggested_replies.append({"title": "补充故障现象", "content": "故障现象："})
|
||
|
||
user_input = interrupt({"response": response,"suggestedReplies": suggested_replies,"waiting_for": "user_input","missing_info": missing_info,})
|
||
|
||
user_query = user_input.get("query", "")
|
||
|
||
update_dict = {"combined_query": user_query,"current_node": "agent_think","ask_round": ask_round + 1,}
|
||
|
||
if user_query:
|
||
extract_prompt = FAULT_DIAGNOSIS_PROMPTS["ask_user_extract"].format(
|
||
missing_str=missing_str,
|
||
user_query=user_query
|
||
)
|
||
|
||
try:
|
||
extract_response = await OpenaiAPI.open_api_chat_without_thinking(query=extract_prompt,model=None,json_output=True,system_prompt="你是信息提取专家。",messages=[])
|
||
extracted = safe_json_extract(extract_response)
|
||
if isinstance(extracted, dict):
|
||
if extracted.get("ship_number"):
|
||
update_dict["ship_number"] = extracted["ship_number"]
|
||
print(f"[ask_user] 提取到舷号: {extracted['ship_number']}")
|
||
if extracted.get("device_name"):
|
||
update_dict["device_name"] = extracted["device_name"]
|
||
print(f"[ask_user] 提取到设备名称: {extracted['device_name']}")
|
||
if extracted.get("fault"):
|
||
update_dict["fault"] = extracted["fault"]
|
||
print(f"[ask_user] 提取到故障现象: {extracted['fault']}")
|
||
except Exception as e:
|
||
print(f"[ask_user] 信息提取失败: {str(e)}")
|
||
|
||
return Command(
|
||
update=update_dict,
|
||
goto="agent_think"
|
||
)
|
||
|
||
async def confirm_info(state: FaultDiagnosisState) -> Command:
|
||
"""
|
||
确认信息节点：重点询问用户是否需要补充信息
|
||
"""
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
fault_code = state.get("fault_code", "") or ""
|
||
fault_time = state.get("fault_time", "") or ""
|
||
fault_frequency = state.get("fault_frequency", "") or ""
|
||
tried_measures = state.get("tried_measures", "") or ""
|
||
running_condition = state.get("running_condition", "") or ""
|
||
additional_info = state.get("additional_info", "") or ""
|
||
info_completeness = state.get("info_completeness", 0)
|
||
|
||
existing_parts = []
|
||
if ship_number:
|
||
existing_parts.append(f"舷号: {ship_number}")
|
||
if device_name:
|
||
existing_parts.append(f"设备名称: {device_name}")
|
||
if fault:
|
||
existing_parts.append(f"故障现象: {fault}")
|
||
if fault_code:
|
||
existing_parts.append(f"故障码: {fault_code}")
|
||
if fault_time:
|
||
existing_parts.append(f"发生时间: {fault_time}")
|
||
if fault_frequency:
|
||
existing_parts.append(f"故障频率: {fault_frequency}")
|
||
if tried_measures:
|
||
existing_parts.append(f"已尝试措施: {tried_measures}")
|
||
if running_condition:
|
||
existing_parts.append(f"运行工况: {running_condition}")
|
||
if additional_info:
|
||
existing_parts.append(f"其他信息: {additional_info}")
|
||
|
||
existing_str = "\n".join(f"- {part}" for part in existing_parts) if existing_parts else "暂无"
|
||
|
||
additional_prompt = ""
|
||
if not ship_number:
|
||
additional_prompt = "\n\n💡 **提示**：如果您知道舷号，请补充，这将有助于我们更准确地生成报告。"
|
||
|
||
response = f"""**📋 已收集到的信息：**
|
||
|
||
{existing_str}
|
||
|
||
**信息完整性评分：** {info_completeness:.0%}
|
||
|
||
---
|
||
|
||
**💡 请问是否需要补充其他信息？**
|
||
- 如故障码、故障时间、故障频率、已尝试措施、运行工况等
|
||
- 如果信息已足够，请点击"开始诊断"{additional_prompt}
|
||
"""
|
||
|
||
emit_callback_event(["✅ 信息确认", "📋 请确认信息", "🔍 准备检索"], f"等待用户确认或补充信息 (完整性: {info_completeness:.0%})")
|
||
|
||
suggested_replies = [{"title": "开始诊断", "content": "信息已足够，开始诊断"},{"title": "补充信息", "content": "我需要补充："}]
|
||
|
||
user_input = interrupt({"response": response,"suggestedReplies": suggested_replies,"waiting_for": "user_confirm",})
|
||
|
||
user_query = user_input.get("query", "")
|
||
|
||
if await is_confirmation_intent(user_query):
|
||
print(f"[confirm_info] 用户确认信息足够，准备调用工具")
|
||
tools_to_call = ["rag_search"]
|
||
if device_name and fault:
|
||
tools_to_call.append("graph_rag_search")
|
||
return Command(
|
||
update={"user_confirmed_info": True,"current_node": "call_tools","tools_to_call": tools_to_call,},
|
||
goto="call_tools"
|
||
)
|
||
else:
|
||
print(f"[confirm_info] 用户需要补充信息: {user_query}")
|
||
return Command(
|
||
update={"user_confirmed_info": False,"combined_query": user_query,"initialized": False,"current_node": "extract_info",},
|
||
goto="extract_info"
|
||
)
|
||
|
||
async def call_tools(state: FaultDiagnosisState) -> Dict[str, Any]:
|
||
"""调用检索工具"""
|
||
tools_to_call = state.get("tools_to_call", [])
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
additional_info = state.get("additional_info", "") or ""
|
||
|
||
rag_results = state.get("rag_search_result")
|
||
graph_results = state.get("graph_search_result")
|
||
atlas_results = {}
|
||
|
||
matched_kb_name = ""
|
||
matched_kb_id = ""
|
||
mapped_ship_number = ""
|
||
|
||
if device_name:
|
||
normalized_result = await normalize_device_name_by_ship_graph(device_name)
|
||
if normalized_result.get("success"):
|
||
normalized_device_name = normalized_result.get("device_name") or device_name
|
||
if normalized_device_name != device_name:
|
||
print(
|
||
f"[设备标准化] '{device_name}' -> '{normalized_device_name}', "
|
||
f"score={normalized_result.get('score')}"
|
||
)
|
||
device_name = normalized_device_name
|
||
else:
|
||
print(f"[设备标准化] 使用原始设备名称 '{device_name}': {normalized_result.get('error')}")
|
||
|
||
async def _do_rag_search():
|
||
from utils.ship_number_search import search_with_ship_number_strategy
|
||
|
||
base_query = f"设备{device_name}，故障{fault}的维修方案"
|
||
results, kb_name, kb_id, mapped_number = await search_with_ship_number_strategy(
|
||
base_query=base_query,
|
||
ship_number=ship_number,
|
||
top_k=5,
|
||
search_type="fault",
|
||
)
|
||
if not kb_name:
|
||
kb_name = "通用知识库"
|
||
print(f"RAG 检索来源: {kb_name}")
|
||
return results, kb_name, kb_id, mapped_number
|
||
|
||
async def _do_graph_search():
|
||
if not (device_name and fault):
|
||
return ""
|
||
try:
|
||
results = await fault_graph_rag_search.ainvoke(
|
||
{"device_name": device_name, "fault_symptom": fault, "ship_number": ""}
|
||
)
|
||
print(f"图谱检索完成，结果长度: {len(results) if results else 0}")
|
||
return results
|
||
except Exception as e:
|
||
print(f"图谱检索失败: {str(e)}")
|
||
return ""
|
||
|
||
async def _do_atlas_search():
|
||
if not device_name:
|
||
return {}
|
||
try:
|
||
print(f"[图册检索] 使用设备名称: {device_name}")
|
||
result = await atlas_retrieval(node_names=[device_name], top_k=10)
|
||
if result.get("success", False):
|
||
data = result.get("data", {})
|
||
print(f"[图册检索] 图册检索结果: {list(data.keys())}")
|
||
return data
|
||
except Exception as e:
|
||
print(f"[图册检索] 图册检索失败: {str(e)}")
|
||
return {}
|
||
|
||
tasks = []
|
||
task_names = []
|
||
if "rag_search" in tools_to_call and rag_results is None:
|
||
tasks.append(_do_rag_search())
|
||
task_names.append("rag")
|
||
if "graph_rag_search" in tools_to_call and graph_results is None:
|
||
tasks.append(_do_graph_search())
|
||
task_names.append("graph")
|
||
if device_name:
|
||
tasks.append(_do_atlas_search())
|
||
task_names.append("atlas")
|
||
|
||
if tasks:
|
||
task_results = await asyncio.gather(*tasks)
|
||
for task_name, task_result in zip(task_names, task_results):
|
||
if task_name == "rag":
|
||
rag_results, matched_kb_name, matched_kb_id, mapped_ship_number = task_result
|
||
elif task_name == "graph":
|
||
graph_results = task_result
|
||
elif task_name == "atlas":
|
||
atlas_results = task_result
|
||
|
||
if mapped_ship_number and mapped_ship_number != ship_number:
|
||
print(f"[call_tools] 舷号映射: '{ship_number}' -> '{mapped_ship_number}'，更新工作流状态")
|
||
ship_number = mapped_ship_number
|
||
|
||
has_rag = bool(rag_results and len(rag_results) > 0)
|
||
has_graph = bool(graph_results and len(graph_results) > 100)
|
||
|
||
if not has_rag and not has_graph:
|
||
print(f"[call_tools] 未找到任何相关资料，需要用户确认使用模型能力")
|
||
return {"rag_search_result": rag_results if isinstance(rag_results, list) else [],"graph_search_result": graph_results if isinstance(graph_results, str) else "","atlas_results": atlas_results,"matched_kb_id": matched_kb_id,"matched_kb_name": matched_kb_name,"ship_number": ship_number,"device_name": device_name,"has_rag_result": False,"has_graph_result": False,"need_model_confirm": True,"current_node": "model_confirm",}
|
||
|
||
return {"rag_search_result": rag_results,"graph_search_result": graph_results,"atlas_results": atlas_results,"matched_kb_id": matched_kb_id,"matched_kb_name": matched_kb_name,"ship_number": ship_number,"device_name": device_name,"has_rag_result": has_rag,"has_graph_result": has_graph,"need_model_confirm": False,"current_node": "generate_response",}
|
||
|
||
async def model_confirm(state: FaultDiagnosisState) -> Command:
|
||
"""
|
||
模型能力确认节点：当没有找到任何资料时，让用户确认是否使用模型本身能力
|
||
"""
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
|
||
response = f"""**⚠️ 未找到相关资料**
|
||
|
||
抱歉，在知识库中未找到与以下信息相关的资料：
|
||
- 舷号：{ship_number or '未提供'}
|
||
- 设备名称：{device_name or '未提供'}
|
||
- 故障现象：{fault or '未提供'}
|
||
|
||
**💡 您可以选择：**
|
||
- 使用 AI 模型的知识库为您生成一般性建议
|
||
- 提供更详细的信息重新检索
|
||
|
||
请问是否使用 AI 模型为您生成建议？"""
|
||
|
||
emit_callback_event(["🤖 AI 模型确认", "⚠️ 资料未找到", "💡 使用模型能力"], "使用模型自身知识进行回复")
|
||
|
||
suggested_replies = [{"title": "使用 AI 模型", "content": "确认，使用 AI 模型生成建议"},{"title": "补充信息", "content": "我需要补充更多信息："},]
|
||
|
||
user_input = interrupt({"response": response,"suggestedReplies": suggested_replies,"waiting_for": "model_confirm",})
|
||
|
||
user_query = user_input.get("query", "")
|
||
|
||
if "确认" in user_query or "AI" in user_query or "模型" in user_query:
|
||
print(f"[model_confirm] 用户确认使用模型能力")
|
||
return Command(
|
||
update={"user_confirmed_model": True,"combined_query": user_query,"current_node": "generate_response",},
|
||
goto="generate_response"
|
||
)
|
||
else:
|
||
print(f"[model_confirm] 用户选择补充信息: {user_query}")
|
||
return Command(
|
||
update={"user_confirmed_model": False,"combined_query": user_query,"current_node": "agent_think",},
|
||
goto="agent_think"
|
||
)
|
||
|
||
async def generate_response(state: FaultDiagnosisState) -> Dict[str, Any]:
|
||
"""生成最终回复"""
|
||
print("========================生成最终回复========================")
|
||
combined_query = state.get("combined_query", "") or ""
|
||
history_messages = state.get("history_messages", []) or []
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
additional_info = state.get("additional_info", "") or ""
|
||
fault_code = state.get("fault_code", "") or ""
|
||
fault_time = state.get("fault_time", "") or ""
|
||
fault_frequency = state.get("fault_frequency", "") or ""
|
||
tried_measures = state.get("tried_measures", "") or ""
|
||
running_condition = state.get("running_condition", "") or ""
|
||
rag_results = state.get("rag_search_result") or []
|
||
graph_results = state.get("graph_search_result") or ""
|
||
scheme_generated = state.get("scheme_generated", False)
|
||
matched_kb_name = state.get("matched_kb_name", "")
|
||
|
||
print("rag_results", rag_results)
|
||
history_str = build_history_str(history_messages, recent_count=5, assistant_truncate=500)
|
||
|
||
has_rag = rag_results and len(rag_results) > 0
|
||
has_graph = graph_results and len(graph_results) > 100
|
||
|
||
if not has_rag and not has_graph:
|
||
if not ship_number and not device_name and not fault:
|
||
system_prompt = COMMON_PROMPTS["generate_response_friendly_system"].format(biz_label="故障诊断")
|
||
prompt = COMMON_PROMPTS["generate_response_friendly_user"].format(combined_query=combined_query)
|
||
else:
|
||
system_prompt = COMMON_PROMPTS["generate_response_no_rag_system"].format(biz_label="故障诊断")
|
||
prompt = FAULT_DIAGNOSIS_PROMPTS["generate_response_no_rag_user"].format(
|
||
ship_number=ship_number or '未提供',
|
||
device_name=device_name or '未提供',
|
||
fault=fault or '未提供',
|
||
fault_code=fault_code or '未提供',
|
||
fault_time=fault_time or '未提供',
|
||
fault_frequency=fault_frequency or '未提供',
|
||
tried_measures=tried_measures or '未提供',
|
||
running_condition=running_condition or '未提供',
|
||
additional_info=additional_info or '无'
|
||
)
|
||
|
||
try:
|
||
response = await OpenaiAPI.open_api_chat_without_thinking(query=prompt,model=None,system_prompt=system_prompt,messages=[])
|
||
return {"response": response.strip() if response else "抱歉，未能找到相关资料。","suggestedReplies": [{"title": "提供更详细信息", "content": "我来提供更详细的信息"},],"current_node": "end",}
|
||
except Exception as e:
|
||
return {"response": f"抱歉，处理过程中出现错误: {str(e)}","suggestedReplies": [],"current_node": "end",}
|
||
|
||
rag_ctx_parts: List[str] = []
|
||
if isinstance(rag_results, str) and rag_results.strip():
|
||
rag_ctx_parts = [rag_results.strip()]
|
||
elif isinstance(rag_results, list):
|
||
for item in rag_results[:1]:
|
||
if isinstance(item, dict):
|
||
text = (item.get("text") or "").strip()
|
||
if text:
|
||
rag_ctx_parts.append(text)
|
||
elif isinstance(item, str) and item.strip():
|
||
rag_ctx_parts.append(item.strip())
|
||
|
||
if len(graph_results) <= 50:
|
||
graph_results = ""
|
||
|
||
all_source_text = graph_results + "\n" + "\n".join(rag_ctx_parts) if graph_results else "\n".join(rag_ctx_parts)
|
||
original_image_paths = extract_image_paths(all_source_text)
|
||
print("提取到的原始图片路径:", original_image_paths)
|
||
|
||
rag_answer = ""
|
||
rag_type = "none"
|
||
|
||
if graph_results and len(graph_results) > 50:
|
||
rag_answer = graph_results
|
||
rag_type = "graph"
|
||
elif rag_ctx_parts:
|
||
rag_answer = "\n\n".join(rag_ctx_parts[:1])
|
||
rag_type = "rag"
|
||
|
||
if not rag_answer or rag_answer == "NO_RELEVANT_RESULT":
|
||
system_prompt = COMMON_PROMPTS["generate_response_no_rag_simple_system"].format(biz_label="故障诊断")
|
||
prompt = FAULT_DIAGNOSIS_PROMPTS["generate_response_no_rag_simple_user"].format(
|
||
ship_number=ship_number or '未提供',
|
||
device_name=device_name or '未提供',
|
||
fault=fault or '未提供'
|
||
)
|
||
|
||
try:
|
||
response = await OpenaiAPI.open_api_chat_without_thinking(query=prompt,model=None,system_prompt=system_prompt,messages=[])
|
||
return {"response": response.strip() if response else "抱歉，未能找到相关资料。","suggestedReplies": [],"current_node": "end",}
|
||
except Exception as e:
|
||
return {"response": f"抱歉，处理过程中出现错误: {str(e)}","suggestedReplies": [],"current_node": "end",}
|
||
|
||
rag_type = "图谱" if rag_type == "graph" else "文本"
|
||
|
||
emit_callback_event([f"✨当前使用的检索类型为{rag_type}", f"✨本次检索策略为{rag_type}"], f"内容为{rag_answer[:50]}...")
|
||
|
||
detail_info = ""
|
||
if fault_code:
|
||
detail_info += f"- 故障码/报警代码：{fault_code}\n"
|
||
if fault_time:
|
||
detail_info += f"- 故障发生时间：{fault_time}\n"
|
||
if fault_frequency:
|
||
detail_info += f"- 故障频率/持续时间：{fault_frequency}\n"
|
||
if tried_measures:
|
||
detail_info += f"- 已尝试的维修措施：{tried_measures}\n"
|
||
if running_condition:
|
||
detail_info += f"- 设备运行工况：{running_condition}\n"
|
||
if additional_info:
|
||
detail_info += f"- 其他相关信息：{additional_info}\n"
|
||
|
||
is_regenerating = state.get("is_regenerating", False)
|
||
user_feedback_for_regenerate = state.get("user_feedback_for_regenerate", "")
|
||
is_follow_up = scheme_generated and (has_rag or has_graph) and not is_regenerating
|
||
|
||
image_guidance = get_image_prompt_guidance()
|
||
if is_regenerating and user_feedback_for_regenerate:
|
||
prompt = COMMON_PROMPTS["generate_response_regenerating"].format(
|
||
biz_label="维修",
|
||
item_label="故障现象",
|
||
item_name=fault or '未提供',
|
||
item_action="分析故障原因，给出维修方案",
|
||
ship_number=ship_number or '未提供',
|
||
device_name=device_name or '未提供',
|
||
fault=fault or '未提供',
|
||
detail_info=detail_info,
|
||
user_feedback_for_regenerate=user_feedback_for_regenerate,
|
||
rag_answer=rag_answer,
|
||
image_guidance=image_guidance
|
||
)
|
||
elif is_follow_up:
|
||
prompt = COMMON_PROMPTS["generate_response_follow_up"].format(
|
||
biz_label="维修",
|
||
item_label="故障现象",
|
||
item_name=fault or '未提供',
|
||
ship_number=ship_number or '未提供',
|
||
device_name=device_name or '未提供',
|
||
fault=fault or '未提供',
|
||
detail_info=detail_info,
|
||
rag_answer=rag_answer,
|
||
combined_query=combined_query,
|
||
image_guidance=image_guidance
|
||
)
|
||
else:
|
||
prompt = COMMON_PROMPTS["generate_response_normal"].format(
|
||
biz_label="维修",
|
||
item_label="故障现象",
|
||
item_name=fault or '未提供',
|
||
item_action="分析故障原因，给出维修方案",
|
||
ship_number=ship_number or '未提供',
|
||
device_name=device_name or '未提供',
|
||
fault=fault or '未提供',
|
||
detail_info=detail_info,
|
||
rag_answer=rag_answer
|
||
)
|
||
|
||
try:
|
||
# ========== 第一步：专注于生成高质量内容 ==========
|
||
content_system_prompt = COMMON_PROMPTS["generate_response_content_system"].format(biz_label="维修")
|
||
|
||
raw_content = await OpenaiAPI.open_api_chat_without_thinking(model=None,system_prompt=content_system_prompt,messages=[{"role": "user", "content": prompt}],)
|
||
|
||
raw_content = raw_content.strip() if raw_content else "抱歉，无法生成回答。"
|
||
|
||
emit_callback_event(["🤖 诊断分析中", "🤖 故障预检中"], raw_content[:30] + "...")
|
||
|
||
answer = await stream_format_and_postprocess(raw_content, original_image_paths, temperature=0.3)
|
||
|
||
scheme_source = f"本方案依据{rag_type}知识库生成。"
|
||
answer = f"{scheme_source}\n\n{answer}"
|
||
|
||
atlas_results = state.get("atlas_results", {})
|
||
answer = append_atlas_section(answer, atlas_results)
|
||
|
||
if not is_follow_up:
|
||
supplement_prompt = """
|
||
|
||
---
|
||
|
||
**💡 请您确认维修方案是否正确：**
|
||
- 如果方案有误或需要调整，请告诉我具体的错误之处
|
||
- 我会根据您的反馈重新生成更准确的方案
|
||
|
||
请问方案是否有需要修改的地方？"""
|
||
answer += supplement_prompt
|
||
|
||
suggested_replies = [{"title": "方案有误", "content": "方案有误，需要修改："},]
|
||
|
||
# 生成报告并添加下载按钮
|
||
actions = _generate_report_from_scheme(scheme_content=answer,ship_number=ship_number,device_name=device_name,fault=fault)
|
||
|
||
if not is_follow_up:
|
||
try:
|
||
await record_fault_from_state({"ship_number": ship_number,"device_name": device_name,"fault": fault,"last_generated_scheme": answer,})
|
||
except Exception as rec_err:
|
||
print(f"[generate_response] 记录故障信息失败: {rec_err}")
|
||
|
||
return {"response": answer,"actions": actions,"suggestedReplies": suggested_replies,"scheme_generated": True,"waiting_feedback": not is_follow_up,"last_generated_scheme": answer,"scheme_type": "normal","current_node": "end","last_combined_query": combined_query,}
|
||
|
||
except Exception as e:
|
||
return {"response": f"分析生成失败: {str(e)}","actions": [],"suggestedReplies": [],"current_node": "end",}
|
||
|
||
async def analyze_follow_up_intent(state: FaultDiagnosisState) -> Dict[str, Any]:
|
||
"""
|
||
分析用户后续意图节点：直接使用 agent_think 已经判断好的意图
|
||
agent_think 已经用大模型判断并设置了 follow_up_intent
|
||
"""
|
||
# 直接使用已经判断好的意图
|
||
intent = state.get("follow_up_intent", "other")
|
||
|
||
print(f"[意图分析] 使用已判断的意图: {intent}")
|
||
|
||
return {"follow_up_intent": intent,}
|
||
|
||
async def deep_rag_search(state: FaultDiagnosisState) -> Dict[str, Any]:
|
||
"""
|
||
深层次RAG检索节点：分析之前方案并选择深度分析点进行检索
|
||
"""
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
fault_code = state.get("fault_code", "") or ""
|
||
additional_info = state.get("additional_info", "") or ""
|
||
fault_time = state.get("fault_time", "") or ""
|
||
fault_frequency = state.get("fault_frequency", "") or ""
|
||
tried_measures = state.get("tried_measures", "") or ""
|
||
running_condition = state.get("running_condition", "") or ""
|
||
combined_query = state.get("combined_query", "") or ""
|
||
history_messages = state.get("history_messages", []) or []
|
||
last_generated_scheme = state.get("last_generated_scheme", "") or ""
|
||
|
||
deep_results = {}
|
||
deep_analysis_points = []
|
||
|
||
emit_callback_event(["🔍 深度检索中", "🔍 多维度检索", "🔍 精细分析"], "分析问题并选择深度分析点")
|
||
|
||
# ========== 第一步：分析之前方案和用户问题，选择深度分析点 ==========
|
||
if last_generated_scheme or combined_query:
|
||
try:
|
||
history_str = build_history_str(history_messages, recent_count=5, assistant_truncate=300)
|
||
|
||
detail_info = ""
|
||
if fault_code:
|
||
detail_info += f"- 故障码/报警代码：{fault_code}\n"
|
||
if fault_time:
|
||
detail_info += f"- 故障发生时间：{fault_time}\n"
|
||
if fault_frequency:
|
||
detail_info += f"- 故障频率/持续时间：{fault_frequency}\n"
|
||
if tried_measures:
|
||
detail_info += f"- 已尝试的维修措施：{tried_measures}\n"
|
||
if running_condition:
|
||
detail_info += f"- 设备运行工况：{running_condition}\n"
|
||
if additional_info:
|
||
detail_info += f"- 其他相关信息：{additional_info}\n"
|
||
|
||
analysis_system_prompt = COMMON_PROMPTS["deep_rag_analysis_system"].format(biz_label="维修")
|
||
|
||
analysis_prompt = COMMON_PROMPTS["deep_rag_analysis_user"].format(
|
||
item_label="故障现象",
|
||
item_name=fault or '未提供',
|
||
ship_number=ship_number or '未提供',
|
||
device_name=device_name or '未提供',
|
||
fault=fault or '未提供',
|
||
detail_info=detail_info,
|
||
history_str=history_str or '（无历史对话）',
|
||
last_generated_scheme=last_generated_scheme or '（无之前的方案）',
|
||
combined_query=combined_query
|
||
)
|
||
|
||
analysis_response = await OpenaiAPI.open_api_chat_without_thinking(query=analysis_prompt,model=None,json_output=True,system_prompt=analysis_system_prompt,messages=[])
|
||
print("深度分析点", analysis_response)
|
||
parsed_analysis = safe_json_extract(analysis_response)
|
||
deep_analysis_points = parsed_analysis
|
||
print(f"[深度RAG] 选择的分析点: {deep_analysis_points}")
|
||
except Exception as e:
|
||
print(f"[深度RAG] 分析失败: {str(e)}")
|
||
|
||
# ========== 第二步：对选中的分析点进行检索 ==========
|
||
for idx, point in enumerate(deep_analysis_points):
|
||
try:
|
||
print(point)
|
||
# 结合设备名和故障名进行检索，提高准确性
|
||
search_query = point
|
||
if device_name and device_name not in point:
|
||
search_query = f"{device_name} {search_query}"
|
||
if fault and fault not in search_query:
|
||
search_query = f"{search_query} {fault}"
|
||
|
||
rag_result = await rag_search(search_query, top_k=3)
|
||
if rag_result.get("success", False):
|
||
deep_results[f"analysis_point_{idx}"] = convert_rag_result(rag_result)
|
||
else:
|
||
deep_results[f"analysis_point_{idx}"] = []
|
||
except Exception as e:
|
||
print(f"[深度RAG] 分析点 {point} 检索失败: {str(e)}")
|
||
deep_results[f"analysis_point_{idx}"] = []
|
||
|
||
# ========== 第三步：图册检索 ==========
|
||
atlas_results = {}
|
||
try:
|
||
# 从历史记录和当前信息中抽取实体名称
|
||
entity_names = await extract_entity_names_from_history(history_messages=history_messages,device_name=device_name,fault=fault,last_generated_scheme=last_generated_scheme)
|
||
|
||
if entity_names:
|
||
print(f"[深度RAG] 图册检索实体: {entity_names}")
|
||
atlas_result = await atlas_retrieval(node_names=entity_names, top_k=10)
|
||
if atlas_result.get("success", False):
|
||
atlas_results = atlas_result.get("data", {})
|
||
print(f"[深度RAG] 图册检索结果: {list(atlas_results.keys())}")
|
||
except Exception as e:
|
||
print(f"[深度RAG] 图册检索失败: {str(e)}")
|
||
|
||
return {"deep_rag_results": deep_results,"deep_analysis_points": deep_analysis_points,"atlas_results": atlas_results,"current_node": "generate_deep_response",}
|
||
|
||
async def generate_deep_response(state: FaultDiagnosisState) -> Dict[str, Any]:
|
||
"""
|
||
生成深度响应节点：基于深度RAG结果进行深度润色和推理
|
||
"""
|
||
combined_query = state.get("combined_query", "") or ""
|
||
history_messages = state.get("history_messages", []) or []
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
fault_code = state.get("fault_code", "") or ""
|
||
fault_time = state.get("fault_time", "") or ""
|
||
fault_frequency = state.get("fault_frequency", "") or ""
|
||
tried_measures = state.get("tried_measures", "") or ""
|
||
running_condition = state.get("running_condition", "") or ""
|
||
additional_info = state.get("additional_info", "") or ""
|
||
|
||
deep_rag_results = state.get("deep_rag_results", {})
|
||
deep_analysis_points = state.get("deep_analysis_points", [])
|
||
atlas_results = state.get("atlas_results", {})
|
||
graph_results = state.get("graph_search_result", "") or ""
|
||
original_rag_results = state.get("rag_search_result", [])
|
||
|
||
# 整理深度分析点资料
|
||
analysis_points_text = ""
|
||
if deep_analysis_points:
|
||
for idx, point in enumerate(deep_analysis_points):
|
||
key = f"analysis_point_{idx}"
|
||
results = deep_rag_results.get(key, [])
|
||
if results:
|
||
analysis_points_text += f"\n【分析点：{point}】\n"
|
||
for i, item in enumerate(results[:2], 1):
|
||
if isinstance(item, dict):
|
||
text = item.get("text", "")
|
||
if text:
|
||
analysis_points_text += f"[{i}] {text}\n"
|
||
|
||
# 整理原始RAG结果
|
||
original_rag_text = ""
|
||
if original_rag_results:
|
||
original_rag_text = "\n【原始检索资料】\n"
|
||
for i, item in enumerate(original_rag_results[:3], 1):
|
||
if isinstance(item, dict):
|
||
text = item.get("text", "")
|
||
if text:
|
||
original_rag_text += f"[{i}] {text}\n"
|
||
|
||
# 整理图册资料
|
||
atlas_text = format_atlas_results(atlas_results)
|
||
|
||
# 提取所有资料中的图片路径
|
||
all_source_text = analysis_points_text + original_rag_text + atlas_text + graph_results
|
||
original_image_paths = extract_image_paths(all_source_text)
|
||
|
||
detail_info = ""
|
||
if fault_code:
|
||
detail_info += f"- 故障码/报警代码：{fault_code}\n"
|
||
if fault_time:
|
||
detail_info += f"- 故障发生时间：{fault_time}\n"
|
||
if fault_frequency:
|
||
detail_info += f"- 故障频率/持续时间：{fault_frequency}\n"
|
||
if tried_measures:
|
||
detail_info += f"- 已尝试的维修措施：{tried_measures}\n"
|
||
if running_condition:
|
||
detail_info += f"- 设备运行工况：{running_condition}\n"
|
||
if additional_info:
|
||
detail_info += f"- 其他相关信息：{additional_info}\n"
|
||
|
||
# 构建深度分析点的提示
|
||
analysis_points_prompt = ""
|
||
if deep_analysis_points:
|
||
analysis_points_prompt = f"\n【本次深度分析重点】\n" + "\n".join([f"- {p}" for p in deep_analysis_points]) + "\n"
|
||
|
||
prompt = COMMON_PROMPTS["generate_deep_response"].format(
|
||
biz_label="维修",
|
||
item_label="故障现象",
|
||
item_name=fault or '未提供',
|
||
ship_number=ship_number or '未提供',
|
||
device_name=device_name or '未提供',
|
||
fault=fault or '未提供',
|
||
detail_info=detail_info,
|
||
combined_query=combined_query,
|
||
analysis_points_prompt=analysis_points_prompt,
|
||
analysis_points_text=analysis_points_text,
|
||
original_rag_text=original_rag_text,
|
||
graph_results=graph_results if len(graph_results) > 50 else '(无)'
|
||
)
|
||
|
||
try:
|
||
# ========== 第一步：专注于生成高质量内容 ==========
|
||
content_system_prompt = COMMON_PROMPTS["generate_deep_response_content_system"].format(biz_label="维修")
|
||
|
||
raw_content = await OpenaiAPI.open_api_chat_without_thinking(model=None,system_prompt=content_system_prompt,messages=[{"role": "user", "content": prompt}],)
|
||
|
||
raw_content = raw_content.strip() if raw_content else "抱歉，无法生成深度分析。"
|
||
|
||
emit_callback_event(["🤖 诊断分析中", "🤖 故障预检中"], raw_content[:30] + "...")
|
||
|
||
answer = await stream_format_and_postprocess(raw_content, original_image_paths, temperature=0.1)
|
||
|
||
answer = append_atlas_section(answer, atlas_results)
|
||
|
||
suggested_replies = [{"title": "方案有误", "content": "方案还是有问题："},]
|
||
|
||
actions = _generate_report_from_scheme(scheme_content=answer,ship_number=ship_number,device_name=device_name,fault=fault)
|
||
|
||
try:
|
||
await record_fault_from_state({"ship_number": ship_number,"device_name": device_name,"fault": fault,"last_generated_scheme": answer,})
|
||
except Exception as rec_err:
|
||
print(f"[generate_deep_response] 记录故障信息失败: {rec_err}")
|
||
|
||
return {"response": answer,"actions": actions,"suggestedReplies": suggested_replies,"scheme_generated": True,"last_generated_scheme": answer,"scheme_type": "deep","current_node": "end","last_combined_query": combined_query,}
|
||
|
||
except Exception as e:
|
||
return {"response": f"深度分析生成失败: {str(e)}","actions": [],"suggestedReplies": [],"current_node": "end",}
|
||
|
||
async def regenerate_scheme_from_feedback(state: FaultDiagnosisState) -> Dict[str, Any]:
|
||
"""
|
||
基于用户反馈直接修改方案节点：使用之前保存的方案和用户反馈进行修改
|
||
"""
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
additional_info = state.get("additional_info", "") or ""
|
||
fault_code = state.get("fault_code", "") or ""
|
||
fault_time = state.get("fault_time", "") or ""
|
||
fault_frequency = state.get("fault_frequency", "") or ""
|
||
tried_measures = state.get("tried_measures", "") or ""
|
||
running_condition = state.get("running_condition", "") or ""
|
||
|
||
user_feedback = state.get("user_feedback_for_regenerate", "") or ""
|
||
last_scheme = state.get("last_generated_scheme", "") or ""
|
||
scheme_type = state.get("scheme_type", "normal")
|
||
|
||
if not last_scheme:
|
||
# 如果没有保存的方案，回退到重新检索
|
||
print("[regenerate_scheme_from_feedback] 没有保存的方案，回退到重新检索")
|
||
return {"is_regenerating": True,"rag_search_result": None,"graph_search_result": None,"has_rag_result": False,"has_graph_result": False,"scheme_generated": False,"current_node": "call_tools","tools_to_call": ["rag_search", "graph_rag_search"],}
|
||
|
||
detail_info = ""
|
||
if fault_code:
|
||
detail_info += f"- 故障码/报警代码：{fault_code}\n"
|
||
if fault_time:
|
||
detail_info += f"- 故障发生时间：{fault_time}\n"
|
||
if fault_frequency:
|
||
detail_info += f"- 故障频率/持续时间：{fault_frequency}\n"
|
||
if tried_measures:
|
||
detail_info += f"- 已尝试的维修措施：{tried_measures}\n"
|
||
if running_condition:
|
||
detail_info += f"- 设备运行工况：{running_condition}\n"
|
||
if additional_info:
|
||
detail_info += f"- 其他相关信息：{additional_info}\n"
|
||
|
||
emit_callback_event(["✏️ 修改方案中", "🔧 调整方案", "📝 优化方案"], f"根据反馈修改方案: {user_feedback[:30]}...")
|
||
|
||
image_guidance = get_image_prompt_guidance()
|
||
|
||
prompt = COMMON_PROMPTS["regenerate_scheme_from_feedback"].format(
|
||
biz_label="维修",
|
||
item_label="故障现象",
|
||
item_name=fault or '未提供',
|
||
item_action="分析故障原因，给出维修方案",
|
||
ship_number=ship_number or '未提供',
|
||
device_name=device_name or '未提供',
|
||
fault=fault or '未提供',
|
||
detail_info=detail_info,
|
||
last_scheme=last_scheme,
|
||
user_feedback=user_feedback,
|
||
image_guidance=image_guidance
|
||
)
|
||
|
||
try:
|
||
# ========== 第一步：专注于生成高质量内容 ==========
|
||
content_system_prompt = COMMON_PROMPTS["regenerate_scheme_from_feedback_system"].format(biz_label="维修")
|
||
|
||
# 构建第一步的内容提示词（简化格式要求）
|
||
content_prompt = prompt.replace("【输出要求】", "【输出要求】\n- 请专注于内容的准确性、完整性和逻辑性，不要担心格式的美观性\n- 可以适当使用简单的段落分隔，但不要使用复杂的格式")
|
||
|
||
raw_content = await OpenaiAPI.open_api_chat_without_thinking(model=None,system_prompt=content_system_prompt,messages=[{"role": "user", "content": content_prompt}],)
|
||
|
||
raw_content = raw_content.strip() if raw_content else "抱歉，无法根据反馈修改方案。"
|
||
|
||
original_image_paths = extract_image_paths(last_scheme)
|
||
answer = await stream_format_and_postprocess(raw_content, original_image_paths, temperature=0.3, content_label="维修方案")
|
||
|
||
suggested_replies = [{"title": "方案有误", "content": "方案还是有问题："},]
|
||
|
||
actions = _generate_report_from_scheme(scheme_content=answer,ship_number=ship_number,device_name=device_name,fault=fault)
|
||
|
||
try:
|
||
await record_fault_from_state({"ship_number": ship_number,"device_name": device_name,"fault": fault,"last_generated_scheme": answer,})
|
||
except Exception as rec_err:
|
||
print(f"[regenerate_scheme_from_feedback] 记录故障信息失败: {rec_err}")
|
||
|
||
return {"response": answer,"actions": actions,"suggestedReplies": suggested_replies,"scheme_generated": True,"last_generated_scheme": answer,"scheme_type": scheme_type,"current_node": "end","last_combined_query": combined_query,}
|
||
|
||
except Exception as e:
|
||
print(f"[regenerate_scheme_from_feedback] 修改方案失败: {str(e)}")
|
||
# 如果修改失败，回退到重新检索
|
||
return {"is_regenerating": True,"rag_search_result": None,"graph_search_result": None,"has_rag_result": False,"has_graph_result": False,"scheme_generated": False,"current_node": "call_tools","tools_to_call": ["rag_search", "graph_rag_search"],}
|
||
|
||
async def follow_up_qa(state: FaultDiagnosisState) -> Dict[str, Any]:
|
||
"""
|
||
后续问答节点：调用 baike 智能体处理用户的后续问题
|
||
"""
|
||
combined_query = state.get("combined_query", "") or ""
|
||
history_messages = state.get("history_messages", []) or []
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
|
||
history_message_json = json.dumps(history_messages, ensure_ascii=False) if history_messages else ""
|
||
|
||
context_prefix = ""
|
||
if ship_number or device_name or fault:
|
||
context_parts = []
|
||
if ship_number:
|
||
context_parts.append(f"舷号: {ship_number}")
|
||
if device_name:
|
||
context_parts.append(f"设备名称: {device_name}")
|
||
if fault:
|
||
context_parts.append(f"故障现象: {fault}")
|
||
context_prefix = f"【当前故障诊断上下文】\n{'; '.join(context_parts)}\n\n"
|
||
|
||
enhanced_query = f"{context_prefix}用户后续问题: {combined_query}"
|
||
|
||
emit_callback_event(["💬 后续问答", "💬 执行指导", "💬 问题解答"], f"调用 百科 智能体: {combined_query[:30]}...")
|
||
|
||
try:
|
||
baike_result = await run_baike_workflow(extracted_text=enhanced_query,combined_query=enhanced_query,history_message=history_message_json,route_flag="baike")
|
||
|
||
response = baike_result.get("response", "")
|
||
suggested_replies = baike_result.get("suggestedReplies", [])
|
||
|
||
return {"response": response,"suggestedReplies": suggested_replies,"actions": [],"scheme_generated": True,"current_node": "end",}
|
||
|
||
except Exception as e:
|
||
return {"response": f"处理您的问题时出现错误: {str(e)}","suggestedReplies": [],"actions": [],"current_node": "end",}
|
||
|
||
async def generate_report(state: FaultDiagnosisState) -> Dict[str, Any]:
|
||
"""报告生成节点"""
|
||
from workflows.workflow_other_report import run_other_report_workflow
|
||
|
||
combined_query = state.get("combined_query", "") or ""
|
||
history_message = state.get("history_messages", [])
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
|
||
history_message_json = json.dumps(history_message, ensure_ascii=False) if history_message else ""
|
||
|
||
try:
|
||
report_result = await run_other_report_workflow(extracted_text=combined_query,combined_query=combined_query,history_message=history_message_json,route_flag="report")
|
||
|
||
return {"response": report_result.get("response", ""),"actions": report_result.get("actions", []),"suggestedReplies": report_result.get("suggestedReplies", []),"scheme_generated": True,"current_node": "end",}
|
||
except Exception as e:
|
||
return {"response": f"报告生成失败: {str(e)}","actions": [],"suggestedReplies": [],"current_node": "end",}
|
||
|
||
async def handle_feedback(state: FaultDiagnosisState) -> Command:
|
||
"""
|
||
处理用户反馈节点：接收用户对方案的反馈，提供上报和重新生成方案按钮
|
||
"""
|
||
combined_query = state.get("combined_query", "") or ""
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
|
||
response = f"""**📝 已收到您的反馈**
|
||
|
||
感谢您对维修方案的建议！您的反馈已记录：
|
||
|
||
{combined_query}
|
||
|
||
---
|
||
|
||
**📋 反馈信息摘要：**
|
||
- 舷号：{ship_number or '未提供'}
|
||
- 设备名称：{device_name or '未提供'}
|
||
- 故障现象：{fault or '未提供'}
|
||
- 反馈内容：{combined_query[:100]}{'...' if len(combined_query) > 100 else ''}
|
||
|
||
---
|
||
|
||
**💡 您可以选择：**
|
||
- 点击「重新生成方案」根据您的反馈重新生成方案
|
||
- 点击「上报」将反馈提交给相关部门
|
||
- 继续对话获取更多帮助"""
|
||
|
||
actions = [
|
||
{"type": "content","label": "重新生成方案","text": "重新生成方案"},
|
||
{"type": "content","label": "上报","text": "上报"}
|
||
]
|
||
|
||
suggested_replies = []
|
||
|
||
emit_callback_event(["📝 反馈处理", "📋 方案反馈", "✅ 反馈确认"], f"处理用户反馈: {combined_query[:30]}...")
|
||
|
||
user_input = interrupt({"response": response,"actions": actions,"suggestedReplies": suggested_replies,"waiting_for": "feedback_confirm",})
|
||
|
||
user_text = user_input.get("query", "") or ""
|
||
|
||
if user_text == "重新生成方案":
|
||
print(f"[handle_feedback] 用户选择重新生成方案")
|
||
return Command(
|
||
update={"user_feedback_for_regenerate": combined_query,"current_node": "regenerate_scheme_from_feedback",},
|
||
goto="regenerate_scheme_from_feedback"
|
||
)
|
||
elif user_text == "上报":
|
||
print(f"[handle_feedback] 用户选择上报反馈")
|
||
return Command(
|
||
update={"user_feedback": combined_query,"waiting_report_confirm": True,"current_node": "confirm_report",},
|
||
goto="confirm_report"
|
||
)
|
||
else:
|
||
print(f"[handle_feedback] 用户继续对话: {user_text}")
|
||
return Command(
|
||
update={"user_feedback": combined_query,"combined_query": user_text,"current_node": "agent_think",},
|
||
goto="agent_think"
|
||
)
|
||
|
||
async def confirm_report(state: FaultDiagnosisState) -> Dict[str, Any]:
|
||
"""
|
||
上报确认节点：用户点击上报后确认完成
|
||
"""
|
||
ship_number = state.get("ship_number", "") or ""
|
||
device_name = state.get("device_name", "") or ""
|
||
fault = state.get("fault", "") or ""
|
||
user_feedback = state.get("user_feedback", "") or ""
|
||
|
||
response = f"""**✅ 上报成功**
|
||
|
||
您的反馈已成功上报！
|
||
|
||
---
|
||
|
||
**📋 上报信息：**
|
||
- 舷号：{ship_number or '未提供'}
|
||
- 设备名称：{device_name or '未提供'}
|
||
- 故障现象：{fault or '未提供'}
|
||
- 反馈内容：{user_feedback[:100]}{'...' if len(user_feedback) > 100 else ''}
|
||
|
||
---
|
||
|
||
**💡 感谢您的反馈，相关部门将及时处理。**"""
|
||
|
||
suggested_replies = [{"title": "生成维修报告", "content": "请帮我生成维修报告"},]
|
||
|
||
emit_callback_event(["✅ 上报成功", "📋 反馈已提交", "🎉 完成"], "反馈上报成功")
|
||
|
||
return {"response": response,"actions": [],"suggestedReplies": suggested_replies,"report_confirmed": True,"waiting_report_confirm": False,"current_node": "end",}
|
||
|
||
def _route_after_think(state: FaultDiagnosisState) -> str:
|
||
"""Agent 思考后的路由"""
|
||
decision = state.get("agent_decision", "ask_user")
|
||
|
||
return decision
|
||
|
||
def _route_after_intent(state: FaultDiagnosisState) -> str:
|
||
"""意图分析后的路由"""
|
||
intent = state.get("follow_up_intent", "other")
|
||
|
||
if intent == "not_solved":
|
||
return "deep_rag_search"
|
||
elif intent == "related_question":
|
||
return "follow_up_qa"
|
||
elif intent == "has_error":
|
||
return "handle_feedback"
|
||
elif intent == "modify_info":
|
||
# 修改信息，重置相关状态，重新开始信息提取
|
||
return "extract_info"
|
||
else:
|
||
return "follow_up_qa"
|
||
|
||
def _route_after_tools(state: FaultDiagnosisState) -> str:
|
||
"""检索工具调用后的路由"""
|
||
need_model_confirm = state.get("need_model_confirm", False)
|
||
if need_model_confirm:
|
||
return "model_confirm"
|
||
return "generate_response"
|
||
|
||
def create_fault_diagnosis_workflow(checkpointer=None):
|
||
"""
|
||
创建故障诊断工作流（基于 Checkpointer 的状态持久化版本）
|
||
"""
|
||
workflow = StateGraph(FaultDiagnosisState)
|
||
|
||
workflow.add_node("classify_intent", classify_intent)
|
||
workflow.add_node("extract_info", extract_info)
|
||
workflow.add_node("agent_think", agent_think)
|
||
workflow.add_node("ask_user", ask_user)
|
||
workflow.add_node("confirm_info", confirm_info)
|
||
workflow.add_node("call_tools", call_tools)
|
||
workflow.add_node("generate_response", generate_response)
|
||
workflow.add_node("follow_up_qa", follow_up_qa)
|
||
workflow.add_node("model_confirm", model_confirm)
|
||
workflow.add_node("generate_report", generate_report)
|
||
workflow.add_node("handle_feedback", handle_feedback)
|
||
workflow.add_node("confirm_report", confirm_report)
|
||
workflow.add_node("analyze_follow_up_intent", analyze_follow_up_intent)
|
||
workflow.add_node("deep_rag_search", deep_rag_search)
|
||
workflow.add_node("generate_deep_response", generate_deep_response)
|
||
workflow.add_node("regenerate_scheme_from_feedback", regenerate_scheme_from_feedback)
|
||
|
||
workflow.add_edge(START, "classify_intent")
|
||
workflow.add_conditional_edges("classify_intent", _route_after_classify,
|
||
{"extract_info": "extract_info","follow_up_qa": "follow_up_qa",})
|
||
workflow.add_edge("extract_info", "agent_think")
|
||
workflow.add_conditional_edges("agent_think", _route_after_think,
|
||
{"ask_user": "ask_user","confirm_info": "confirm_info","call_tools": "call_tools","generate_response": "generate_response","follow_up": "follow_up_qa","follow_up_qa": "follow_up_qa","generate_report": "generate_report","handle_feedback": "handle_feedback","extract_info": "extract_info","analyze_follow_up_intent": "analyze_follow_up_intent",})
|
||
workflow.add_conditional_edges("call_tools", _route_after_tools, {"model_confirm": "model_confirm","generate_response": "generate_response",})
|
||
workflow.add_conditional_edges("analyze_follow_up_intent", _route_after_intent, {"deep_rag_search": "deep_rag_search","follow_up_qa": "follow_up_qa","handle_feedback": "handle_feedback","extract_info": "extract_info",})
|
||
workflow.add_edge("deep_rag_search", "generate_deep_response")
|
||
workflow.add_edge("generate_deep_response", END)
|
||
workflow.add_edge("generate_response", END)
|
||
workflow.add_edge("follow_up_qa", END)
|
||
workflow.add_edge("generate_report", END)
|
||
workflow.add_edge("regenerate_scheme_from_feedback", END)
|
||
workflow.add_edge("handle_feedback", END)
|
||
workflow.add_edge("confirm_report", END)
|
||
|
||
return workflow.compile(checkpointer=checkpointer)
|
||
|
||
async def run_fault_diagnosis_workflow(extracted_text: str,combined_query: str,history_message: str = "",route_flag: str = "",previous_state_json: str = "",chat_id: str = "",message_id: str = "",checkpointer=None,user_response: Dict[str, Any] = None) -> Dict[str, Any]:
|
||
"""
|
||
执行故障诊断工作流
|
||
|
||
Args:
|
||
extracted_text: 文本描述
|
||
combined_query: 组合查询
|
||
history_message: 历史消息（JSON格式）
|
||
route_flag: 路由标识
|
||
previous_state_json: 兼容参数（已废弃）
|
||
chat_id: 聊天会话 ID
|
||
message_id: 消息 ID
|
||
checkpointer: LangGraph checkpointer 实例
|
||
user_response: 用户恢复执行的响应数据
|
||
|
||
Returns:
|
||
工作流执行结果
|
||
"""
|
||
thread_id = chat_id if chat_id else None
|
||
|
||
config = {"configurable": {"thread_id": thread_id}} if thread_id else {}
|
||
|
||
print(f"[DEBUG] run_fault_diagnosis_workflow: thread_id={thread_id}, checkpointer={checkpointer is not None}")
|
||
|
||
if checkpointer is None:
|
||
from checkpointer_config import checkpointer_manager
|
||
checkpointer = await checkpointer_manager.get_async_checkpointer()
|
||
|
||
app = create_fault_diagnosis_workflow(checkpointer=checkpointer)
|
||
|
||
if thread_id:
|
||
try:
|
||
current_state = await app.aget_state(config)
|
||
print(
|
||
f"[DEBUG] current_state exists: {current_state is not None}, has values: {current_state.values is not None if current_state else False}, tasks: {len(current_state.tasks) if current_state else 0}")
|
||
|
||
if current_state and current_state.values:
|
||
saved_state = current_state.values
|
||
print(
|
||
f"[Checkpointer] 发现已保存状态: ship_number={saved_state.get('ship_number')}, device_name={saved_state.get('device_name')}, fault={saved_state.get('fault')}")
|
||
|
||
if current_state.tasks:
|
||
print(f"[Checkpointer] 恢复执行，注入用户响应")
|
||
resume_value = user_response if user_response else {"query": combined_query}
|
||
result = await app.ainvoke(Command(resume=resume_value), config)
|
||
else:
|
||
# 只使用当前的输入，不拼接之前的！
|
||
merged_query = combined_query
|
||
|
||
# 如果当前输入是确认意图，且已有完整信息，直接视为用户已确认
|
||
_is_confirm = await is_confirmation_intent(combined_query)
|
||
_has_full_info = (saved_state.get("ship_number") and saved_state.get("device_name") and saved_state.get("fault"))
|
||
_override_confirmed = _is_confirm and _has_full_info and not saved_state.get("user_confirmed_info", False)
|
||
|
||
initial_state = {**saved_state,"combined_query": merged_query,"history_messages": parse_history(history_message),"initialized": False,"user_intent": "",}
|
||
if _override_confirmed:
|
||
print(f"[Checkpointer] 检测到确认意图，自动设置 user_confirmed_info=True")
|
||
initial_state["user_confirmed_info"] = True
|
||
result = await app.ainvoke(initial_state, config)
|
||
else:
|
||
initial_state = {"extracted_text": extracted_text,"combined_query": combined_query,"history_messages": parse_history(history_message),"ship_number": "","device_name": "","fault": "","additional_info": "",
|
||
"fault_code": "","fault_time": "","fault_frequency": "","tried_measures": "","running_condition": "","rag_search_result": None,"graph_search_result": None,"response": "",
|
||
"suggestedReplies": [],"actions": [],"error_message": "","agent_decision": "","agent_reasoning": "","tools_to_call": [],"missing_info": [],"info_completeness": 0.0,
|
||
"matched_kb_id": "","matched_kb_name": "","iteration_count": 0,"max_iterations": 5,"scheme_generated": False,"waiting_for_supplement": False,"extracted_info": {},
|
||
"has_rag_result": False,"has_graph_result": False,"is_execution_feedback": False,"execution_detail": "","ask_round": 0,"user_confirmed_info": False,"need_model_confirm": False,
|
||
"user_confirmed_model": False,"waiting_feedback": False,"user_feedback": "","waiting_report_confirm": False,"report_confirmed": False,"initialized": False,"current_node": "","user_intent": "","last_combined_query": "",}
|
||
result = await app.ainvoke(initial_state, config)
|
||
except Exception as e:
|
||
print(f"[Checkpointer] 状态恢复失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
result = {}
|
||
else:
|
||
initial_state = {"extracted_text": extracted_text,"combined_query": combined_query,"history_messages": parse_history(history_message),"ship_number": "","device_name": "","fault": "","additional_info": "","fault_code": "","fault_time": "",
|
||
"fault_frequency": "","tried_measures": "","running_condition": "","rag_search_result": None,"graph_search_result": None,"response": "","suggestedReplies": [],"actions": [],"error_message": "","agent_decision": "","agent_reasoning": "","tools_to_call": [],
|
||
"missing_info": [],"info_completeness": 0.0,"matched_kb_id": "","matched_kb_name": "","iteration_count": 0,"max_iterations": 5,"scheme_generated": False,"waiting_for_supplement": False,"extracted_info": {},"has_rag_result": False,"has_graph_result": False,
|
||
"is_execution_feedback": False,"execution_detail": "","ask_round": 0,"user_confirmed_info": False,"need_model_confirm": False,"user_confirmed_model": False,"waiting_feedback": False,"user_feedback": "","waiting_report_confirm": False,"report_confirmed": False,"initialized": False,"current_node": "","user_intent": "","last_combined_query": "",}
|
||
|
||
try:
|
||
result = await app.ainvoke(initial_state, config)
|
||
except Exception as e:
|
||
print(f"[工作流执行失败] {str(e)}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return {"response": f"故障诊断工作流执行失败: {str(e)}","actions": [],"result_tag": "fault_diagnosis","suggestedReplies": [],}
|
||
|
||
if result is None:
|
||
result = {}
|
||
|
||
interrupts = result.get("__interrupt__", [])
|
||
if interrupts:
|
||
interrupt_data = interrupts[0].value if interrupts else {}
|
||
print(f"[Interrupt] 工作流中断，等待用户输入: {interrupt_data}")
|
||
return {"response": interrupt_data.get("response", "请提供更多信息"),"actions": interrupt_data.get("actions", []),"result_tag": "fault_diagnosis","suggestedReplies": interrupt_data.get("suggestedReplies", []),"waiting_for": interrupt_data.get("waiting_for", ""),}
|
||
|
||
if result.get("error_message"):
|
||
return {"response": f"故障诊断工作流执行失败: {result['error_message']}","actions": [],"result_tag": "fault_diagnosis","suggestedReplies": [],}
|
||
|
||
route_flag = result.get("route_flag", "")
|
||
if not route_flag:
|
||
route_flag = "fault_diagnosis"
|
||
|
||
return {"response": result.get("response", ""),"actions": result.get("actions", []),"result_tag": route_flag,"suggestedReplies": result.get("suggestedReplies", []),"route_flag": route_flag,}
|