545 lines
19 KiB
Python
545 lines
19 KiB
Python
import asyncio
|
||
import re
|
||
import random
|
||
from typing import TypedDict, List, Dict, Any, Literal
|
||
from langgraph.graph import StateGraph, START, END
|
||
import os
|
||
import sys
|
||
|
||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||
parent_dir = os.path.dirname(current_dir)
|
||
if parent_dir not in sys.path:
|
||
sys.path.append(parent_dir)
|
||
|
||
from modelsAPI.model_api import OpenaiAPI
|
||
from utils.function_tracker import track_function_calls
|
||
from utils.word_generator import generate_report_word
|
||
from config import SERVER_CONFIG
|
||
import json
|
||
|
||
|
||
from workflows.history_manager import init_history, build_history_str
|
||
|
||
|
||
def _extract_images_from_text(text: str) -> List[Dict[str, str]]:
|
||
"""从文本中提取图片URL和描述信息"""
|
||
if not text:
|
||
return []
|
||
|
||
images = []
|
||
url_pattern = r'https?://[^\s<>"{}|\\^`\[\]]+\.(?:png|jpg|jpeg|gif|bmp|webp)'
|
||
urls = re.findall(url_pattern, text, re.IGNORECASE)
|
||
|
||
for url in urls:
|
||
summary = ""
|
||
url_index = text.find(url)
|
||
if url_index != -1:
|
||
after_url = text[url_index + len(url):]
|
||
result_match = re.search(
|
||
r'(?:\*\*图片识别结果：\*\*|图片识别结果[：:]|识别结果[：:])\s*(.*?)(?=\n\n\n|\n\n[A-Za-z\u4e00-\u9fa5]|$)',
|
||
after_url,
|
||
re.DOTALL
|
||
)
|
||
if result_match:
|
||
summary = result_match.group(1).strip()[:300]
|
||
|
||
images.append({
|
||
"url": url,
|
||
"summary": summary or "用户上传的图片"
|
||
})
|
||
|
||
return images
|
||
|
||
|
||
class HistoryQAAnalysisState(TypedDict):
|
||
"""历史问答分析工作流状态"""
|
||
|
||
extracted_text: str
|
||
combined_query: str
|
||
|
||
history_message: str
|
||
history_messages: List[Dict[str, Any]]
|
||
|
||
extracted_items: List[Dict[str, Any]]
|
||
extracted_images: List[Dict[str, str]]
|
||
report_title: str
|
||
report_text: str
|
||
|
||
response: str
|
||
actions: List[Dict[str, Any]]
|
||
route_flag: Literal["", "report"]
|
||
suggestedReplies: List[Dict[str, Any]]
|
||
|
||
error_message: str
|
||
|
||
|
||
async def extract_analysis_items(state: HistoryQAAnalysisState) -> Dict[str, Any]:
|
||
from utils.function_tracker import get_all_callbacks
|
||
|
||
combined_query = state.get("combined_query", "") or state.get("extracted_text", "") or ""
|
||
history_messages = state.get("history_messages", [])
|
||
history_message = state.get("history_message", "")
|
||
|
||
pre_extracted_images = _extract_images_from_text(history_message)
|
||
|
||
history_context = build_history_str(history_messages, recent_count=10, assistant_truncate=200)
|
||
|
||
if not history_context and history_message:
|
||
history_context = history_message
|
||
|
||
loading_prompts = [
|
||
{"title": "提取相关历史问答数据中...", "details": f"历史消息数: {len(history_messages)}条"},
|
||
{"title": "🔍 正在分析历史对话内容...", "details": f"共处理 {len(history_messages)} 条消息"},
|
||
{"title": "📝 解析历史问答信息中...", "details": f"已加载 {len(history_messages)} 条历史记录"},
|
||
{"title": "📊 整理历史对话数据...", "details": f"当前历史消息: {len(history_messages)} 条"},
|
||
{"title": "🔑 提取对话关键信息...", "details": f"历史对话总数: {len(history_messages)}条"},
|
||
]
|
||
selected_prompt = random.choice(loading_prompts)
|
||
parent_event = {
|
||
"type": "function_execution",
|
||
"title": selected_prompt["title"],
|
||
"details": selected_prompt["details"],
|
||
}
|
||
for callback in get_all_callbacks():
|
||
try:
|
||
callback(parent_event)
|
||
except Exception:
|
||
pass
|
||
|
||
prompt = f"""
|
||
你是一名情报分析与知识管理专家。
|
||
请从以下【用户输入】中抽取用于统计分析的结构化信息。
|
||
|
||
【用户输入】
|
||
{combined_query}
|
||
|
||
【历史对话上下文】
|
||
{history_context if history_context else "（无）"}
|
||
|
||
|
||
【抽取规则】
|
||
输出一个JSON对象，包含以下字段：
|
||
- report_title：报告标题（根据历史对话内容生成一个简洁的标题，格式为"XXX故障处理报告"，不超过20个字）
|
||
- items：抽取记录数组，每条记录包含：
|
||
- topic：问题主题（如 设备、系统、管理、流程、保障、训练 等）
|
||
- key_entity：关键对象或名词
|
||
- conclusion：结论 / 判断 / 回答要点
|
||
- action_suggestion：建议或行动（若无可为空）
|
||
- date：时间（若无法判断可为空）
|
||
- images：图片信息数组（仅当有用户上传图片时提取，每条记录包含）：
|
||
- url：图片URL（直接从【用户上传图片信息】中提取）
|
||
- summary：图片内容摘要（根据图片识别结果提炼关键信息，如设备类型、故障现象、异常情况等，不超过100字）
|
||
|
||
【输出要求】
|
||
1. 仅输出 JSON 对象
|
||
2. 不要附加任何解释
|
||
3. 如果没有图片信息，images字段返回空数组
|
||
"""
|
||
|
||
try:
|
||
result = await OpenaiAPI.open_api_chat_without_thinking(
|
||
query=prompt,
|
||
model=None,
|
||
json_output=True
|
||
)
|
||
parsed_result = json.loads(result)
|
||
if isinstance(parsed_result, dict):
|
||
extracted_items = parsed_result.get("items", [])
|
||
report_title = parsed_result.get("report_title", "历史问答分析报告")
|
||
model_images = parsed_result.get("images", [])
|
||
|
||
|
||
else:
|
||
extracted_items = parsed_result if isinstance(parsed_result, list) else []
|
||
report_title = "历史问答分析报告"
|
||
model_images = []
|
||
except Exception as e:
|
||
extracted_items = []
|
||
report_title = "历史问答分析报告"
|
||
model_images = []
|
||
|
||
final_images = pre_extracted_images if pre_extracted_images else model_images
|
||
|
||
return {"extracted_items": extracted_items, "report_title": report_title, "extracted_images": final_images}
|
||
|
||
|
||
# @track_function_calls
|
||
async def generate_analysis_report_text(state: HistoryQAAnalysisState) -> Dict[str, Any]:
|
||
from utils.function_tracker import get_all_callbacks
|
||
import time
|
||
|
||
extracted_items = state.get("extracted_items", [])
|
||
report_title = state.get("report_title", "历史问答分析报告")
|
||
combined_query = state.get("combined_query", "") or state.get("extracted_text", "") or ""
|
||
history_messages = state.get("history_messages", [])
|
||
extracted_images = state.get("extracted_images", [])
|
||
history_message = state.get("history_message", "")
|
||
|
||
def summarize_extracted_items(items) -> str:
|
||
"""将抽取结果转为中文描述，兼容 list / dict"""
|
||
|
||
if not items:
|
||
return "未提取到有效分析数据"
|
||
|
||
# 如果是单个 dict，转为 list
|
||
if isinstance(items, dict):
|
||
items = [items]
|
||
|
||
# 如果不是 list，直接返回
|
||
if not isinstance(items, list):
|
||
return "分析数据格式异常"
|
||
|
||
topics = set()
|
||
entities = set()
|
||
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
|
||
topic = item.get("topic")
|
||
entity = item.get("key_entity")
|
||
|
||
if topic:
|
||
topics.add(str(topic))
|
||
if entity:
|
||
entities.add(str(entity))
|
||
|
||
topic_text = "、".join(list(topics)[:5]) if topics else "未识别"
|
||
entity_text = "、".join(list(entities)[:5]) if entities else "未识别"
|
||
|
||
return f"识别主题 {len(items)} 条，涉及主题：{topic_text}；关键对象：{entity_text}"
|
||
|
||
details_text = summarize_extracted_items(extracted_items)
|
||
loading_prompts = [
|
||
{"title": "📄 生成分析报告文本", "details": details_text},
|
||
{"title": "✍️ 正在撰写分析报告", "details": details_text},
|
||
{"title": "📋 生成历史问答分析报告", "details": details_text},
|
||
{"title": "📝 构建分析报告结构", "details": details_text},
|
||
{"title": "🎯 汇总历史问答数据生成报告", "details": details_text},
|
||
]
|
||
selected_prompt = random.choice(loading_prompts)
|
||
start_event = {
|
||
"type": "function_execution",
|
||
"title": selected_prompt["title"],
|
||
"details": selected_prompt["details"]
|
||
}
|
||
for callback in get_all_callbacks():
|
||
try:
|
||
callback(start_event)
|
||
except Exception:
|
||
pass
|
||
|
||
history_context = build_history_str(history_messages, recent_count=10, assistant_truncate=300)
|
||
|
||
if not history_context and history_message:
|
||
history_context = history_message
|
||
|
||
if not history_context and not combined_query:
|
||
return {
|
||
"report_text": "",
|
||
"response": "【无法生成分析报告】\n\n缺少历史对话信息和用户输入，请先进行相关问答。",
|
||
"actions": [],
|
||
"route_flag": "report",
|
||
"suggestedReplies": [],
|
||
"error_message": "缺少历史对话信息和用户输入"
|
||
}
|
||
|
||
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")
|
||
|
||
image_section = ""
|
||
if extracted_images:
|
||
image_parts = []
|
||
for idx, img in enumerate(extracted_images, 1):
|
||
image_parts.append(f"图片{idx}：{img['url']}")
|
||
summary = img.get('summary') or img.get('description', '')
|
||
if summary:
|
||
image_parts.append(f"识别结果：{summary}")
|
||
image_parts.append("")
|
||
image_section = "\n".join(image_parts)
|
||
|
||
prompt = f"""
|
||
<role>
|
||
你是一名设备运维与故障分析报告编写专家。
|
||
你的任务是根据用户需求、历史对话和抽取信息生成一份正式的故障处理报告。
|
||
</role>
|
||
|
||
<user_requirement>
|
||
{combined_query}
|
||
</user_requirement>
|
||
|
||
<extracted_items>
|
||
{extracted_items}
|
||
</extracted_items>
|
||
|
||
<history_context>
|
||
{history_context if history_context else "（无历史对话上下文）"}
|
||
</history_context>
|
||
|
||
{f'''<user_images>
|
||
{image_section}
|
||
</user_images>''' if extracted_images else ''}
|
||
|
||
<output_rules>
|
||
|
||
必须严格遵守以下规则：
|
||
|
||
1. 输出为 **正式技术报告文本**
|
||
2. **使用段落形式，如果有表格生成，生成表格必须arkdown表格，不要html形式**
|
||
3. **如果给定结果在历史对话信息中没有，可以进行删除**
|
||
4. **输出内容必须严格按照历史对话信息内容进行生成，不得编造不实信息和不相干信息**
|
||
5. 不得输出解释说明
|
||
6. 不要使用列表符号
|
||
7. 所有内容使用完整中文句子。
|
||
8. **每个段落首行必须缩进，使用两个全角空格（ ）实现首行缩进两个中文字符的效果**
|
||
{"9. **报告中必须包含用户上传的图片信息，在故障信息部分引用图片URL**" if extracted_images else ""}
|
||
|
||
</output_rules>
|
||
|
||
<report_template>
|
||
|
||
{report_title}
|
||
|
||
报告编号： {report_number}
|
||
编制时间： {compile_time}
|
||
事件类型： 设备故障诊断与处置
|
||
|
||
一、事件概述
|
||
简要说明故障发生背景、报警情况、系统影响以及运维团队启动排查流程的情况。
|
||
|
||
二、故障信息
|
||
说明设备名称、型号、所属系统、报警时间、故障现象以及参与处置人员。
|
||
{" **附件图片：**" + chr(10) + chr(10).join([f" " for idx, img in enumerate(extracted_images, 1)]) if extracted_images else ""}
|
||
|
||
三、故障排查过程
|
||
按照时间顺序描述排查步骤，包括：用户反馈 → 初步判断 → 数据分析 → 现场检查 → 故障确认。
|
||
|
||
四、维修措施
|
||
描述维修人员采取的具体维修步骤，包括设备停机、拆卸检查、更换部件、紧固连接、恢复运行和测试过程。
|
||
|
||
五、后续建议
|
||
提出预防类似故障的改进措施，例如：维护周期优化、知识库记录、巡检项目增加、技术培训等。
|
||
|
||
六、总结
|
||
总结本次故障原因、处理效果以及经验意义。
|
||
|
||
</report_template>
|
||
|
||
<final_rule>
|
||
只输出完整报告正文，不要任何解释说明。
|
||
</final_rule>
|
||
"""
|
||
|
||
try:
|
||
report_text = await OpenaiAPI.open_api_chat_without_thinking(
|
||
query=prompt,
|
||
model=None,
|
||
messages=history_messages
|
||
)
|
||
if not report_text:
|
||
report_text = "抱歉，当前无法生成分析报告，请稍后重试。"
|
||
|
||
word_dir = os.path.join(parent_dir, "created_word")
|
||
os.makedirs(word_dir, exist_ok=True)
|
||
|
||
safe_topic = re.sub(r'[\\/:*?"<>|]', '_', report_title)
|
||
file_name = f"{safe_topic}.docx"
|
||
output_path = os.path.join(word_dir, file_name)
|
||
|
||
word_result = generate_report_word(
|
||
markdown=report_text,
|
||
title=report_title,
|
||
output=output_path
|
||
)
|
||
|
||
actions: List[Dict[str, Any]] = []
|
||
|
||
if word_result and os.path.exists(word_result):
|
||
base_url = SERVER_CONFIG.get("base_url")
|
||
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_text
|
||
})
|
||
else:
|
||
error_msg = f"\n\n[注意] Word 文档生成失败"
|
||
report_text += error_msg
|
||
|
||
# actions.append({
|
||
# "type": "regenerate",
|
||
# "label": "重新生成",
|
||
# "icon": "report"
|
||
# })
|
||
|
||
suggested_replies = await generate_suggested_replies_qa_report(report_text)
|
||
|
||
return {
|
||
"report_text": report_text,
|
||
"response": report_text,
|
||
"actions": actions,
|
||
"route_flag": "report",
|
||
"suggestedReplies": [],
|
||
# "suggestedReplies": suggested_replies,
|
||
"__details__": "分析报告生成完成"
|
||
}
|
||
except Exception as e:
|
||
# error_event = {
|
||
# "type": "function_execution",
|
||
# "title": "步骤二失败：生成分析报告",
|
||
# "details": f"错误: {str(e)}",
|
||
# "has_result": False
|
||
# }
|
||
# for callback in get_all_callbacks():
|
||
# try:
|
||
# callback(error_event)
|
||
# except Exception:
|
||
# pass
|
||
|
||
return {
|
||
"report_text": f"生成分析报告失败: {str(e)}",
|
||
"response": f"生成分析报告失败: {str(e)}",
|
||
"actions": [],
|
||
"route_flag": "report",
|
||
"suggestedReplies": [],
|
||
"error_message": str(e),
|
||
"__details__": f"生成分析报告失败: {str(e)}"
|
||
}
|
||
|
||
|
||
async def generate_suggested_replies_qa_report(report: str) -> List[Dict[str, Any]]:
|
||
"""
|
||
生成针对历史问答分析报告的建议回复问题
|
||
使用大模型生成两个相关问题
|
||
"""
|
||
prompt = f"""
|
||
你是一个专业的数据分析助手。根据以下历史问答分析报告，生成两个用户可能关心的后续问题。
|
||
|
||
【分析报告】
|
||
{report[:500]}
|
||
|
||
请生成两个简洁、实用的问题，这些问题应该：
|
||
1. 不得出现需要归档和下载等操作相关的问题
|
||
2. 与分析报告相关，能够帮助用户更好地理解或使用分析结果
|
||
3. 问题要具体、可操作
|
||
4. 每个问题不超过20个字
|
||
|
||
请以JSON格式输出，格式如下：
|
||
[
|
||
{{"title": "问题1的标题", "content": "问题1的完整内容"}},
|
||
{{"title": "问题2的标题", "content": "问题2的完整内容"}}
|
||
]
|
||
|
||
只输出JSON，不要有其他文字说明。
|
||
"""
|
||
try:
|
||
answer = await OpenaiAPI.open_api_chat_without_thinking(
|
||
query=prompt,
|
||
model=None
|
||
)
|
||
json_match = re.search(r'\[.*\]', answer, re.DOTALL)
|
||
if json_match:
|
||
json_str = json_match.group(0)
|
||
suggested_replies = json.loads(json_str)
|
||
if isinstance(suggested_replies, list) and len(suggested_replies) >= 2:
|
||
formatted_replies = []
|
||
for reply in suggested_replies[:2]:
|
||
if isinstance(reply, dict) and "title" in reply and "content" in reply:
|
||
formatted_replies.append({
|
||
"title": str(reply["title"]),
|
||
"content": str(reply["content"])
|
||
})
|
||
if len(formatted_replies) == 2:
|
||
return formatted_replies
|
||
|
||
return [
|
||
{"title": "如何优化问题分布？", "content": "如何优化问题分布？"},
|
||
{"title": "有哪些改进建议？", "content": "有哪些改进建议？"}
|
||
]
|
||
except Exception as e:
|
||
return [
|
||
{"title": "如何优化问题分布？", "content": "如何优化问题分布？"},
|
||
{"title": "有哪些改进建议？", "content": "有哪些改进建议？"}
|
||
]
|
||
|
||
|
||
def create_history_qa_analysis_workflow():
|
||
graph = StateGraph(HistoryQAAnalysisState)
|
||
|
||
graph.add_node("extract_items", extract_analysis_items)
|
||
graph.add_node("generate_text", generate_analysis_report_text)
|
||
|
||
graph.add_edge(START, "extract_items")
|
||
graph.add_edge("extract_items", "generate_text")
|
||
graph.add_edge("generate_text", END)
|
||
|
||
return graph.compile()
|
||
|
||
|
||
# @track_function_calls
|
||
async def run_qa_report_agent(
|
||
extracted_text: str,
|
||
combined_query: str,
|
||
history_message: str = "",
|
||
route_flag: str = "",
|
||
extracted_images: List[Dict[str, str]] = None
|
||
) -> Dict[str, Any]:
|
||
app = create_history_qa_analysis_workflow()
|
||
|
||
initial_state: HistoryQAAnalysisState = {
|
||
"extracted_text": extracted_text,
|
||
"combined_query": combined_query,
|
||
**init_history(history_message),
|
||
"extracted_items": [],
|
||
"extracted_images": extracted_images or [],
|
||
"report_title": "",
|
||
"report_text": "",
|
||
"response": "",
|
||
"actions": [],
|
||
"route_flag": "",
|
||
"suggestedReplies": [],
|
||
"error_message": ""
|
||
}
|
||
|
||
return await app.ainvoke(initial_state)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import asyncio
|
||
|
||
|
||
async def test():
|
||
history_message = json.dumps(
|
||
[
|
||
{"role": "user", "content": "我需要分析最近的历史问答数据"},
|
||
{"role": "assistant", "content": "好的，我可以帮您生成历史问答分析报告，请提供相关数据。"},
|
||
{"role": "user", "content": "请重点关注设备故障和维护相关的内容"}
|
||
],
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
result = await run_qa_report_agent(
|
||
extracted_text="请重点关注设备故障和维护相关的内容",
|
||
combined_query="用户问题：请重点关注设备故障和维护相关的内容",
|
||
history_message=history_message
|
||
)
|
||
|
||
print("====== 历史问答分析报告文本 ======")
|
||
print(result["response"])
|
||
print("\n====== 推荐问题 ======")
|
||
for reply in result.get("suggestedReplies", []):
|
||
print(f"- {reply['title']}: {reply['content']}")
|
||
print("\n====== 操作按钮 ======")
|
||
for action in result.get("actions", []):
|
||
print(f"- {action.get('label', 'Unknown')}: {action.get('type', 'Unknown')}")
|
||
|
||
|
||
asyncio.run(test())
|
||
|
||
|
||
|