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"""
你是一名设备运维与故障分析报告编写专家。
你的任务是根据用户需求、历史对话和抽取信息生成一份正式的故障处理报告。
{combined_query}
{extracted_items}
{history_context if history_context else "(无历史对话上下文)"}
{f'''
{image_section}
''' if extracted_images else ''}
必须严格遵守以下规则:
1. 输出为 **正式技术报告文本**
2. **使用段落形式,如果有表格生成,生成表格必须arkdown表格,不要html形式**
3. **如果给定结果在历史对话信息中没有,可以进行删除**
4. **输出内容必须严格按照历史对话信息内容进行生成,不得编造不实信息和不相干信息**
5. 不得输出解释说明
6. 不要使用列表符号
7. 所有内容使用完整中文句子。
8. **每个段落首行必须缩进,使用两个全角空格( )实现首行缩进两个中文字符的效果**
{"9. **报告中必须包含用户上传的图片信息,在故障信息部分引用图片URL**" if extracted_images else ""}
{report_title}
报告编号: {report_number}
编制时间: {compile_time}
事件类型: 设备故障诊断与处置
一、事件概述
简要说明故障发生背景、报警情况、系统影响以及运维团队启动排查流程的情况。
二、故障信息
说明设备名称、型号、所属系统、报警时间、故障现象以及参与处置人员。
{" **附件图片:**" + chr(10) + chr(10).join([f" " for idx, img in enumerate(extracted_images, 1)]) if extracted_images else ""}
三、故障排查过程
按照时间顺序描述排查步骤,包括:用户反馈 → 初步判断 → 数据分析 → 现场检查 → 故障确认。
四、维修措施
描述维修人员采取的具体维修步骤,包括设备停机、拆卸检查、更换部件、紧固连接、恢复运行和测试过程。
五、后续建议
提出预防类似故障的改进措施,例如:维护周期优化、知识库记录、巡检项目增加、技术培训等。
六、总结
总结本次故障原因、处理效果以及经验意义。
只输出完整报告正文,不要任何解释说明。
"""
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("report")
download_url = f"http://192.168.1.164:9088/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())