From e84cd1f2aa9e23d06a0639a5680434850c23a34d Mon Sep 17 00:00:00 2001 From: ChenRui Date: Wed, 22 Jul 2026 17:17:32 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=8C=E6=AD=A546=E6=9C=8D=E5=8A=A1=E5=99=A8?= =?UTF-8?q?=E4=BB=A3=E7=A0=81=E5=B9=B6=E4=BF=AE=E5=A4=8DRAG=E6=BA=AF?= =?UTF-8?q?=E6=BA=90=E5=90=88=E5=B9=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app.py | 16 +- config.py | 10 + modelsAPI/model_api.py | 86 +++- prompts.py | 85 ++-- tools/__init__.py | 82 ++-- tools/graph_tools.py | 2 + utils/__init__.py | 8 +- utils/function_tracker.py | 720 ++++++++++++++++---------------- utils/source_citations.py | 71 ++++ workflow_registry.py | 316 +++++++------- workflows/__init__.py | 40 +- workflows/workflow_baike.py | 806 ++++++++++++++++++------------------ workflows/workflow_other.py | 586 +++++++++++++------------- workflows/workflow_utils.py | 7 + 14 files changed, 1497 insertions(+), 1338 deletions(-) create mode 100644 utils/source_citations.py diff --git a/app.py b/app.py index 1d76b11..708a46f 100644 --- a/app.py +++ b/app.py @@ -31,6 +31,7 @@ from workflow_registry import WORKFLOW_CONFIG, VALID_ROUTE_FLAGS from main_agent import extract_conversation_title from workflows.history_manager import filter_image_urls, preprocess_from_request from config import set_request_user_config +from utils.source_citations import merge_rag_source_citations from checkpointer_config import ( CheckpointerManager, checkpointer_manager @@ -462,7 +463,7 @@ async def stream_main_agent_execution( agent_task = asyncio.create_task(run_agent()) execution_complete_received = False - rag_result = [] + rag_result = {} graph_result = [] while True: try: @@ -482,16 +483,7 @@ async def stream_main_agent_execution( if title == "知识库搜索工具" and result and isinstance(result, dict): print("===============================", result) rag_result_raw = result.get("sourceCitation", {}) - rag_result = {} - for doc_name, chunks in rag_result_raw.items(): - if isinstance(chunks, list): - rag_result[doc_name] = [ - {k: v for k, v in item.items() if k != "text"} - if isinstance(item, dict) else item - for item in chunks - ] - else: - rag_result[doc_name] = chunks + merge_rag_source_citations(rag_result, rag_result_raw) elif title == "图谱检索工具" and result and isinstance(result, dict): print("===============================", result) graph_result = result.get("xxxx.graph", []) @@ -1017,5 +1009,3 @@ if __name__ == "__main__": - - diff --git a/config.py b/config.py index 96e8e04..28150fd 100644 --- a/config.py +++ b/config.py @@ -220,3 +220,13 @@ NEO4J_CONFIG = { "user": "neo4j", "password": "zdht123@" } + +# ==================== Rerank 服务配置 ==================== +RERANK_CONFIG = { + "enabled": "true".lower() == "true", + "endpoint": "http://192.168.0.46:59600/v1/rerank", + "model": "bge-rerank", + "timeout": 30.0, + "max_candidates": 24, + "top_k": 10, +} diff --git a/modelsAPI/model_api.py b/modelsAPI/model_api.py index e55e56b..3581eae 100644 --- a/modelsAPI/model_api.py +++ b/modelsAPI/model_api.py @@ -4,7 +4,7 @@ from typing import List, Union import os import numpy as np from openai import OpenAI -from config import LLM_CONFIG, EMBEDDING_CONFIG +from config import LLM_CONFIG, EMBEDDING_CONFIG, RERANK_CONFIG from neo4j_graphrag.embeddings.base import Embedder from openai import AsyncOpenAI @@ -24,6 +24,25 @@ def _get_async_client() -> AsyncOpenAI: return _async_client +def _log_model_messages(label: str, model: str, message_list: list) -> None: + if os.getenv("LLM_LOG_FULL_PROMPTS", "false").lower() == "true": + print(label, message_list) + return + if os.getenv("LLM_LOG_CALLS", "true").lower() != "true": + return + summary = [] + total_chars = 0 + for message in message_list or []: + content = message.get("content") if isinstance(message, dict) else "" + if isinstance(content, str): + chars = len(content) + else: + chars = len(str(content)) + total_chars += chars + summary.append({"role": message.get("role", ""), "chars": chars}) + print(label, {"model": model, "messages": summary, "total_chars": total_chars}) + + class OpenaiAPI: @staticmethod async def open_api_chat_without_thinking( @@ -64,7 +83,7 @@ class OpenaiAPI: if json_output: kwargs["response_format"] = {"type": "json_object"} - print("model API. history message", message_list) + _log_model_messages("model API call", model, message_list) response = await client.chat.completions.create(**kwargs) return response.choices[0].message.content @@ -112,7 +131,7 @@ class OpenaiAPI: if json_output: kwargs["response_format"] = {"type": "json_object"} - print("model API stream. history message", message_list) + _log_model_messages("model API stream call", model, message_list) stream = await client.chat.completions.create(**kwargs) async for chunk in stream: @@ -224,6 +243,66 @@ class OpenaiAPI: embedding = response.json()["data"][0]["embedding"] return embedding + @staticmethod + async def rerank( + query: str, + documents: List[str], + top_k: int = None, + model: str = None, + ) -> List[dict]: + """ + 调用 59600 rerank 服务。 + + 返回格式: + [ + {"index": 0, "relevance_score": 0.98}, + ... + ] + """ + if not RERANK_CONFIG.get("enabled", True) or not query or not documents: + return [] + + max_candidates = int(RERANK_CONFIG.get("max_candidates") or 24) + selected_documents = [str(x or "")[:4000] for x in documents[:max_candidates]] + + try: + async with httpx.AsyncClient( + timeout=float(RERANK_CONFIG.get("timeout") or 30), + verify=False, + trust_env=False, + ) as client: + response = await client.post( + str(RERANK_CONFIG["endpoint"]), + json={ + "model": model or RERANK_CONFIG.get("model") or "bge-rerank", + "query": query, + "documents": selected_documents, + }, + ) + response.raise_for_status() + payload = response.json() + except Exception as exc: + print(f"[rerank] rerank service failed: {exc}") + return [] + + rows: List[dict] = [] + for row in payload.get("results", []): + try: + idx = int(row.get("index")) + except Exception: + continue + if idx < 0 or idx >= len(selected_documents): + continue + rows.append({ + "index": idx, + "relevance_score": float(row.get("relevance_score") or 0.0), + }) + + rows.sort(key=lambda x: float(x.get("relevance_score") or 0), reverse=True) + if top_k is not None: + rows = rows[:top_k] + return rows + @staticmethod async def get_embeddings_batch_async(texts: List[str]) -> List[List[float]]: embedding_base_url = EMBEDDING_CONFIG["base_url"] @@ -384,3 +463,4 @@ if __name__ == "__main__": + diff --git a/prompts.py b/prompts.py index 3a47400..fbfae05 100644 --- a/prompts.py +++ b/prompts.py @@ -427,29 +427,19 @@ BAIKE_PROMPTS = { # 输入变量: history_messages, original_query # ------------------------------------------------------------------ "judge_need_retrieval": ( - "你是一个专业的问题分析助手，请判断用户的问题是否需要检索知识库才能回答。\n" + "You are an intent classifier for a knowledge-base QA assistant. Decide whether the user question needs retrieval.\n" "\n" - "【判断标准】\n" - "需要检索的情况：\n" - "- 询问特定设备、技术、规程、规范的具体内容\n" - "- 需要专业知识或特定数据支持的问题\n" - "- 询问故障诊断、维修方法等专业领域问题\n" - "- 询问历史事件、法规条款、技术标准等需要查证的内容\n" - "- 例如发动机的组成等\n" + "Return true when the question asks about documents, files, knowledge-base content, equipment, systems, faults, operations, standards, procedures, meeting notes, database-related materials, or any factual/domain-specific information.\n" + "Return false only for pure greetings, thanks, cancellation/confirmation with no substantive question, or a follow-up that can be fully answered from the conversation history alone.\n" + "Default rule: when unsure, return true.\n" "\n" - "不需要检索的情况：\n" - "- 简单的问候、寒暄\n" - "- 常识性问题（如常识性知识、简单计算等）\n" - "- 明确不需要专业知识的问题\n" - "- 对之前对话的简单追问，上下文已足够回答\n" - "\n" - "【对话历史】\n" + "Conversation history:\n" "{history_messages}\n" "\n" - "【当前用户问题】\n" + "User question:\n" "{original_query}\n" "\n" - "请仅输出 \"true\" 或 \"false\"（不带引号），true 表示需要检索，false 表示不需要检索。" + "Output only true or false." ), # ------------------------------------------------------------------ @@ -457,31 +447,35 @@ BAIKE_PROMPTS = { # 输入变量: last_user_query, history_str, original_query # ------------------------------------------------------------------ "rewrite_query_with_history": ( - "你是一个专业的检索query重写助手，负责根据对话历史和当前用户问题，生成**适合向量检索/RAG 的单条查询语句**。\n" + "You are a query understanding assistant for a Chinese knowledge-base retrieval system.\n" "\n" - "请严格遵守以下要求：\n" - "1. 只融合与当前问题语义强相关的**最近几轮对话信息**，不要机械拼接全部历史。\n" - "2. 消除指代和省略（例如：这个、刚才那个问题、上面说的故障等），补全成自洽、完整的描述。\n" - "3. 不要加入与用户问题无关的背景信息，不要过度扩展检索范围。\n" - "4. 输出必须是一句自然的中文问句或陈述句，便于检索；不要包含条目符号、编号或解释性文字。\n" - "5. 只输出改写后的query本身：\n" - " - 不要输出说明文字\n" - " - 不要包含「改写结果：」「检索query：」等前缀\n" - " - 不要添加引号\n" + "Tasks:\n" + "1. Rewrite the current user question into a self-contained Chinese retrieval query.\n" + "2. Generate 2-5 short search queries for hybrid retrieval.\n" "\n" - "【特别提示】\n" - "- 请优先参考**上一轮用户的问题**来重写本轮query。\n" - "- 上一轮用户的问题是：{last_user_query}\n" + "Rules:\n" + "- Resolve references and ellipsis using only relevant recent history.\n" + "- Preserve concrete entities, file names, database/table terms, equipment names, fault names, standards, procedure names, and exact keywords.\n" + "- Do not output meta instructions such as “在知识库中查找”. Output the actual search content.\n" + "- Include compact keyword queries that help exact matching, for example document titles, equipment names, or core nouns.\n" + "- The rewritten query and search queries must be in Chinese unless the original keyword is English.\n" "\n" - "【对话历史（JSON 格式，已按时间排序）】\n" + "Last user question:\n" + "{last_user_query}\n" + "\n" + "Conversation history JSON:\n" "{history_str}\n" "\n" - "【当前用户问题】\n" + "Current user question:\n" "{original_query}\n" "\n" - "请给出最终用于检索的单条中文query。" + "Output ONLY valid JSON:\n" + "{{\"rewrite_query\":\"string\",\"search_queries\":[\"string\"]}}" ), + + + # ------------------------------------------------------------------ # generate_suggested_replies_baike 函数：生成百科问答的建议回复问题 # 输入变量: question, answer @@ -514,8 +508,15 @@ BAIKE_PROMPTS = { # generate_answer 节点 - 有检索结果时的系统提示词 # 输入变量: domain_desc # ------------------------------------------------------------------ + # ------------------------------------------------------------------ "generate_answer_with_retrieval_system": ( - "我是一个专业的{domain_desc}知识问答助手，根据用户的问题综合检索结果，提供准确、专业、完整的回答。回答应围绕问题组织知识点、依据和必要图示，把参考资料中的相关图片作为资料证据放在对应说明附近。只能使用参考资料中真实存在的图片链接，不得使用占位链接或自行构造图片链接。" + "You are a professional knowledge-base assistant for {domain_desc}.\n" + "Answer ONLY from the retrieved evidence provided by the user message. Do not use prior knowledge as factual evidence.\n" + "If the evidence is insufficient, say clearly what is missing and what can be confirmed from the evidence.\n" + "Prefer concrete facts, names, parameters, steps, conclusions, and source relationships from the evidence.\n" + "When sources conflict, explain the conflict instead of inventing a merged fact.\n" + "Use images only when their URLs appear in the retrieved evidence; never fabricate or repair image URLs.\n" + "Always respond in Chinese. Do not expose internal tool names or implementation details." ), # ------------------------------------------------------------------ @@ -523,24 +524,24 @@ BAIKE_PROMPTS = { # 输入变量: extracted_text, rag_count, all_source_text # ------------------------------------------------------------------ "generate_answer_with_retrieval_user": ( + "用户问题：\n" "{extracted_text}\n" "\n" - "# 检索到的相关信息（共 {rag_count} 条）：\n" + "# Retrieved evidence ({rag_count} items)\n" "\n" "{all_source_text}\n" "\n" "【输出要求】\n" - "1. 先直接回答用户问题，再按需要补充结构、原理、参数、应用场景、注意事项或资料依据\n" - "2. 参考资料中真实存在的图片可在当前问题、设备结构、部件位置、流程步骤或现象说明对应段落附近自然展示；图片规则不要写入正文\n" - "3. 图片只作为资料引用输出为单个 Markdown 图片标签，URL 和文件名必须逐字来自参考资料，不要改写、拆分、补全、连续重复或自行构造；同一图片资源重复出现时只引用一次\n" - "4. 如果参考资料包含 LaTeX 公式、链接等，请在对应位置正常使用，不得修改或删除\n" - "5. 请专注于内容的准确性、完整性和逻辑性，不要担心格式的美观性\n" - "6. 可以适当使用简单的段落分隔，但不要使用复杂的Markdown格式\n" - "7. 如果参考资料包含表格，请不要输出表格，而要直接描述表格的内容\n" + "1. 先直接回答用户问题，再补充必要依据。\n" + "2. 每个关键事实都必须能在 Retrieved evidence 中找到依据；不要补充资料外结论。\n" + "3. 参考资料中真实存在的图片可在对应段落附近展示；URL 必须逐字来自资料。\n" + "4. 表格内容请转成自然语言描述，不要输出复杂表格。\n" "\n" "请给出回复：" ), + + # ------------------------------------------------------------------ # generate_answer 节点 - 无检索结果时的系统提示词 # 输入变量: domain_desc diff --git a/tools/__init__.py b/tools/__init__.py index 929ca88..4947b2a 100644 --- a/tools/__init__.py +++ b/tools/__init__.py @@ -1,41 +1,41 @@ - -""" -工具模块 -包含RAG、GraphRAG、VLM、ASR、TTS等工具 -""" -from .function_tool import ( - detect_input_type, - graph_rag_search, - fault_graph_rag_search, - operation_graph_rag_search, - rag_search_tool, - vlm_image_to_text, - asr_audio_to_text, - extract_device_and_fault, - image_rag_describe, - file_to_text_tool, - fault_type_statistics_tool, - fault_frequency_statistics_tool, - spare_parts_statistics_tool, - ALL_TOOLS, - TOOL_CATEGORIES -) - -__all__ = [ - "detect_input_type", - "graph_rag_search", - "fault_graph_rag_search", - "operation_graph_rag_search", - "rag_search_tool", - "vlm_image_to_text", - "asr_audio_to_text", - "extract_device_and_fault", - "image_rag_describe", - "file_to_text_tool", - "fault_type_statistics_tool", - "fault_frequency_statistics_tool", - "spare_parts_statistics_tool", - "ALL_TOOLS", - "TOOL_CATEGORIES" -] - + +""" +工具模块 +包含RAG、GraphRAG、VLM、ASR、TTS等工具 +""" +from .function_tool import ( + detect_input_type, + graph_rag_search, + fault_graph_rag_search, + operation_graph_rag_search, + rag_search_tool, + vlm_image_to_text, + asr_audio_to_text, + extract_device_and_fault, + image_rag_describe, + file_to_text_tool, + fault_type_statistics_tool, + fault_frequency_statistics_tool, + spare_parts_statistics_tool, + ALL_TOOLS, + TOOL_CATEGORIES +) + +__all__ = [ + "detect_input_type", + "graph_rag_search", + "fault_graph_rag_search", + "operation_graph_rag_search", + "rag_search_tool", + "vlm_image_to_text", + "asr_audio_to_text", + "extract_device_and_fault", + "image_rag_describe", + "file_to_text_tool", + "fault_type_statistics_tool", + "fault_frequency_statistics_tool", + "spare_parts_statistics_tool", + "ALL_TOOLS", + "TOOL_CATEGORIES" +] + diff --git a/tools/graph_tools.py b/tools/graph_tools.py index 05f11ff..df61bb8 100644 --- a/tools/graph_tools.py +++ b/tools/graph_tools.py @@ -536,6 +536,7 @@ def _node_record_to_dict(record: Any) -> Dict[str, Any]: return result +@track_function_calls async def normalize_device_name_by_ship_graph( device_name: str, fulltext_index: str = GLOBAL_FULLTEXT_INDEX, @@ -902,6 +903,7 @@ def _build_fault_graph_results_text( return "\n".join(parts) +@track_function_calls async def find_system_by_device( device_name: str, min_hops: int = 2, diff --git a/utils/__init__.py b/utils/__init__.py index 82c3782..00b5d34 100644 --- a/utils/__init__.py +++ b/utils/__init__.py @@ -1,4 +1,4 @@ -""" -工具模块 -""" - +""" +工具模块 +""" + diff --git a/utils/function_tracker.py b/utils/function_tracker.py index 29b5a2f..773393c 100644 --- a/utils/function_tracker.py +++ b/utils/function_tracker.py @@ -1,360 +1,360 @@ -""" -使用装饰器实现全栈函数调用追踪 -解决LangGraph无法捕获子函数内部状态的问题 -""" - -import functools -import inspect -import asyncio -import contextvars -from typing import Callable, Any, Dict, Optional, List -from datetime import datetime -from collections import deque - -# 使用 contextvars 实现请求级别的隔离，避免多个并发请求互相干扰 -_event_callbacks: contextvars.ContextVar[List[Callable]] = contextvars.ContextVar('event_callbacks', default=None) -_current_stream_handler: contextvars.ContextVar[Optional[Any]] = contextvars.ContextVar('current_stream_handler', - default=None) - - -def register_event_callback(callback: Callable[[Dict[str, Any]], None]): - """ - 注册事件回调函数（使用 contextvars 实现请求隔离） - """ - callbacks = _event_callbacks.get() - if callbacks is None: - callbacks = [] - _event_callbacks.set(callbacks) - callbacks.append(callback) - - -def unregister_event_callback(callback: Callable[[Dict[str, Any]], None]): - """ - 移除事件回调函数（使用 contextvars 实现请求隔离） - """ - callbacks = _event_callbacks.get() - if callbacks and callback in callbacks: - callbacks.remove(callback) - - -def clear_event_callbacks(): - """ - 清空所有事件回调（使用 contextvars 实现请求隔离） - """ - callbacks = _event_callbacks.get() - if callbacks: - callbacks.clear() - - -def get_all_callbacks(): - """ - 获取所有注册的回调函数（使用 contextvars 实现请求隔离） - """ - callbacks = _event_callbacks.get() - return callbacks.copy() if callbacks else [] - - -def extract_function_description(func: Callable) -> str: - """ - 提取函数的描述（docstring的第一行或前100个字符） - """ - if func.__doc__: - doc = func.__doc__.strip() - # 取第一行或前100个字符 - first_line = doc.split('\n')[0].strip() - return first_line[:100] if len(first_line) > 100 else first_line - return "" - - -def _extract_details_from_result(result: Any) -> str: - """ - 从函数返回值中提取 details，顺序固定为：先 details 再中文。 - 1) 先看返回的 dict 里是否有「details 类」字段（__details__、_details、details 或键名含 "details"）， - 且其值为非空，则用该值； - 2) 若没有或为空，再走原本的中文兜底：从返回结构中找一段包含中文的内容作为 details。 - """ - default = "完成" - if not result: - return default - - # 第一步：优先用 details 类字段（仅当有值且非空时采用） - if isinstance(result, dict): - for key in ("__details__", "_details", "details"): - if key in result and result[key] is not None: - s = str(result[key]).strip() - if s: - return str(result[key]) - for k, v in result.items(): - if "details" in k.lower() and v is not None: - s = str(v).strip() - if s: - return str(v) - - # 第二步：保留原本的中文兜底 - def _contains_chinese(text: str) -> bool: - if not text: - return False - return any("\u4e00" <= c <= "\u9fff" for c in str(text)) - - def _find_chinese(obj: Any, max_length: int = 200) -> Optional[str]: - if isinstance(obj, str): - return (obj[:max_length] + "...") if _contains_chinese(obj) and len(obj) > max_length else ( - obj if _contains_chinese(obj) else None) - if isinstance(obj, dict): - for key, value in obj.items(): - if "details" in key.lower(): - continue - found = _find_chinese(value, max_length) - if found: - return found - if isinstance(obj, list): - for item in obj: - found = _find_chinese(item, max_length) - if found: - return found - return None - - chinese = _find_chinese(result) - return chinese if chinese else default - - -def track_function_calls(func: Callable) -> Callable: - """ - 装饰器：追踪函数调用过程，支持同步和异步函数 - - Args: - func: 被装饰的函数 - """ - - @functools.wraps(func) - def sync_wrapper(*args, **kwargs): - # 获取函数签名信息 - sig = inspect.signature(func) - bound_args = sig.bind(*args, **kwargs) - bound_args.apply_defaults() - - func_name = func.__name__ - module_name = func.__module__ - - # 尝试从函数参数中获取 state - state = None - if args and isinstance(args[0], dict): - state = args[0] - elif 'state' in bound_args.arguments: - state = bound_args.arguments.get('state') - - func_description = extract_function_description(func) - - # 准备函数参数信息 - params = {k: v for k, v in bound_args.arguments.items()} - - try: - # 执行原函数 - result = func(*args, **kwargs) - - # 构建 details：优先用返回值中的 details 类字段（__details__、_details、details 或键名含 details） - details = _extract_details_from_result(result) - - # 合并函数执行事件（只在执行完成后发送一次） - execution_event = { - "type": "function_execution", - "title": func_description, - "details": details, - "result": result, # 添加函数返回结果 - "timestamp": datetime.now().isoformat() - } - - # 调用所有注册的回调（从上下文变量获取） - callbacks = _event_callbacks.get() or [] - for callback in callbacks: - try: - callback(execution_event) - except Exception as e: - print(f"回调执行出错: {e}") - - return result - - except Exception as e: - # 发送函数执行失败事件 - error_event = { - "type": "function_error", - "title": func_description, - "details": f"错误: {str(e)}", - "timestamp": datetime.now().isoformat() - } - - callbacks = _event_callbacks.get() or [] - for callback in callbacks: - try: - callback(error_event) - except Exception as ex: - print(f"回调执行出错: {ex}") - - raise - - @functools.wraps(func) - async def async_wrapper(*args, **kwargs): - # 获取函数签名信息 - sig = inspect.signature(func) - bound_args = sig.bind(*args, **kwargs) - bound_args.apply_defaults() - - func_name = func.__name__ - module_name = func.__module__ - - # 尝试从函数参数中获取 state - state = None - if args and isinstance(args[0], dict): - state = args[0] - elif 'state' in bound_args.arguments: - state = bound_args.arguments.get('state') - - func_description = extract_function_description(func) - - # 准备函数参数信息 - params = {k: v for k, v in bound_args.arguments.items()} - - try: - # 执行原函数 - result = await func(*args, **kwargs) - - # 构建 details：优先用返回值中的 details 类字段（__details__、_details、details 或键名含 details） - details = _extract_details_from_result(result) - - # 合并函数执行事件（只在执行完成后发送一次） - execution_event = { - "type": "function_execution", - "title": func_description, - "details": details, - "result": result, # 添加函数返回结果 - "timestamp": datetime.now().isoformat() - } - - # 调用所有注册的回调（从上下文变量获取） - callbacks = _event_callbacks.get() or [] - for callback in callbacks: - try: - callback(execution_event) - except Exception as e: - print(f"回调执行出错: {e}") - - return result - - except Exception as e: - # 发送函数执行失败事件 - error_event = { - "type": "function_error", - "title": func_description, - "details": f"错误: {str(e)}", - "timestamp": datetime.now().isoformat() - } - - callbacks = _event_callbacks.get() or [] - for callback in callbacks: - try: - callback(error_event) - except Exception as ex: - print(f"回调执行出错: {ex}") - - raise - - # 根据函数是否为协程函数返回对应的装饰器 - if asyncio.iscoroutinefunction(func): - return async_wrapper - else: - return sync_wrapper - - -# 事件处理器 -def console_event_handler(event: Dict[str, Any]): - """ - 控制台事件处理器 - """ - event_type = event["type"] - title = event.get("title", "") - details = event.get("details", "") - - if event_type == "function_execution": - print(f"✅ 函数执行: {title}") - if details: - print(f" 详情: {details}") - - elif event_type == "function_error": - print(f"❌ 函数错误: {title}") - if details: - print(f" 详情: {details}") - - -class StreamEventHandler: - """ - 流式事件处理器，将装饰器事件放入队列供流式输出使用 - """ - def __init__(self, queue: asyncio.Queue): - self.queue = queue - - def __call__(self, event: Dict[str, Any]): - """ - 处理事件，将事件放入队列 - 注意：这是同步方法，但队列是异步的，需要特殊处理 - """ - try: - # 使用线程安全的方式将事件放入队列 - # 如果队列已满，使用 put_nowait 并忽略错误 - try: - self.queue.put_nowait(event) - except asyncio.QueueFull: - # 队列已满，跳过此事件（不应该发生，因为队列大小足够） - pass - except Exception as e: - # 忽略错误，避免影响主流程 - pass - - def send_stream_content(self, content: str, title: str = ""): - """ - 发送流式内容事件 - - Args: - content: 流式内容片段 - title: 函数描述/标题（可选） - """ - try: - event = { - "type": "stream_content", - "content": content, - "title": title, - "timestamp": datetime.now().isoformat() - } - self.queue.put_nowait(event) - except Exception as e: - # 忽略错误，避免影响主流程 - pass - - -def create_stream_event_handler() -> tuple[StreamEventHandler, asyncio.Queue]: - """ - 创建流式事件处理器和对应的队列（使用 contextvars 实现请求隔离） - - Returns: - (handler, queue): 事件处理器和事件队列 - """ - queue = asyncio.Queue(maxsize=1000) # 设置足够大的队列大小 - handler = StreamEventHandler(queue) - _current_stream_handler.set(handler) # 保存到上下文变量 - return handler, queue - - -def get_current_stream_handler() -> Optional[StreamEventHandler]: - """ - 获取当前活动的流式事件处理器（使用 contextvars 实现请求隔离） - - Returns: - 当前流式事件处理器，如果没有则返回 None - """ - return _current_stream_handler.get() - - -def clear_current_stream_handler(): - """ - 清除当前活动的流式事件处理器（使用 contextvars 实现请求隔离） - """ - _current_stream_handler.set(None) +""" +使用装饰器实现全栈函数调用追踪 +解决LangGraph无法捕获子函数内部状态的问题 +""" + +import functools +import inspect +import asyncio +import contextvars +from typing import Callable, Any, Dict, Optional, List +from datetime import datetime +from collections import deque + +# 使用 contextvars 实现请求级别的隔离，避免多个并发请求互相干扰 +_event_callbacks: contextvars.ContextVar[List[Callable]] = contextvars.ContextVar('event_callbacks', default=None) +_current_stream_handler: contextvars.ContextVar[Optional[Any]] = contextvars.ContextVar('current_stream_handler', + default=None) + + +def register_event_callback(callback: Callable[[Dict[str, Any]], None]): + """ + 注册事件回调函数（使用 contextvars 实现请求隔离） + """ + callbacks = _event_callbacks.get() + if callbacks is None: + callbacks = [] + _event_callbacks.set(callbacks) + callbacks.append(callback) + + +def unregister_event_callback(callback: Callable[[Dict[str, Any]], None]): + """ + 移除事件回调函数（使用 contextvars 实现请求隔离） + """ + callbacks = _event_callbacks.get() + if callbacks and callback in callbacks: + callbacks.remove(callback) + + +def clear_event_callbacks(): + """ + 清空所有事件回调（使用 contextvars 实现请求隔离） + """ + callbacks = _event_callbacks.get() + if callbacks: + callbacks.clear() + + +def get_all_callbacks(): + """ + 获取所有注册的回调函数（使用 contextvars 实现请求隔离） + """ + callbacks = _event_callbacks.get() + return callbacks.copy() if callbacks else [] + + +def extract_function_description(func: Callable) -> str: + """ + 提取函数的描述（docstring的第一行或前100个字符） + """ + if func.__doc__: + doc = func.__doc__.strip() + # 取第一行或前100个字符 + first_line = doc.split('\n')[0].strip() + return first_line[:100] if len(first_line) > 100 else first_line + return "" + + +def _extract_details_from_result(result: Any) -> str: + """ + 从函数返回值中提取 details，顺序固定为：先 details 再中文。 + 1) 先看返回的 dict 里是否有「details 类」字段（__details__、_details、details 或键名含 "details"）， + 且其值为非空，则用该值； + 2) 若没有或为空，再走原本的中文兜底：从返回结构中找一段包含中文的内容作为 details。 + """ + default = "完成" + if not result: + return default + + # 第一步：优先用 details 类字段（仅当有值且非空时采用） + if isinstance(result, dict): + for key in ("__details__", "_details", "details"): + if key in result and result[key] is not None: + s = str(result[key]).strip() + if s: + return str(result[key]) + for k, v in result.items(): + if "details" in k.lower() and v is not None: + s = str(v).strip() + if s: + return str(v) + + # 第二步：保留原本的中文兜底 + def _contains_chinese(text: str) -> bool: + if not text: + return False + return any("\u4e00" <= c <= "\u9fff" for c in str(text)) + + def _find_chinese(obj: Any, max_length: int = 200) -> Optional[str]: + if isinstance(obj, str): + return (obj[:max_length] + "...") if _contains_chinese(obj) and len(obj) > max_length else ( + obj if _contains_chinese(obj) else None) + if isinstance(obj, dict): + for key, value in obj.items(): + if "details" in key.lower(): + continue + found = _find_chinese(value, max_length) + if found: + return found + if isinstance(obj, list): + for item in obj: + found = _find_chinese(item, max_length) + if found: + return found + return None + + chinese = _find_chinese(result) + return chinese if chinese else default + + +def track_function_calls(func: Callable) -> Callable: + """ + 装饰器：追踪函数调用过程，支持同步和异步函数 + + Args: + func: 被装饰的函数 + """ + + @functools.wraps(func) + def sync_wrapper(*args, **kwargs): + # 获取函数签名信息 + sig = inspect.signature(func) + bound_args = sig.bind(*args, **kwargs) + bound_args.apply_defaults() + + func_name = func.__name__ + module_name = func.__module__ + + # 尝试从函数参数中获取 state + state = None + if args and isinstance(args[0], dict): + state = args[0] + elif 'state' in bound_args.arguments: + state = bound_args.arguments.get('state') + + func_description = extract_function_description(func) + + # 准备函数参数信息 + params = {k: v for k, v in bound_args.arguments.items()} + + try: + # 执行原函数 + result = func(*args, **kwargs) + + # 构建 details：优先用返回值中的 details 类字段（__details__、_details、details 或键名含 details） + details = _extract_details_from_result(result) + + # 合并函数执行事件（只在执行完成后发送一次） + execution_event = { + "type": "function_execution", + "title": func_description, + "details": details, + "result": result, # 添加函数返回结果 + "timestamp": datetime.now().isoformat() + } + + # 调用所有注册的回调（从上下文变量获取） + callbacks = _event_callbacks.get() or [] + for callback in callbacks: + try: + callback(execution_event) + except Exception as e: + print(f"回调执行出错: {e}") + + return result + + except Exception as e: + # 发送函数执行失败事件 + error_event = { + "type": "function_error", + "title": func_description, + "details": f"错误: {str(e)}", + "timestamp": datetime.now().isoformat() + } + + callbacks = _event_callbacks.get() or [] + for callback in callbacks: + try: + callback(error_event) + except Exception as ex: + print(f"回调执行出错: {ex}") + + raise + + @functools.wraps(func) + async def async_wrapper(*args, **kwargs): + # 获取函数签名信息 + sig = inspect.signature(func) + bound_args = sig.bind(*args, **kwargs) + bound_args.apply_defaults() + + func_name = func.__name__ + module_name = func.__module__ + + # 尝试从函数参数中获取 state + state = None + if args and isinstance(args[0], dict): + state = args[0] + elif 'state' in bound_args.arguments: + state = bound_args.arguments.get('state') + + func_description = extract_function_description(func) + + # 准备函数参数信息 + params = {k: v for k, v in bound_args.arguments.items()} + + try: + # 执行原函数 + result = await func(*args, **kwargs) + + # 构建 details：优先用返回值中的 details 类字段（__details__、_details、details 或键名含 details） + details = _extract_details_from_result(result) + + # 合并函数执行事件（只在执行完成后发送一次） + execution_event = { + "type": "function_execution", + "title": func_description, + "details": details, + "result": result, # 添加函数返回结果 + "timestamp": datetime.now().isoformat() + } + + # 调用所有注册的回调（从上下文变量获取） + callbacks = _event_callbacks.get() or [] + for callback in callbacks: + try: + callback(execution_event) + except Exception as e: + print(f"回调执行出错: {e}") + + return result + + except Exception as e: + # 发送函数执行失败事件 + error_event = { + "type": "function_error", + "title": func_description, + "details": f"错误: {str(e)}", + "timestamp": datetime.now().isoformat() + } + + callbacks = _event_callbacks.get() or [] + for callback in callbacks: + try: + callback(error_event) + except Exception as ex: + print(f"回调执行出错: {ex}") + + raise + + # 根据函数是否为协程函数返回对应的装饰器 + if asyncio.iscoroutinefunction(func): + return async_wrapper + else: + return sync_wrapper + + +# 事件处理器 +def console_event_handler(event: Dict[str, Any]): + """ + 控制台事件处理器 + """ + event_type = event["type"] + title = event.get("title", "") + details = event.get("details", "") + + if event_type == "function_execution": + print(f"✅ 函数执行: {title}") + if details: + print(f" 详情: {details}") + + elif event_type == "function_error": + print(f"❌ 函数错误: {title}") + if details: + print(f" 详情: {details}") + + +class StreamEventHandler: + """ + 流式事件处理器，将装饰器事件放入队列供流式输出使用 + """ + def __init__(self, queue: asyncio.Queue): + self.queue = queue + + def __call__(self, event: Dict[str, Any]): + """ + 处理事件，将事件放入队列 + 注意：这是同步方法，但队列是异步的，需要特殊处理 + """ + try: + # 使用线程安全的方式将事件放入队列 + # 如果队列已满，使用 put_nowait 并忽略错误 + try: + self.queue.put_nowait(event) + except asyncio.QueueFull: + # 队列已满，跳过此事件（不应该发生，因为队列大小足够） + pass + except Exception as e: + # 忽略错误，避免影响主流程 + pass + + def send_stream_content(self, content: str, title: str = ""): + """ + 发送流式内容事件 + + Args: + content: 流式内容片段 + title: 函数描述/标题（可选） + """ + try: + event = { + "type": "stream_content", + "content": content, + "title": title, + "timestamp": datetime.now().isoformat() + } + self.queue.put_nowait(event) + except Exception as e: + # 忽略错误，避免影响主流程 + pass + + +def create_stream_event_handler() -> tuple[StreamEventHandler, asyncio.Queue]: + """ + 创建流式事件处理器和对应的队列（使用 contextvars 实现请求隔离） + + Returns: + (handler, queue): 事件处理器和事件队列 + """ + queue = asyncio.Queue(maxsize=1000) # 设置足够大的队列大小 + handler = StreamEventHandler(queue) + _current_stream_handler.set(handler) # 保存到上下文变量 + return handler, queue + + +def get_current_stream_handler() -> Optional[StreamEventHandler]: + """ + 获取当前活动的流式事件处理器（使用 contextvars 实现请求隔离） + + Returns: + 当前流式事件处理器，如果没有则返回 None + """ + return _current_stream_handler.get() + + +def clear_current_stream_handler(): + """ + 清除当前活动的流式事件处理器（使用 contextvars 实现请求隔离） + """ + _current_stream_handler.set(None) diff --git a/utils/source_citations.py b/utils/source_citations.py new file mode 100644 index 0000000..c072972 --- /dev/null +++ b/utils/source_citations.py @@ -0,0 +1,71 @@ +"""Utilities for accumulating RAG source citations across multiple searches.""" + +import json +from typing import Any, Dict + + +def _citation_item_key(item: Any) -> str: + """Build a stable identity for one RAG citation chunk.""" + if not isinstance(item, dict): + return json.dumps(item, ensure_ascii=False, sort_keys=True, default=str) + + chunk_id = item.get("id") + if chunk_id not in (None, ""): + return f"id:{chunk_id}" + + location = { + "file_id": item.get("file_id"), + "page_idx": item.get("page_idx"), + "positions": item.get("positions"), + } + if any(value not in (None, "", []) for value in location.values()): + return "location:" + json.dumps( + location, ensure_ascii=False, sort_keys=True, default=str + ) + + # index/score may differ between rewritten queries and do not identify a source. + fallback = { + key: value + for key, value in item.items() + if key not in {"index", "score", "text"} + } + return "fallback:" + json.dumps( + fallback, ensure_ascii=False, sort_keys=True, default=str + ) + + +def merge_rag_source_citations( + accumulated: Dict[str, Any], incoming: Any +) -> Dict[str, Any]: + """ + Append a RAG call's citations without losing earlier calls or duplicating chunks. + + Citation text is intentionally removed here because the stream response has never + exposed it; the full text remains available inside the workflow for answer generation. + """ + if not isinstance(incoming, dict) or not incoming: + return accumulated + + for doc_name, chunks in incoming.items(): + if isinstance(chunks, list): + target = accumulated.setdefault(doc_name, []) + if not isinstance(target, list): + # Be defensive about malformed/legacy data while preserving both values. + target = [target] + accumulated[doc_name] = target + + seen = {_citation_item_key(item) for item in target} + for item in chunks: + sanitized_item = ( + {key: value for key, value in item.items() if key != "text"} + if isinstance(item, dict) + else item + ) + item_key = _citation_item_key(sanitized_item) + if item_key not in seen: + target.append(sanitized_item) + seen.add(item_key) + elif doc_name not in accumulated: + accumulated[doc_name] = chunks + + return accumulated diff --git a/workflow_registry.py b/workflow_registry.py index b1468a0..ac2bfb0 100644 --- a/workflow_registry.py +++ b/workflow_registry.py @@ -1,158 +1,158 @@ -""" -工作流注册表 - 主代理与 app 共用 -主代理阶段四根据此配置路由到所有注册的子 agent； -app 据此执行对应工作流。 -添加新工作流时，在此配置 WORKFLOW_CONFIG 与 ROUTE_DESCRIPTIONS。 -""" -# ============================================================================ -# =============== 工作流配置区域 - 添加新工作流时只需修改这里 =============== -# ============================================================================ -# 格式： -# "route_flag": { -# "module": "workflows.workflow_xxx", -# "function": "run_xxx_workflow", -# "is_async": True/False, -# } -# ============================================================================ - -WORKFLOW_CONFIG = { - "fault_diagnosis": { - "module": "workflows.workflow_fault_diagnosis", - "function": "run_fault_diagnosis_workflow", - "is_async": True, - }, - "qa": { - "module": "workflows.workflow_qa", - "function": "run_qa_workflow", - "is_async": True, - }, - "问题排查": { - "module": "workflows.workflow_fault_diagnosis", - "function": "run_fault_diagnosis_workflow", - "is_async": True, - }, - "spare_parts_statistics": { - "module": "workflows.workflow_spare_parts_statistics", - "function": "run_spare_parts_statistics_workflow", - "is_async": True, - }, - "rules": { - "module": "workflows.workflow_qa", - "function": "run_qa_workflow", - "is_async": True, - }, - "other": { - "module": "workflows.workflow_fault_diagnosis", - "function": "run_fault_diagnosis_workflow", - "is_async": True, -# -# "module": "workflows.workflow_other", -# "function": "run_other_workflow", -# "is_async": True, - }, - "report":{ -# "module": "workflows.workflow_qa_report", -# "function": "run_qa_report_agent", -# "is_async": True, - "module": "workflows.workflow_fault_diagnosis", - "function": "run_fault_diagnosis_workflow", - "is_async": True, - }, - "百科问答":{ - "module": "workflows.workflow_baike", - "function": "run_baike_workflow", - "is_async": True, - }, - "操作使用":{ - "module": "workflows.workflow_operate", - "function": "run_operate_workflow", - "is_async": True, - }, - "统计":{ - "module": "workflows.workflow_statistics", - "function": "run_statistics_workflow", - "is_async": True, - } - -} - -# 路由标志 -> 任务分类描述（供主代理阶段四 LLM 使用） -ROUTE_DESCRIPTIONS = { - "fault_diagnosis": """ - - 特征：用户描述**当前或近期发生的设备异常、报警、性能下降等具体故障现象**，需要分析原因或定位问题。 - - 关键词：报警、异常、异响、冒烟、过热、无法启动、突然停机、振动大、压力低、温度高、故障代码、不工作 - - 示例： - • 主机滑油压力突然降到0.2MPa并报警 - • 柴油机启动后冒黑烟且转速不稳 - • 舵机发出异常噪音，疑似齿轮损坏 - - 注意：若历史记录中包含“请补充”、“【需要更多信息】”，通常属于此类，因已聚焦具体故障场景。 - 当用户问题中只有舷号、设备或故障现象时属于此类，舷号一般为连续的阿拉伯数字如“101”,"111","122","163"，"16","17","554","602","545","985","31","3261","799","905","781"等,或者数字后加舰，比如“101舰","199舰","18舰"等。 - """, - - "qa": """ - - 特征：询问定义、原理、结构、功能、分类、标准配置、常见类型、安全规范等**一般性知识**；也包括对专业术语、现象、部件作用的解释。 - **即使问题表述模糊、不完整、缺少上下文，只要涉及船舶、机械、电气、动力等技术内容，均优先归为此类。** - - 关键词：什么是、如何工作、为什么、请问、介绍、解释、有哪些、常见、典型、一般、作用、组成、是不是、叫什么 - - 示例： - • 船舶主发动机有哪些常见故障类型？ - • 涡轮增压器的工作原理是什么？ - • 主机由哪些主要部件组成？ - • 什么是MAN B&W柴油机？ - - 注意： - • **本类作为技术相关问题的默认兜底类别**——凡涉及设备、系统、部件、工况、参数等技术语境的问题，若无法明确归入 fault_diagnosis / repair / spare_parts_statistics / rules，则优先视为 qa。 - """, - - "repair": """ - - 特征：用户明确请求**维修方法、操作步骤、检修流程、更换方案或现场处理建议**，通常在已知故障或需预防性维护时提出。 - - 关键词：怎么修、如何维修、维修步骤、检修方法、更换、拆卸、安装、处理、修复、维护、保养、调整 - - 示例： - • 主机喷油器漏油，怎么更换？ - • 给出主机缸套磨损的维修方案 - • 如何拆卸增压器进行检修？ - """, - - "spare_parts_statistics": """ - - 特征：用户需要对**备品备件的数量、使用记录、库存、消耗趋势、预测需求或成本分析**进行查询或统计。 - - 关键词：备件、备品、库存、数量、统计、预测、分析、消耗、需求、清单、台账、寿命、更换周期、用了多少、还剩多少 - - 示例： - • 统计过去一年主机喷油嘴的使用数量 - • 预测下季度所需空冷器备件数量 - • 主机活塞环的平均更换周期是多少？ - - 注意：若仅问“某个部件是不是备件”或“这个备件叫什么名字”，属于知识问答（qa），而非本类。 - """, - - "rules": """ - - 特征：用户询问与船舶设备维修、检验、安全、环保、保密等相关的**国家法规、军用标准（如GJB）、行业规范、船级社要求或管理制度**。 - - 关键词：法规、标准、规范、规定、合规、条例、依据、要求、准则、制度、认证、资质、船级社、CCS、IMO、SOLAS、MARPOL、资格 - - 示例： - • 船舶主机修理需遵循哪些GJB标准？ - • IMO对主机排放有哪些强制要求？ - • 舰船维修单位需要哪些资质？ - • 保密资格申请流程是什么？ - - 注意：若问题提到“标准里有没有…”但未指明具体法规/标准名称，且无“法规”“标准”等关键词，可先归为 qa；一旦明确提及“法规”“标准”“规范”“SOLAS”等，则归为 rules。 - """, - "other": """ - - 特征：**完全不涉及新的技术问题、故障诊断、维修请求、备件统计、法规查询或报告生成**，而是针对**已有对话内容、方案、步骤或文本的反馈、修正、补充、顺序调整或格式优化**。典型场景包括： - • 指出之前提供的操作步骤有误（如“异物清理步骤有误”） - • 要求调整操作顺序（如“这一步应当先怎样，然后再怎么操作，最后……”） - • 对生成内容提出修改意见（如“步骤有误/有补充”“请重写第三点”） - • 请求优化语言、格式或结构（如“请用更简洁的方式表达”） - • 闲聊、非技术性确认或与当前技术任务无关的交互 - - - 注意： - • **只要用户意图是修正、补充或调整已有内容，而非提出新问题，即使涉及技术术语，也应归为 other**。 - • 不要将此类反馈误判为 repair 或 qa——它们不是在问“该怎么修”或“原理是什么”，而是在说“你刚才说的不对/不全”。 - • 此类通常出现在多轮对话中，是对前文输出的回应。 - """, - "report": """ - - 特征：用户明确要求**生成、整理、导出或汇总**某种形式的文档、报表、清单或总结材料。通常涉及将分散的故障记录、维修历史、备件数据或问答内容整合成结构化文本（如日报、周报、故障分析报告、维修总结、备件消耗表等）。 - - 关键词：生成报告、写总结、导出表格、整理清单、汇总数据、形成文档、日报、周报、月报、分析报告、维修记录单、台账报表 - - 示例： - • 请根据本周的故障记录生成一份主机系统运行分析报告 - • 总结这次对话 - • 导出过去一年的备件消耗统计报告 - • 帮我写一份关于此次主机停机事故的详细处理总结 - """ -} - -VALID_ROUTE_FLAGS = tuple(WORKFLOW_CONFIG.keys()) +""" +工作流注册表 - 主代理与 app 共用 +主代理阶段四根据此配置路由到所有注册的子 agent； +app 据此执行对应工作流。 +添加新工作流时，在此配置 WORKFLOW_CONFIG 与 ROUTE_DESCRIPTIONS。 +""" +# ============================================================================ +# =============== 工作流配置区域 - 添加新工作流时只需修改这里 =============== +# ============================================================================ +# 格式： +# "route_flag": { +# "module": "workflows.workflow_xxx", +# "function": "run_xxx_workflow", +# "is_async": True/False, +# } +# ============================================================================ + +WORKFLOW_CONFIG = { + "fault_diagnosis": { + "module": "workflows.workflow_fault_diagnosis", + "function": "run_fault_diagnosis_workflow", + "is_async": True, + }, + "qa": { + "module": "workflows.workflow_qa", + "function": "run_qa_workflow", + "is_async": True, + }, + "问题排查": { + "module": "workflows.workflow_fault_diagnosis", + "function": "run_fault_diagnosis_workflow", + "is_async": True, + }, + "spare_parts_statistics": { + "module": "workflows.workflow_spare_parts_statistics", + "function": "run_spare_parts_statistics_workflow", + "is_async": True, + }, + "rules": { + "module": "workflows.workflow_qa", + "function": "run_qa_workflow", + "is_async": True, + }, + "other": { + "module": "workflows.workflow_fault_diagnosis", + "function": "run_fault_diagnosis_workflow", + "is_async": True, +# +# "module": "workflows.workflow_other", +# "function": "run_other_workflow", +# "is_async": True, + }, + "report":{ +# "module": "workflows.workflow_qa_report", +# "function": "run_qa_report_agent", +# "is_async": True, + "module": "workflows.workflow_fault_diagnosis", + "function": "run_fault_diagnosis_workflow", + "is_async": True, + }, + "百科问答":{ + "module": "workflows.workflow_baike", + "function": "run_baike_workflow", + "is_async": True, + }, + "操作使用":{ + "module": "workflows.workflow_operate", + "function": "run_operate_workflow", + "is_async": True, + }, + "统计":{ + "module": "workflows.workflow_statistics", + "function": "run_statistics_workflow", + "is_async": True, + } + +} + +# 路由标志 -> 任务分类描述（供主代理阶段四 LLM 使用） +ROUTE_DESCRIPTIONS = { + "fault_diagnosis": """ + - 特征：用户描述**当前或近期发生的设备异常、报警、性能下降等具体故障现象**，需要分析原因或定位问题。 + - 关键词：报警、异常、异响、冒烟、过热、无法启动、突然停机、振动大、压力低、温度高、故障代码、不工作 + - 示例： + • 主机滑油压力突然降到0.2MPa并报警 + • 柴油机启动后冒黑烟且转速不稳 + • 舵机发出异常噪音，疑似齿轮损坏 + - 注意：若历史记录中包含“请补充”、“【需要更多信息】”，通常属于此类，因已聚焦具体故障场景。 + 当用户问题中只有舷号、设备或故障现象时属于此类，舷号一般为连续的阿拉伯数字如“101”,"111","122","163"，"16","17","554","602","545","985","31","3261","799","905","781"等,或者数字后加舰，比如“101舰","199舰","18舰"等。 + """, + + "qa": """ + - 特征：询问定义、原理、结构、功能、分类、标准配置、常见类型、安全规范等**一般性知识**；也包括对专业术语、现象、部件作用的解释。 + **即使问题表述模糊、不完整、缺少上下文，只要涉及船舶、机械、电气、动力等技术内容，均优先归为此类。** + - 关键词：什么是、如何工作、为什么、请问、介绍、解释、有哪些、常见、典型、一般、作用、组成、是不是、叫什么 + - 示例： + • 船舶主发动机有哪些常见故障类型？ + • 涡轮增压器的工作原理是什么？ + • 主机由哪些主要部件组成？ + • 什么是MAN B&W柴油机？ + - 注意： + • **本类作为技术相关问题的默认兜底类别**——凡涉及设备、系统、部件、工况、参数等技术语境的问题，若无法明确归入 fault_diagnosis / repair / spare_parts_statistics / rules，则优先视为 qa。 + """, + + "repair": """ + - 特征：用户明确请求**维修方法、操作步骤、检修流程、更换方案或现场处理建议**，通常在已知故障或需预防性维护时提出。 + - 关键词：怎么修、如何维修、维修步骤、检修方法、更换、拆卸、安装、处理、修复、维护、保养、调整 + - 示例： + • 主机喷油器漏油，怎么更换？ + • 给出主机缸套磨损的维修方案 + • 如何拆卸增压器进行检修？ + """, + + "spare_parts_statistics": """ + - 特征：用户需要对**备品备件的数量、使用记录、库存、消耗趋势、预测需求或成本分析**进行查询或统计。 + - 关键词：备件、备品、库存、数量、统计、预测、分析、消耗、需求、清单、台账、寿命、更换周期、用了多少、还剩多少 + - 示例： + • 统计过去一年主机喷油嘴的使用数量 + • 预测下季度所需空冷器备件数量 + • 主机活塞环的平均更换周期是多少？ + - 注意：若仅问“某个部件是不是备件”或“这个备件叫什么名字”，属于知识问答（qa），而非本类。 + """, + + "rules": """ + - 特征：用户询问与船舶设备维修、检验、安全、环保、保密等相关的**国家法规、军用标准（如GJB）、行业规范、船级社要求或管理制度**。 + - 关键词：法规、标准、规范、规定、合规、条例、依据、要求、准则、制度、认证、资质、船级社、CCS、IMO、SOLAS、MARPOL、资格 + - 示例： + • 船舶主机修理需遵循哪些GJB标准？ + • IMO对主机排放有哪些强制要求？ + • 舰船维修单位需要哪些资质？ + • 保密资格申请流程是什么？ + - 注意：若问题提到“标准里有没有…”但未指明具体法规/标准名称，且无“法规”“标准”等关键词，可先归为 qa；一旦明确提及“法规”“标准”“规范”“SOLAS”等，则归为 rules。 + """, + "other": """ + - 特征：**完全不涉及新的技术问题、故障诊断、维修请求、备件统计、法规查询或报告生成**，而是针对**已有对话内容、方案、步骤或文本的反馈、修正、补充、顺序调整或格式优化**。典型场景包括： + • 指出之前提供的操作步骤有误（如“异物清理步骤有误”） + • 要求调整操作顺序（如“这一步应当先怎样，然后再怎么操作，最后……”） + • 对生成内容提出修改意见（如“步骤有误/有补充”“请重写第三点”） + • 请求优化语言、格式或结构（如“请用更简洁的方式表达”） + • 闲聊、非技术性确认或与当前技术任务无关的交互 + + - 注意： + • **只要用户意图是修正、补充或调整已有内容，而非提出新问题，即使涉及技术术语，也应归为 other**。 + • 不要将此类反馈误判为 repair 或 qa——它们不是在问“该怎么修”或“原理是什么”，而是在说“你刚才说的不对/不全”。 + • 此类通常出现在多轮对话中，是对前文输出的回应。 + """, + "report": """ + - 特征：用户明确要求**生成、整理、导出或汇总**某种形式的文档、报表、清单或总结材料。通常涉及将分散的故障记录、维修历史、备件数据或问答内容整合成结构化文本（如日报、周报、故障分析报告、维修总结、备件消耗表等）。 + - 关键词：生成报告、写总结、导出表格、整理清单、汇总数据、形成文档、日报、周报、月报、分析报告、维修记录单、台账报表 + - 示例： + • 请根据本周的故障记录生成一份主机系统运行分析报告 + • 总结这次对话 + • 导出过去一年的备件消耗统计报告 + • 帮我写一份关于此次主机停机事故的详细处理总结 + """ +} + +VALID_ROUTE_FLAGS = tuple(WORKFLOW_CONFIG.keys()) diff --git a/workflows/__init__.py b/workflows/__init__.py index 12a2821..be63ba4 100644 --- a/workflows/__init__.py +++ b/workflows/__init__.py @@ -1,20 +1,20 @@ -""" -工作流模块 -包含所有工作流Agent -""" -from .workflow_fault_diagnosis import run_fault_diagnosis_workflow - -# 其他工作流暂未实现，待后续添加 -# from .workflow_repair import run_repair_workflow -# from .workflow_statistics import run_statistics_workflow -# from .workflow_qa import run_qa_workflow -# from .workflow_report import run_report_workflow - -__all__ = [ - "run_fault_diagnosis_workflow", - # "run_repair_workflow", - # "run_statistics_workflow", - # "run_qa_workflow", - # "run_report_workflow" -] - +""" +工作流模块 +包含所有工作流Agent +""" +from .workflow_fault_diagnosis import run_fault_diagnosis_workflow + +# 其他工作流暂未实现，待后续添加 +# from .workflow_repair import run_repair_workflow +# from .workflow_statistics import run_statistics_workflow +# from .workflow_qa import run_qa_workflow +# from .workflow_report import run_report_workflow + +__all__ = [ + "run_fault_diagnosis_workflow", + # "run_repair_workflow", + # "run_statistics_workflow", + # "run_qa_workflow", + # "run_report_workflow" +] + diff --git a/workflows/workflow_baike.py b/workflows/workflow_baike.py index 1b00de5..ca887f3 100644 --- a/workflows/workflow_baike.py +++ b/workflows/workflow_baike.py @@ -1,546 +1,544 @@ """ -工作流：普通问答Agent（使用LangGraph） -输入：文本（转化后和原文本query）、图片描述、语音文本 -步骤： -1. 使用RAG和GraphRAG检索相关信息 -2. 生成回答 -使用 OpenaiAPI 方法调用大模型，尽可能使用异步 +百科问答工作流。 + +核心链路借鉴 WeKnora： +1. 意图判断默认偏向检索 +2. 基于历史重写检索 query，并生成多个检索变体 +3. RAG / Graph 并行召回 +4. 多源 evidence 去重、融合排序、rerank +5. 只基于 evidence 生成可追溯回答 """ -import sys -import os - -# 添加项目根目录到Python搜索路径 -sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -from typing import TypedDict, Optional, Dict, Any, List -from langgraph.graph import StateGraph, START, END -from workflows.history_manager import init_history, DEFAULT_HISTORY_MAX_MESSAGES -# from workflows.workflow_utils import stream_generate_and_postprocess -from workflows.workflow_utils import dedupe_rag_results, stream_generate_and_postprocess - -from tools.function_tool import ( - graph_rag_search -) -from tools.rag_tools import rag_search -from modelsAPI.model_api import OpenaiAPI -from utils.function_tracker import track_function_calls, get_all_callbacks -from utils.image_utils import extract_image_paths -from prompts import BAIKE_PROMPTS import asyncio import json -import re +import os import random +import re +import sys +from typing import Any, Dict, List, Optional, TypedDict + +from langgraph.graph import END, START, StateGraph + +sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from modelsAPI.model_api import OpenaiAPI +from prompts import BAIKE_PROMPTS +from tools.function_tool import graph_rag_search +from tools.rag_tools import rag_search +from utils.function_tracker import get_all_callbacks, track_function_calls +from utils.image_utils import extract_image_paths +from workflows.history_manager import DEFAULT_HISTORY_MAX_MESSAGES, init_history +from workflows.workflow_utils import dedupe_rag_results, stream_generate_and_postprocess + BAIKE_RAG_TOP_K = 8 BAIKE_GRAPH_TOP_K = 240 BAIKE_RESULT_LIMIT = 8 +BAIKE_EVIDENCE_LIMIT = 10 -# =============== 定义状态 =============== class QAState(TypedDict): - """普通问答工作流状态""" - extracted_text: str # 已提取的文本（统一接口，只接收文本） + extracted_text: str combined_query: str - retrieval_query: str # 用于RAG/图谱检索的重写后query - route_flag: str # 路由标识 - history_message: str # 历史消息（JSON格式） - history_messages: List[Dict[str, Any]] # 过滤后的历史消息列表 - need_retrieval: Optional[bool] # 是否需要检索 + retrieval_query: str + retrieval_queries: List[str] + route_flag: str + history_message: str + history_messages: List[Dict[str, Any]] + need_retrieval: Optional[bool] + rag_search_result: Optional[List[Dict[str, Any]]] + graph_search_result: Optional[str] + evidence_items: Optional[List[Dict[str, Any]]] + response: str + actions: List[Dict[str, Any]] + suggestedReplies: List[Dict[str, Any]] + error_message: str - # RAG检索结果 - rag_search_result: Optional[List[Dict[str, Any]]] # RAG检索结果 - graph_search_result: Optional[str] # 图谱检索结果 - - # 最终响应 - response: str # 最终响应 - actions: List[Dict[str, Any]] # 按钮信息列表 [{"type": "...", "label": "...", "icon": "..."}] - suggestedReplies: List[Dict[str, Any]] # 建议回复问题列表 - error_message: str # 错误信息 - - -# =============== 节点函数 =============== def get_baike_history_context(state: QAState) -> Dict[str, Any]: - """ - 知识问答步骤：获取问答类历史信息 - 问答工作流只关注历史问答记录，用于提供上下文 - """ - history_message = state.get("history_message", "") or "" - return init_history(history_message) + return init_history(state.get("history_message", "") or "") async def judge_need_retrieval(state: QAState) -> Dict[str, Any]: - """ - 判断用户问题是否需要检索知识库 - """ original_query = (state.get("combined_query") or state.get("extracted_text") or "").strip() history_messages = state.get("history_messages", []) or [] - prompt = BAIKE_PROMPTS["judge_need_retrieval"].format( - history_messages=history_messages if history_messages else '无', - original_query=original_query + history_messages=history_messages if history_messages else "无", + original_query=original_query, ) - - need_retrieval = True + + substantive_query = _looks_like_substantive_query(original_query) + need_retrieval = substantive_query try: result = await OpenaiAPI.open_api_chat_without_thinking(prompt, model=None) - if result: - result = result.strip().lower() - if result == "false": - need_retrieval = False - except Exception as e: - print(f"判断是否需要检索失败: {e}") - need_retrieval = True - - + if not substantive_query and result and result.strip().lower() == "false": + need_retrieval = False + except Exception as exc: + print(f"判断是否需要检索失败: {exc}") print(f"是否需要检索: {need_retrieval}") return {"need_retrieval": need_retrieval} +def _looks_like_substantive_query(query: str) -> bool: + text = re.sub(r"\s+", "", (query or "").strip()) + if not text: + return False + if re.fullmatch(r"(你好|您好|hi|hello|在吗|谢谢|多谢|好的|好|嗯|哦|确认|取消|不用了)[。！!?\s]*", text, re.I): + return False + if len(text) >= 4: + return True + return bool(re.search(r"[？?]|[A-Za-z0-9]{2,}", text)) + + async def rewrite_query_with_history(state: QAState) -> Dict[str, Any]: - """ - 使用历史上下文对当前问题进行query重写，生成适合RAG/图谱检索的独立查询语句。 - 只让大模型在有限的最近历史中挑选与当前问题语义相关的内容进行融合，避免简单拼接全量历史。 - """ - import json - - # 原始问题：优先使用 combined_query，其次使用 extracted_text original_query = (state.get("combined_query") or state.get("extracted_text") or "").strip() - history_messages = state.get("history_messages", []) or [] trimmed_history = history_messages[-DEFAULT_HISTORY_MAX_MESSAGES:] if history_messages else [] - - try: - history_str = json.dumps(trimmed_history, ensure_ascii=False) - except Exception: - history_str = str(trimmed_history) - - # 如果没有历史或当前问题过短，就直接跳过重写 - if not original_query: - return {"retrieval_query": original_query} - # 在构造 prompt 前，提取上一轮用户 query（如果有） + history_str = _json_dumps(trimmed_history) last_user_query = "" - if trimmed_history: - # 从后往前找最近一条 role == "user" 的消息 - for msg in reversed(trimmed_history): - if msg.get("role") == "user": - last_user_query = msg.get("content", "").strip() - break + for msg in reversed(trimmed_history): + if msg.get("role") == "user": + last_user_query = str(msg.get("content") or "").strip() + break + + if not original_query: + return {"retrieval_query": "", "retrieval_queries": []} + prompt = BAIKE_PROMPTS["rewrite_query_with_history"].format( - last_user_query=last_user_query or '（无）', + last_user_query=last_user_query or "无", history_str=history_str, - original_query=original_query + original_query=original_query, ) rewritten_query = original_query + query_variants: List[str] = [] try: - result = await OpenaiAPI.open_api_chat_without_thinking(prompt, model=None) - if result: - candidate = result.strip().strip("“”\"'") - # 只在模型返回的内容明显为单行短文本时替换，避免异常输出 - if candidate and len(candidate) <= 200 and "\n" not in candidate: - rewritten_query = candidate - except Exception as e: - print(f"query 重写失败: {e}") - - - title_options = [f"✨正在分析中，请稍等..."] - start_event = { - "type": "function_execution", - "title": random.choice(title_options), - "details": f"正在调取百科数据..." - } - for callback in get_all_callbacks(): - try: - callback(start_event) - except Exception: - pass + raw = await OpenaiAPI.open_api_chat_without_thinking(prompt, model=None, json_output=True) + data = _parse_json_object(raw) + if data: + candidate = str(data.get("rewrite_query") or "").strip() + if candidate: + rewritten_query = candidate[:180] + for q in data.get("search_queries") or []: + if isinstance(q, str) and q.strip(): + query_variants.append(q.strip()[:180]) + except Exception as exc: + print(f"query 重写失败: {exc}") + query_variants = _build_search_queries(rewritten_query, original_query, query_variants) + _emit_event("✨正在分析中，请稍等...", "正在生成检索策略...") print("原始检索query:", original_query) - print("重写后检索query:", last_user_query+rewritten_query) - - return {"retrieval_query":last_user_query+rewritten_query} + print("重写后检索query:", rewritten_query) + print("检索query变体:", query_variants) + return {"retrieval_query": rewritten_query, "retrieval_queries": query_variants} @track_function_calls async def search_rag_and_graph(state: QAState) -> Dict[str, Any]: - """ - 进行文本知识检索和图谱知识检索 - """ - kb_id = None - if state.get("route_flag") == "rules": - kb_id = '3' - - # 优先使用重写后的检索query，其次使用 combined_query，最后回退到原始 extracted_text - query = state.get("retrieval_query") or state.get("combined_query") or state.get("extracted_text", "") - if "用户问题：" in query: - query = query.split("用户问题：", 1)[1] - - print("Graph 检索query", query) - extracted_text = state.get("extracted_text", "") - if "用户问题：" in extracted_text: - extracted_text = extracted_text.split("用户问题：", 1)[1] - + kb_id = "3" if state.get("route_flag") == "rules" else None + query = (state.get("retrieval_query") or state.get("combined_query") or state.get("extracted_text") or "").strip() + extracted_text = (state.get("extracted_text") or "").strip() + query_variants = _build_search_queries(query, extracted_text, state.get("retrieval_queries") or []) try: - # 异步并行调用RAG检索和图谱检索 - print(kb_id) - print(33333333333333333333333333333333333333) - - async def _do_rag_search(): - if kb_id: - return await rag_search( - query=extracted_text, - kb_id=kb_id, - top_k=BAIKE_RAG_TOP_K - ) - else: - return await rag_search( - query=extracted_text, - top_k=BAIKE_RAG_TOP_K - ) - - async def _do_graph_search(): - return await graph_rag_search.ainvoke({ - "query": query, - "top_k": BAIKE_GRAPH_TOP_K - }) - - rag_result, graph_results = await asyncio.gather( - _do_rag_search(), - _do_graph_search() + rag_task = _search_rag_variants(query_variants[:3], kb_id=kb_id) + graph_task = graph_rag_search.ainvoke({"query": query, "top_k": BAIKE_GRAPH_TOP_K}) + rag_results, graph_results = await asyncio.gather(rag_task, graph_task) + evidence_items = _build_evidence_items( + rag_results=rag_results, + graph_results=graph_results if isinstance(graph_results, str) else "", + query=query, ) - - # 处理 RAG 结果，提取成列表格式 - rag_search_result = [] - if rag_result.get("success", False): - source_citation = rag_result.get("sourceCitation", {}) - for resource, items in source_citation.items(): - for item in items: - rag_search_result.append({ - "text": item.get("text", ""), - "filename": item.get("filename", ""), - "resource": item.get("resource", ""), - "score": item.get("score", 0.0) - }) - - rag_search_result = sorted(rag_search_result,key=lambda x:float(x.get("score") or 0), reverse=True)[:BAIKE_RESULT_LIMIT] - rag_search_result = dedupe_rag_results(rag_search_result, limit=BAIKE_RESULT_LIMIT) + evidence_items = await _rerank_evidence(query, evidence_items) return { - "rag_search_result": rag_search_result, - "graph_search_result": graph_results if isinstance(graph_results, str) else "" + "rag_search_result": rag_results, + "graph_search_result": graph_results if isinstance(graph_results, str) else "", + "evidence_items": evidence_items, } - except Exception as e: + except Exception as exc: return { "rag_search_result": [], "graph_search_result": "", - "error_message": f"检索失败: {str(e)}" + "evidence_items": [], + "error_message": f"检索失败: {exc}", } - async def generate_suggested_replies_baike(answer: str, question: str) -> List[Dict[str, Any]]: - """ - 生成针对问答结果的建议回复问题 - 使用大模型生成两个相关问题 - """ - prompt = BAIKE_PROMPTS["generate_suggested_replies"].format( - question=question, - answer=answer[:500] - ) + prompt = BAIKE_PROMPTS["generate_suggested_replies"].format(question=question, answer=answer[:500]) try: result = await OpenaiAPI.open_api_chat_without_thinking(prompt, model=None, json_output=True) - # 尝试解析JSON - # 提取JSON部分 - json_match = re.search(r'\[.*\]', result, re.DOTALL) + json_match = re.search(r"\[.*\]", result or "", 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: - # 确保每个元素都有title和content - 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": "还有哪些相关的知识点？"} - ] + replies = json.loads(json_match.group(0)) + formatted = [ + {"title": str(x["title"]), "content": str(x["content"])} + for x in replies[:2] + if isinstance(x, dict) and x.get("title") and x.get("content") + ] + if len(formatted) == 2: + return formatted + except Exception: + pass + return [ + {"title": "查看相关依据", "content": "这个回答对应的资料依据有哪些？"}, + {"title": "继续展开说明", "content": "请把这个问题再展开说明一下。"}, + ] @track_function_calls async def generate_answer(state: QAState) -> Dict[str, Any]: - """ - 正在生成问答结果 - """ - extracted_text = state.get("combined_query", "") - rag_results = state.get("rag_search_result", []) - graph_results = state.get("graph_search_result", "") - history_messages = state.get("history_messages", []) or [] + question = state.get("combined_query") or state.get("extracted_text") or "" + evidence_items = state.get("evidence_items") or [] need_retrieval = state.get("need_retrieval", True) - print("history_message(qa)", history_messages) - print("是否使用检索结果:", need_retrieval) + history_messages = state.get("history_messages", []) or [] try: - # 1. 处理检索结果（如果有） - rag_ctx_parts: List[str] = [] - file_name_list = [] - all_source_text = "" - original_image_paths = [] - - if need_retrieval: - # 收集所有 RAG 结果文本（只取前 4 条） - for item in sorted(rag_results or [], key=lambda x: float(x.get("score") or 0) if isinstance(x,dict) else 0, reverse=True)[:4]: -# for item in (rag_results or [])[:1]: - print(item) - print(111111111111111111111111111) - if isinstance(item, dict): - text = (item.get("text") or "").strip() - file_name = (item.get("filename") or "").strip() - if text: - rag_ctx_parts.append(text) - if file_name: - file_name_list.append(file_name) - - # 处理图谱结果 - print("123456", graph_results) - if graph_results and "图谱检索完成，但未返回有效数据。" in graph_results: - graph_results = "" - print("654321", graph_results) - - # 从所有结果中提取图片路径 - if graph_results: - all_source_text = graph_results + "\n" + "\n".join(rag_ctx_parts) - else: - all_source_text = "\n".join(rag_ctx_parts) - original_image_paths = extract_image_paths(all_source_text) - print("提取到的原始图片路径:", original_image_paths) - - # 发送事件通知（可选） - title_options = [f"✨已综合 {len(rag_ctx_parts)} 条检索结果", f"✨本次检索策略为综合多源信息"] - start_event = { - "type": "function_execution", - "title": random.choice(title_options), - "details": f"综合了 {len(rag_ctx_parts)} 条检索结果" - } - for callback in get_all_callbacks(): - try: - callback(start_event) - except Exception: - pass + source_text = "" + original_image_paths: List[str] = [] + if need_retrieval and evidence_items: + source_text = _format_evidence_context(evidence_items) + original_image_paths = extract_image_paths(source_text) + source_counts = _count_sources(evidence_items) + _emit_event( + f"✨已融合 {len(evidence_items)} 条证据", + f"RAG {source_counts.get('rag', 0)} 条，图谱 {source_counts.get('graph', 0)} 条", + ) - query_desc = extracted_text if extracted_text else "当前未明确描述问题" + domain_desc = ( + "舰船维修相关的法规、规范、行业标准及技术文件" + if state.get("route_flag") == "rules" + else "工业设备的结构原理、维护规程、故障诊断与维修操作" + ) - # 构建系统提示词 - if state.get("route_flag") == "rules": - domain_desc = "舰船修理相关的法规、规范、行业标准及技术文件" - else: - domain_desc = "工业设备的结构原理、维护规程、故障诊断与维修操作" - - # ========== 第一步：专注于生成高质量内容 ========== - # 根据是否有检索结果构建不同的提示词 - if need_retrieval and all_source_text.strip(): - content_system_prompt = BAIKE_PROMPTS["generate_answer_with_retrieval_system"].format(domain_desc=domain_desc) - content_user_content = BAIKE_PROMPTS["generate_answer_with_retrieval_user"].format( - extracted_text=extracted_text, - rag_count=len(rag_ctx_parts), - all_source_text=all_source_text + if need_retrieval and source_text.strip(): + system_prompt = BAIKE_PROMPTS["generate_answer_with_retrieval_system"].format(domain_desc=domain_desc) + user_prompt = BAIKE_PROMPTS["generate_answer_with_retrieval_user"].format( + extracted_text=question, + rag_count=len(evidence_items), + all_source_text=source_text, ) else: - content_system_prompt = BAIKE_PROMPTS["generate_answer_without_retrieval_system"].format(domain_desc=domain_desc) - content_user_content = BAIKE_PROMPTS["generate_answer_without_retrieval_user"].format(extracted_text=extracted_text) + system_prompt = BAIKE_PROMPTS["generate_answer_without_retrieval_system"].format(domain_desc=domain_desc) + user_prompt = BAIKE_PROMPTS["generate_answer_without_retrieval_user"].format(extracted_text=question) - content_messages = [] + messages = [] if history_messages: - content_messages.extend(history_messages) - content_messages.append({"role": "user", "content": content_user_content}) - - title_options = ["🤖 诊断分析中", "🤖 故障预检中"] - start_event = { - "type": "function_execution", - "title": random.choice(title_options), - "details": "正在生成回答..." - } - for callback in get_all_callbacks(): - try: - callback(start_event) - except Exception: - pass + messages.extend(history_messages) + messages.append({"role": "user", "content": user_prompt}) + _emit_event(random.choice(["🔍 证据整理中", "🧠 正在生成回答"]), "正在基于检索资料生成回答...") answer = await stream_generate_and_postprocess( - system_prompt=content_system_prompt, - messages=content_messages, + system_prompt=system_prompt, + messages=messages, original_image_paths=original_image_paths, - temperature=0.5, - fallback="抱歉，无法生成回答。" + temperature=0.2, + fallback="抱歉，无法生成回答。", ) - print("llm output", answer) + suggested_replies = await generate_suggested_replies_baike(answer, question) + return {"response": answer, "actions": [], "result_tag": "qa", "suggestedReplies": suggested_replies} + except Exception as exc: + return {"response": f"生成回答失败：{exc}", "actions": [], "suggestedReplies": []} + + +async def _search_rag_variants(queries: List[str], kb_id: Optional[str] = None) -> List[Dict[str, Any]]: + tasks = [rag_search(query=q, kb_id=kb_id, top_k=BAIKE_RAG_TOP_K) for q in queries if q.strip()] + results = await asyncio.gather(*tasks, return_exceptions=True) if tasks else [] + items: List[Dict[str, Any]] = [] + for result in results: + if isinstance(result, Exception) or not result.get("success"): + continue + for resource, rows in (result.get("sourceCitation") or {}).items(): + for row in rows: + item = dict(row) + item.setdefault("resource", resource) + items.append(item) + items = sorted(items, key=lambda x: float(x.get("score") or 0), reverse=True) + return dedupe_rag_results(items, limit=BAIKE_RESULT_LIMIT) + + +def _build_evidence_items( + rag_results: List[Dict[str, Any]], + graph_results: str, + query: str, +) -> List[Dict[str, Any]]: + candidates: List[Dict[str, Any]] = [] + for item in rag_results or []: + candidates.append({ + "source_type": "rag", + "title": item.get("filename") or item.get("resource") or "RAG 检索结果", + "filename": item.get("filename") or item.get("resource") or "", + "text": item.get("text") or "", + "score": float(item.get("score") or 0), + "file_id": str(item.get("file_id") or ""), + "slice_id": str(item.get("id") or item.get("slice_id") or ""), + "metadata": item, + }) + + graph_text = str(graph_results or "").strip() + if graph_text and "图谱检索完成，但未返回有效数据。" not in graph_text and graph_text != "查询无结果": + candidates.append({ + "source_type": "graph", + "title": "图谱检索结果", + "filename": "", + "text": graph_text, + "score": 0.45, + "file_id": "", + "slice_id": "", + "metadata": {}, + }) + + deduped: Dict[str, Dict[str, Any]] = {} + for item in candidates: + text = _clean_text(item.get("text") or "") + if not text: + continue + item["text"] = text[:5000] + key = item.get("slice_id") or _content_key(text) + local_score = _local_evidence_score(query, item) + item["local_score"] = local_score + existing = deduped.get(key) + if not existing or local_score > float(existing.get("local_score") or 0): + deduped[key] = item + return sorted(deduped.values(), key=lambda x: float(x.get("local_score") or 0), reverse=True) + + +async def _rerank_evidence(query: str, evidence_items: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + if not query.strip() or not evidence_items: + return evidence_items[:BAIKE_EVIDENCE_LIMIT] + selected = evidence_items[:24] + documents = [ + "\n".join([ + str(item.get("title") or ""), + str(item.get("filename") or ""), + str(item.get("text") or ""), + ])[:4000] + for item in selected + ] + rows = await OpenaiAPI.rerank(query=query, documents=documents, top_k=BAIKE_EVIDENCE_LIMIT) + if not rows: + return evidence_items[:BAIKE_EVIDENCE_LIMIT] + + ranked: List[Dict[str, Any]] = [] + seen = set() + for row in rows: + idx = int(row.get("index") or 0) + if idx < 0 or idx >= len(selected) or idx in seen: + continue + seen.add(idx) + item = dict(selected[idx]) + item["rerank_score"] = float(row.get("relevance_score") or 0.0) + ranked.append(item) + for idx, item in enumerate(selected): + if idx not in seen and len(ranked) < BAIKE_EVIDENCE_LIMIT: + ranked.append(item) + return ranked[:BAIKE_EVIDENCE_LIMIT] + + +def _format_evidence_context(items: List[Dict[str, Any]]) -> str: + parts = [] + label_map = {"rag": "RAG", "graph": "Graph"} + for idx, item in enumerate(items[:BAIKE_EVIDENCE_LIMIT], start=1): + source = label_map.get(item.get("source_type"), item.get("source_type") or "unknown") + score = item.get("rerank_score", item.get("local_score", item.get("score", ""))) + meta_lines = [ + f"[Evidence {idx}] source={source} score={score}", + f"Title: {item.get('title') or ''}", + ] + if item.get("filename"): + meta_lines.append(f"File: {item.get('filename')}") + if item.get("file_id"): + meta_lines.append(f"File ID: {item.get('file_id')}") + if item.get("slice_id"): + meta_lines.append(f"Slice ID: {item.get('slice_id')}") + meta_lines.append("Content:") + meta_lines.append(str(item.get("text") or "")) + parts.append("\n".join(meta_lines)) + return "\n\n---\n\n".join(parts) + + +def _local_evidence_score(query: str, item: Dict[str, Any]) -> float: + base = {"rag": 1.0, "graph": 0.65}.get(item.get("source_type"), 0.8) + raw = float(item.get("score") or 0) + score = base + min(max(raw, 0.0), 1.0) + text = " ".join([ + str(item.get("title") or ""), + str(item.get("filename") or ""), + str(item.get("text") or "")[:1500], + ]).lower() + title = str(item.get("title") or "").lower() + for term in _extract_search_terms(query): + t = term.lower() + if not t: + continue + if t in title: + score += 0.7 + elif t in text: + score += 0.25 + return score + + +def _build_search_queries(main_query: str, original_query: str = "", extra_queries: Optional[List[str]] = None) -> List[str]: + candidates: List[str] = [] + for text in [main_query, original_query, *(extra_queries or [])]: + text = _strip_query_noise(text) + if not text: + continue + candidates.append(text) + regex_terms = _extract_search_terms(text) + if regex_terms: + candidates.append(" ".join(regex_terms[:4])) + candidates.extend(regex_terms[:4]) + return _dedupe_strings(candidates)[:10] + + +def _extract_search_terms(query: str) -> List[str]: + stop = {"什么", "如何", "怎么", "哪些", "主要", "内容", "相关", "信息", "一下", "这个", "那个", "是否", "有没有"} + terms = [] + for token in re.findall(r"[\u4e00-\u9fff]{2,12}|[A-Za-z0-9][A-Za-z0-9_\-]{1,30}", query or ""): + token = token.strip("：:，,。.;；、（）()[]【】") + if len(token) < 2 or token in stop or re.fullmatch(r"\d+", token): + continue + terms.append(token) + return _dedupe_strings(terms) + + +def _strip_query_noise(text: str) -> str: + text = str(text or "").strip().strip('"“”') + if "用户问题：" in text: + text = text.split("用户问题：", 1)[1].strip() + return re.sub(r"\s+", " ", text) + + +def _parse_json_object(raw: str) -> Dict[str, Any]: + if not raw: + return {} + try: + return json.loads(raw) + except Exception: + match = re.search(r"\{.*\}", raw, re.DOTALL) + if match: + try: + return json.loads(match.group(0)) + except Exception: + return {} + return {} + + +def _json_dumps(value: Any) -> str: + try: + return json.dumps(value, ensure_ascii=False) + except Exception: + return str(value) + + +def _content_key(text: str) -> str: + return re.sub(r"\s+", "", text or "")[:220] + + +def _clean_text(text: str) -> str: + return re.sub(r"\n{3,}", "\n\n", str(text or "")).strip() + + +def _dedupe_strings(values: List[str]) -> List[str]: + out = [] + seen = set() + for value in values: + value = str(value or "").strip() + if not value or value in seen: + continue + seen.add(value) + out.append(value) + return out + + +def _count_sources(items: List[Dict[str, Any]]) -> Dict[str, int]: + counts: Dict[str, int] = {} + for item in items: + key = str(item.get("source_type") or "unknown") + counts[key] = counts.get(key, 0) + 1 + return counts + + +def _emit_event(title: str, details: str) -> None: + event = {"type": "function_execution", "title": title, "details": details} + for callback in get_all_callbacks(): + try: + callback(event) + except Exception: + pass - # 生成建议回复问题 - suggested_replies = await generate_suggested_replies_baike(answer, extracted_text) - return { - "response": answer, - "actions": [], - "result_tag": "qa", - "suggestedReplies": suggested_replies - } - except Exception as e: - return { - "response": f"生成回答失败：{str(e)}", - "actions": [], - "suggestedReplies": [] - } def route_after_judge(state: QAState) -> str: - """ - 判断后的路由：根据need_retrieval决定下一步 - """ - need_retrieval = state.get("need_retrieval", True) - if need_retrieval: - return "rewrite_query_with_history" - else: - return "generate_answer" + return "rewrite_query_with_history" if state.get("need_retrieval", True) else "generate_answer" -# =============== 构建工作流 =============== def create_baike_workflow(): - """创建普通问答工作流""" workflow = StateGraph(QAState) - - # 添加节点 workflow.add_node("get_baike_history_context", get_baike_history_context) workflow.add_node("judge_need_retrieval", judge_need_retrieval) workflow.add_node("rewrite_query_with_history", rewrite_query_with_history) workflow.add_node("search_rag_and_graph", search_rag_and_graph) workflow.add_node("generate_answer", generate_answer) - # 设置边 workflow.add_edge(START, "get_baike_history_context") workflow.add_edge("get_baike_history_context", "judge_need_retrieval") workflow.add_conditional_edges( "judge_need_retrieval", route_after_judge, - { - "rewrite_query_with_history": "rewrite_query_with_history", - "generate_answer": "generate_answer" - } + {"rewrite_query_with_history": "rewrite_query_with_history", "generate_answer": "generate_answer"}, ) workflow.add_edge("rewrite_query_with_history", "search_rag_and_graph") workflow.add_edge("search_rag_and_graph", "generate_answer") workflow.add_edge("generate_answer", END) - - # 编译工作流 return workflow.compile() -# =============== 工作流执行函数 =============== async def run_baike_workflow( extracted_text: str, combined_query: str, history_message: str = "", - route_flag: str = "" + route_flag: str = "", ) -> Dict[str, Any]: - """ - 执行普通问答工作流（统一接口） - - Args: - extracted_text: 文本描述（统一的输入参数） - history_message: 历史消息（目前只接收，不做处理） - - Returns: - 工作流执行结果 - """ - # 创建并编译工作流 app = create_baike_workflow() - - # 初始化状态（统一接口，只使用extracted_text和history_message） initial_state = { "extracted_text": extracted_text, "combined_query": combined_query, "retrieval_query": "", + "retrieval_queries": [], "route_flag": route_flag, "history_message": history_message, "history_messages": [], "need_retrieval": None, "rag_search_result": None, "graph_search_result": None, + "evidence_items": None, "response": "", "actions": [], "suggestedReplies": [], - "error_message": "" + "error_message": "", } - + try: - # 执行工作流（使用异步方式） final_state = await app.ainvoke(initial_state) - - # 检查是否有错误 if final_state.get("error_message"): return { "response": f"普通问答工作流执行失败: {final_state['error_message']}", "actions": [], "result_tag": "qa", - "suggestedReplies": [] + "suggestedReplies": [], } - - # 统一输出格式：完全透传 generate_answer 的结果 return { "response": final_state.get("response", ""), "actions": final_state.get("actions", []), "result_tag": "qa", - "suggestedReplies": final_state.get("suggestedReplies", []) + "suggestedReplies": final_state.get("suggestedReplies", []), } - - except Exception as e: + except Exception as exc: return { - "response": f"普通问答工作流执行失败: {str(e)}", + "response": f"普通问答工作流执行失败: {exc}", "actions": [], "result_tag": "qa", - "suggestedReplies": [] + "suggestedReplies": [], } - - -# =============== 测试 =============== -if __name__ == "__main__": - import asyncio - - # 测试用例 - test_cases = [ - "船舰主发动机的安全警告是什么", - ] - - async def test(): - for i, test_text in enumerate(test_cases, 1): - print(f"\n{'='*60}") - print(f"测试用例 {i}") - print(f"{'='*60}") - - # ✅ 修复：传入 combined_query 参数 - result = await run_baike_workflow( - extracted_text=test_text, - combined_query=test_text - ) - - print(f"\n执行结果:") - print(f"成功：{result.get('success', False)}") - print(f"\n响应内容:") - print(result.get("response", "无响应")) - - asyncio.run(test()) - diff --git a/workflows/workflow_other.py b/workflows/workflow_other.py index e4d39e0..ddb144b 100644 --- a/workflows/workflow_other.py +++ b/workflows/workflow_other.py @@ -1,293 +1,293 @@ -""" -工作流：普通问答Agent（使用LangGraph） -输入：文本（转化后和原文本query）、图片描述、语音文本 -步骤： -1. 使用RAG和GraphRAG检索相关信息 -2. 生成回答 -使用 OpenaiAPI 方法调用大模型，尽可能使用异步 -""" -from typing import TypedDict, Optional, Dict, Any, List -from langgraph.graph import StateGraph, START, END -from modelsAPI.model_api import OpenaiAPI -from utils.function_tracker import track_function_calls, get_current_stream_handler -import asyncio -import json -import re - - -# =============== 定义状态 =============== -class Othertate(TypedDict): - """普通问答工作流状态""" - extracted_text: str # 已提取的文本（统一接口，只接收文本） - combined_query: str - route_flag: str # 路由标识 - history_message: str # 历史消息（JSON格式） - history_messages: List[Dict[str, Any]] # 过滤后的历史消息列表 - - # RAG检索结果 - rag_search_result: Optional[List[Dict[str, Any]]] # RAG检索结果 - graph_search_result: Optional[str] # 图谱检索结果 - - # 最终响应 - response: str # 最终响应 - actions: List[Dict[str, Any]] # 按钮信息列表 [{"type": "...", "label": "...", "icon": "..."}] - suggestedReplies: List[Dict[str, Any]] # 建议回复问题列表 - error_message: str # 错误信息 - - -# =============== 节点函数 =============== - -def get_other_history_context(state: Othertate) -> Dict[str, Any]: - """ - 知识问答步骤：获取问答类历史信息 - 问答工作流只关注历史问答记录，用于提供上下文 - 支持格式：[{'role': 'user', 'content': '...'}, {'role': 'assistant', 'content': '...'}] - 返回消息列表（history_message已经在app.py中过滤过了） - """ - import json - - history_message = state.get("history_message", "") or "" - history_messages = [] - - if history_message: - try: - # history_message已经在app.py中过滤过了，直接解析 - if isinstance(history_message, str): - history_data = json.loads(history_message) - else: - history_data = history_message - - if isinstance(history_data, list): - # 只取最近10条历史记录，避免上下文过长 - history_messages = history_data[-10:] - except Exception as e: - pass - - return {"history_messages": history_messages} - - -async def generate_suggested_replies_other(answer: str, question: str) -> List[Dict[str, Any]]: - """ - 生成针对问答结果的建议回复问题 - 使用大模型生成两个相关问题 - """ - prompt = f""" -你是一个问答助手。根据以下问答内容，生成两个用户可能关心的后续问题。 - -【用户问题】 -{question} - -【回答内容】 -{answer[:500]} - -请生成两个简洁、实用的问题，这些问题应该： -1. 不得出现需要归档和下载等操作相关的问题 -2. 与当前问答内容相关，能够帮助用户进一步了解相关信息 -3. 问题要具体、可操作 -4. 每个问题不超过20个字 - -请以JSON格式输出，格式如下： -[ - {{"title": "问题1的标题", "content": "问题1的完整内容"}}, - {{"title": "问题2的标题", "content": "问题2的完整内容"}} -] - -只输出JSON，不要有其他文字说明。 -""" - try: - result = await OpenaiAPI.open_api_chat_without_thinking(prompt, model=None) - # 尝试解析JSON - # 提取JSON部分 - json_match = re.search(r'\[.*\]', result, 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: - # 确保每个元素都有title和content - 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": "还有哪些相关的知识点？"} - ] - - -@track_function_calls -async def generate_answer(state: Othertate) -> Dict[str, Any]: - """ - 正在生成问答结果 - """ - extracted_text = state.get("combined_query", "") - history_messages = state.get("history_messages", []) or [] - print("history_message(qa)", history_messages) - - try: - # 构建系统提示词 - system_prompt = """我是AI助手，根据历史信息和用户问题进行回答。 -请基于以上信息，给出准确、专业的回答。如果有历史问答上下文，请结合历史对话内容理解当前问题。""" - - # 构建消息列表 - messages = [] - - # 添加历史消息 - if history_messages: - messages.extend(history_messages) - - # 添加当前用户查询 - messages.append({"role": "user", "content": f"用户问题：{extracted_text}"}) - - # 调用LLM生成回答（异步，禁用thinking） - answer = await OpenaiAPI.open_api_chat_without_thinking( - model=None, - system_prompt=system_prompt, - messages=messages - ) - - # 过滤掉 标签（如果有） - import re - answer = re.sub(r'.*?', '', answer, flags=re.DOTALL) - answer = answer.strip() if answer else "抱歉，无法生成回答。" - - # 返回按钮：确认结果、重新生成 - actions = [] - - # 生成建议回复问题 - suggested_replies = await generate_suggested_replies_other(answer, extracted_text) - - return { - "response": answer, - "actions": actions, - "result_tag": "other", - "suggestedReplies": suggested_replies - } - except Exception as e: - return { - "response": f"生成回答失败: {str(e)}", - "actions": [], - "suggestedReplies": [] - } - - -# =============== 构建工作流 =============== -def create_qa_workflow(): - """创建普通问答工作流""" - workflow = StateGraph(Othertate) - - # 添加节点 - workflow.add_node("get_other_history_context", get_other_history_context) - workflow.add_node("generate_answer", generate_answer) - - # 设置边 - workflow.add_edge(START, "get_other_history_context") - workflow.add_edge("get_other_history_context", "generate_answer") - workflow.add_edge("generate_answer", END) - - # 编译工作流 - return workflow.compile() - - -# =============== 工作流执行函数 =============== -async def run_other_workflow( - extracted_text: str, - combined_query: str, - history_message: str = "", - route_flag: str = "" -) -> Dict[str, Any]: - """ - 执行普通问答工作流（统一接口） - - Args: - extracted_text: 文本描述（统一的输入参数） - history_message: 历史消息（目前只接收，不做处理） - - Returns: - 工作流执行结果 - """ - # 创建并编译工作流 - app = create_qa_workflow() - - # 初始化状态（统一接口，只使用extracted_text和history_message） - initial_state = { - "extracted_text": extracted_text, - "combined_query": combined_query, - "history_message": history_message, - "history_messages": [], - "response": "", - "actions": [], - "suggestedReplies": [], - "error_message": "" - } - - try: - # 执行工作流（使用异步方式） - final_state = await app.ainvoke(initial_state) - - # 检查是否有错误 - if final_state.get("error_message"): - return { - "response": f"普通问答工作流执行失败: {final_state['error_message']}", - "actions": [], - "result_tag": "qa", - "suggestedReplies": [] - } - - # 统一输出格式：完全透传 generate_answer 的结果 - return { - "response": final_state.get("response", ""), - "actions": final_state.get("actions", []), - "result_tag": "other", - "suggestedReplies": final_state.get("suggestedReplies", []) - } - - except Exception as e: - return { - "response": f"普通问答工作流执行失败: {str(e)}", - "actions": [], - "result_tag": "other", - "suggestedReplies": [] - } - - -# =============== 测试 =============== -if __name__ == "__main__": - import asyncio - - # 测试用例 - test_cases = [ - "什么是主机？", - "主机的作用是什么？" - ] - - - async def test(): - for i, test_text in enumerate(test_cases, 1): - print(f"\n{'=' * 60}") - print(f"测试用例 {i}") - print(f"{'=' * 60}") - - result = await run_other_workflow(extracted_text=test_text) - - print(f"\n执行结果:") - print(f"成功: {result.get('success', False)}") - print(f"\n响应内容:") - print(result.get("response", "无响应")) - - - asyncio.run(test()) - +""" +工作流：普通问答Agent（使用LangGraph） +输入：文本（转化后和原文本query）、图片描述、语音文本 +步骤： +1. 使用RAG和GraphRAG检索相关信息 +2. 生成回答 +使用 OpenaiAPI 方法调用大模型，尽可能使用异步 +""" +from typing import TypedDict, Optional, Dict, Any, List +from langgraph.graph import StateGraph, START, END +from modelsAPI.model_api import OpenaiAPI +from utils.function_tracker import track_function_calls, get_current_stream_handler +import asyncio +import json +import re + + +# =============== 定义状态 =============== +class Othertate(TypedDict): + """普通问答工作流状态""" + extracted_text: str # 已提取的文本（统一接口，只接收文本） + combined_query: str + route_flag: str # 路由标识 + history_message: str # 历史消息（JSON格式） + history_messages: List[Dict[str, Any]] # 过滤后的历史消息列表 + + # RAG检索结果 + rag_search_result: Optional[List[Dict[str, Any]]] # RAG检索结果 + graph_search_result: Optional[str] # 图谱检索结果 + + # 最终响应 + response: str # 最终响应 + actions: List[Dict[str, Any]] # 按钮信息列表 [{"type": "...", "label": "...", "icon": "..."}] + suggestedReplies: List[Dict[str, Any]] # 建议回复问题列表 + error_message: str # 错误信息 + + +# =============== 节点函数 =============== + +def get_other_history_context(state: Othertate) -> Dict[str, Any]: + """ + 知识问答步骤：获取问答类历史信息 + 问答工作流只关注历史问答记录，用于提供上下文 + 支持格式：[{'role': 'user', 'content': '...'}, {'role': 'assistant', 'content': '...'}] + 返回消息列表（history_message已经在app.py中过滤过了） + """ + import json + + history_message = state.get("history_message", "") or "" + history_messages = [] + + if history_message: + try: + # history_message已经在app.py中过滤过了，直接解析 + if isinstance(history_message, str): + history_data = json.loads(history_message) + else: + history_data = history_message + + if isinstance(history_data, list): + # 只取最近10条历史记录，避免上下文过长 + history_messages = history_data[-10:] + except Exception as e: + pass + + return {"history_messages": history_messages} + + +async def generate_suggested_replies_other(answer: str, question: str) -> List[Dict[str, Any]]: + """ + 生成针对问答结果的建议回复问题 + 使用大模型生成两个相关问题 + """ + prompt = f""" +你是一个问答助手。根据以下问答内容，生成两个用户可能关心的后续问题。 + +【用户问题】 +{question} + +【回答内容】 +{answer[:500]} + +请生成两个简洁、实用的问题，这些问题应该： +1. 不得出现需要归档和下载等操作相关的问题 +2. 与当前问答内容相关，能够帮助用户进一步了解相关信息 +3. 问题要具体、可操作 +4. 每个问题不超过20个字 + +请以JSON格式输出，格式如下： +[ + {{"title": "问题1的标题", "content": "问题1的完整内容"}}, + {{"title": "问题2的标题", "content": "问题2的完整内容"}} +] + +只输出JSON，不要有其他文字说明。 +""" + try: + result = await OpenaiAPI.open_api_chat_without_thinking(prompt, model=None) + # 尝试解析JSON + # 提取JSON部分 + json_match = re.search(r'\[.*\]', result, 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: + # 确保每个元素都有title和content + 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": "还有哪些相关的知识点？"} + ] + + +@track_function_calls +async def generate_answer(state: Othertate) -> Dict[str, Any]: + """ + 正在生成问答结果 + """ + extracted_text = state.get("combined_query", "") + history_messages = state.get("history_messages", []) or [] + print("history_message(qa)", history_messages) + + try: + # 构建系统提示词 + system_prompt = """我是AI助手，根据历史信息和用户问题进行回答。 +请基于以上信息，给出准确、专业的回答。如果有历史问答上下文，请结合历史对话内容理解当前问题。""" + + # 构建消息列表 + messages = [] + + # 添加历史消息 + if history_messages: + messages.extend(history_messages) + + # 添加当前用户查询 + messages.append({"role": "user", "content": f"用户问题：{extracted_text}"}) + + # 调用LLM生成回答（异步，禁用thinking） + answer = await OpenaiAPI.open_api_chat_without_thinking( + model=None, + system_prompt=system_prompt, + messages=messages + ) + + # 过滤掉 标签（如果有） + import re + answer = re.sub(r'.*?', '', answer, flags=re.DOTALL) + answer = answer.strip() if answer else "抱歉，无法生成回答。" + + # 返回按钮：确认结果、重新生成 + actions = [] + + # 生成建议回复问题 + suggested_replies = await generate_suggested_replies_other(answer, extracted_text) + + return { + "response": answer, + "actions": actions, + "result_tag": "other", + "suggestedReplies": suggested_replies + } + except Exception as e: + return { + "response": f"生成回答失败: {str(e)}", + "actions": [], + "suggestedReplies": [] + } + + +# =============== 构建工作流 =============== +def create_qa_workflow(): + """创建普通问答工作流""" + workflow = StateGraph(Othertate) + + # 添加节点 + workflow.add_node("get_other_history_context", get_other_history_context) + workflow.add_node("generate_answer", generate_answer) + + # 设置边 + workflow.add_edge(START, "get_other_history_context") + workflow.add_edge("get_other_history_context", "generate_answer") + workflow.add_edge("generate_answer", END) + + # 编译工作流 + return workflow.compile() + + +# =============== 工作流执行函数 =============== +async def run_other_workflow( + extracted_text: str, + combined_query: str, + history_message: str = "", + route_flag: str = "" +) -> Dict[str, Any]: + """ + 执行普通问答工作流（统一接口） + + Args: + extracted_text: 文本描述（统一的输入参数） + history_message: 历史消息（目前只接收，不做处理） + + Returns: + 工作流执行结果 + """ + # 创建并编译工作流 + app = create_qa_workflow() + + # 初始化状态（统一接口，只使用extracted_text和history_message） + initial_state = { + "extracted_text": extracted_text, + "combined_query": combined_query, + "history_message": history_message, + "history_messages": [], + "response": "", + "actions": [], + "suggestedReplies": [], + "error_message": "" + } + + try: + # 执行工作流（使用异步方式） + final_state = await app.ainvoke(initial_state) + + # 检查是否有错误 + if final_state.get("error_message"): + return { + "response": f"普通问答工作流执行失败: {final_state['error_message']}", + "actions": [], + "result_tag": "qa", + "suggestedReplies": [] + } + + # 统一输出格式：完全透传 generate_answer 的结果 + return { + "response": final_state.get("response", ""), + "actions": final_state.get("actions", []), + "result_tag": "other", + "suggestedReplies": final_state.get("suggestedReplies", []) + } + + except Exception as e: + return { + "response": f"普通问答工作流执行失败: {str(e)}", + "actions": [], + "result_tag": "other", + "suggestedReplies": [] + } + + +# =============== 测试 =============== +if __name__ == "__main__": + import asyncio + + # 测试用例 + test_cases = [ + "什么是主机？", + "主机的作用是什么？" + ] + + + async def test(): + for i, test_text in enumerate(test_cases, 1): + print(f"\n{'=' * 60}") + print(f"测试用例 {i}") + print(f"{'=' * 60}") + + result = await run_other_workflow(extracted_text=test_text) + + print(f"\n执行结果:") + print(f"成功: {result.get('success', False)}") + print(f"\n响应内容:") + print(result.get("response", "无响应")) + + + asyncio.run(test()) + diff --git a/workflows/workflow_utils.py b/workflows/workflow_utils.py index a9bf6a6..881d319 100644 --- a/workflows/workflow_utils.py +++ b/workflows/workflow_utils.py @@ -926,12 +926,15 @@ def remove_duplicate_content(text: str) -> str: return "\n".join(result) + + # 推荐调用顺序： # # text = remove_duplicate_content(text) # text = normalize_latex_spacing(text) + def _extract_rag_text(item: Any) -> str: if isinstance(item, dict): return str(item.get("text") or item.get("content") or "") @@ -1316,9 +1319,13 @@ async def stream_generate_and_postprocess( stream_handler.send_stream_content(accumulated_content) answer = answer.strip() if answer else fallback + print("111-",answer) answer = re.sub(r'ynchroneg>.*?ost switching>', '', answer, flags=re.DOTALL) + print("222-",answer) answer = remove_duplicate_content(answer) + print("333-",answer) answer = normalize_markdown_images(answer, original_paths=original_image_paths) + print("444-",answer) answer = normalize_latex_spacing(answer) return answer.strip() if answer else fallback