强制同步9092容器线上运行版本，覆盖远端代码

app.py

@@ -629,6 +629,76 @@ class FeedbackClassifyRequest(BaseModel):
 
 
 
+def _citation_display_filename(
+    value: Any,
+    file_id: Any = "",
+) -> Any:
+    """仅转换文本溯源展示名，不修改 file_id 或实际文件路径。"""
+    if not isinstance(value, str):
+        return value
+    if not re.search(r"(?i)\.(?:pptx|ppt)$", value.strip()):
+        return value
+    normalized_file_id = str(file_id or "").strip()
+    if not normalized_file_id:
+        return value
+    return f"{normalized_file_id}.pdf"
+
+
+def normalize_source_citation_filenames(
+    source_citation: Dict[str, Any],
+) -> Dict[str, Any]:
+    """将最终文本溯源中的 PPT/PPTX 展示后缀统一转换为 PDF。"""
+    if not isinstance(source_citation, dict):
+        return source_citation
+
+    normalized: Dict[str, Any] = {}
+    for resource_name, items in source_citation.items():
+        group_file_id = ""
+        if isinstance(items, list):
+            group_file_id = next(
+                (
+                    item.get("file_id")
+                    for item in items
+                    if isinstance(item, dict) and item.get("file_id")
+                ),
+                "",
+            )
+        display_resource = _citation_display_filename(
+            resource_name,
+            group_file_id,
+        )
+
+        if isinstance(items, list):
+            normalized_items = []
+            for item in items:
+                if not isinstance(item, dict):
+                    normalized_items.append(item)
+                    continue
+                normalized_item = dict(item)
+                item_file_id = normalized_item.get("file_id") or group_file_id
+                for field in ("filename", "resource", "title"):
+                    if field in normalized_item:
+                        normalized_item[field] = _citation_display_filename(
+                            normalized_item[field],
+                            item_file_id,
+                        )
+                normalized_items.append(normalized_item)
+        else:
+            normalized_items = items
+
+        # 极少数同名 .ppt/.pptx 会映射到同一 .pdf，保留全部切片。
+        if (
+            display_resource in normalized
+            and isinstance(normalized[display_resource], list)
+            and isinstance(normalized_items, list)
+        ):
+            normalized[display_resource].extend(normalized_items)
+        else:
+            normalized[display_resource] = normalized_items
+
+    return normalized
+
+
 def format_stream_event(event_type: str,
                         data: Dict[str, Any],
                         chat_id: str = "",
@@ -690,7 +760,9 @@ def format_stream_event(event_type: str,
         if has_rag or has_graph:
             source_citation = {}
             if has_rag:
-                source_citation.update(rag_result)
+                source_citation.update(
+                    normalize_source_citation_filenames(rag_result)
+                )
             if has_graph:
                 source_citation["xxxx.graph"] = graph_result
 
@@ -936,6 +1008,7 @@ async def stream_main_agent_execution(
         agent_task = asyncio.create_task(run_agent())
 
         execution_complete_received = False
+        has_streamed_content = False
         rag_result = []
         graph_result = []
         while True:
@@ -953,6 +1026,11 @@ async def stream_main_agent_execution(
 
                 if event_type == "function_execution":
                     logger.info(f"function_execution event - title: {title}, has_result: {result is not None}, result_type: {type(result)}")
+                    # 装饰器产生的无标题事件属于内部入参/返回值，不应在前端展示。
+                    # 前端会把它们自动编号成“步骤 2”“步骤 5”，从而泄露检索原文
+                    # 和未整理的回答草稿。
+                    if not title.strip():
+                        continue
                     if title == "知识库搜索工具" and result and isinstance(result, dict):
                         print("===============================", result)
                         rag_result_raw = result.get("sourceCitation", {})
@@ -985,9 +1063,11 @@ async def stream_main_agent_execution(
                 elif event_type == "stream_content":
                     content = event.get("content", "")
 
-                    yield format_stream_event("stream_content", {
+                    if content:
-                        "content": content
+                        has_streamed_content = True
-                    }, chat_id=chat_id, message_id=message_id)
+                        yield format_stream_event("stream_content", {
+                            "content": content
+                        }, chat_id=chat_id, message_id=message_id)
 
 
             except asyncio.TimeoutError:
@@ -1005,8 +1085,13 @@ async def stream_main_agent_execution(
             }, chat_id=chat_id, message_id=message_id)
         else:
             title = await extract_conversation_title(final_result.get("content", "") if final_result else "")
+            frontend_final_result = final_result
+            if has_streamed_content and final_result:
+                # 正文已按增量流式发送，完成事件只补充 actions 等元数据。
+                # 再携带完整 content 会被前端追加为第二份重复正文。
+                frontend_final_result = {**final_result, "content": ""}
             yield format_stream_event("execution_complete", {
-                "final_result": final_result
+                "final_result": frontend_final_result
             }, chat_id=chat_id, message_id=message_id)
 
             yield format_stream_event("execution_complete", {

config.py

@@ -37,7 +37,7 @@ _request_user_config: ContextVar[Optional[Dict[str, str]]] = ContextVar('request
 
 # RAG配置的默认值
 _RAG_CONFIG_DEFAULT = {
-    "search_endpoint": "http://192.168.0.46:11000/api/v1/knowledge/search/",
+    "search_endpoint": "http://192.168.0.46:11100/api/v1/knowledge/search/",
     "x-user-id": "1",
     "x-user-name": "testuser",
     "x-role": "admin",
@@ -180,7 +180,7 @@ NEO4J_CONFIG = {
 
 # ==================== 知识库匹配配置 ====================
 KB_TREE_CONFIG = {
-    "url": "http://192.168.0.46:11000/api/v1/knowledge/knowledge_base/tree"
+    "url": "http://192.168.0.46:11100/api/v1/knowledge/knowledge_base/tree"
 }
 
 # ==================== 维修反馈统计配置 ====================
@@ -214,6 +214,17 @@ MAP_INS={
     "舷号" : "ship_number",
 }
 
+# ==================== 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,
+}
+
+
 # ==================== neo4j配置 ====================
 NEO4J_CONFIG = {
     "uri": "neo4j://192.168.0.46:57687",

main_agent.py.bak

@@ -0,0 +1,249 @@
+"""
+主脑Agent - 简化版（无意图分类）
+
+职责：
+1. 输入处理：图片、音频、文本提取
+2. 直接路由：根据外部传入的 route_flag 决定调用哪个工作流
+
+工作流列表（来自 workflow_registry）：
+- 故障排查与修理
+- 舰船百科
+- 操作使用
+"""
+
+from typing import TypedDict, Literal, Optional, Dict, Any, List
+from langgraph.graph import StateGraph, START, END
+from tools.function_tool import (
+    vlm_image_to_text,
+    asr_audio_to_text,
+    detect_input_type
+)
+from modelsAPI.model_api import OpenaiAPI
+from utils.function_tracker import track_function_calls
+from workflow_registry import VALID_ROUTE_FLAGS
+import json
+import re
+
+
+class MainAgentState(TypedDict):
+    """主脑Agent状态"""
+    raw_input: Dict[str, Any]
+    has_image: bool
+    has_audio: bool
+    has_query: bool
+    has_file_text: bool
+    input_type: str
+    
+    history_message: str
+    extracted_text: str
+    image_description: str
+    asr_text: str
+    file_text: str
+    combined_query: str
+    
+    workflow_result: Optional[Dict[str, Any]]
+    final_response: str
+    error_message: str
+    
+    route_flag: str
+    route_params: Dict[str, Any]
+
+
+def detect_input_type_node(state: MainAgentState) -> Dict[str, Any]:
+    """检测输入类型"""
+    raw = state["raw_input"]
+    print(raw)
+    print(22222222222222222222222222)
+    result = detect_input_type.invoke({"raw_input": raw})
+    
+    has_image = result.get("has_image", False)
+    has_audio = result.get("has_audio", False)
+    has_query = result.get("has_query", False)
+    
+    has_file_text = False
+    file_text_v = raw.get("file_text")
+    if isinstance(file_text_v, str) and file_text_v.strip():
+        has_file_text = True
+    
+    input_type = "mixed"
+    count = sum([has_image, has_audio, has_query, has_file_text])
+    if count == 1:
+        if has_image:
+            input_type = "image"
+        elif has_audio:
+            input_type = "audio"
+        elif has_query:
+            input_type = "text"
+        elif has_file_text:
+            input_type = "file"
+    
+    return {
+        "has_image": has_image,
+        "has_audio": has_audio,
+        "has_query": has_query,
+        "has_file_text": has_file_text,
+        "input_type": input_type,
+    }
+
+
+async def process_multimodal_input(state: MainAgentState) -> Dict[str, Any]:
+    """处理多模态输入"""
+    has_image = state.get("has_image", False)
+    has_audio = state.get("has_audio", False)
+    has_query = state.get("has_query", False)
+    has_file_text = state.get("has_file_text", False)
+    raw = state["raw_input"]
+    
+    image_description = ""
+    asr_text = ""
+    extracted_text = ""
+    
+    try:
+        if has_image:
+            image_path = raw.get("image", "")
+            if image_path:
+                try:
+                    image_description = await vlm_image_to_text.ainvoke({
+                        "image_path": image_path
+                    })
+                    if not image_description or image_description.startswith("VLM 调用失败"):
+                        image_description = ""
+                except Exception:
+                    image_description = ""
+        
+        if has_audio:
+            audio_path = raw.get("audio", "")
+            if audio_path:
+                try:
+                    asr_text = await asr_audio_to_text.ainvoke({
+                        "audio_path": audio_path
+                    })
+                    if not asr_text or asr_text.startswith("ASR 调用失败"):
+                        asr_text = ""
+                except Exception:
+                    asr_text = ""
+        
+        text_parts = []
+        if raw.get("query"):
+            text_parts.append(raw["query"])
+        if image_description:
+            text_parts.append(f"[图片描述] {image_description}")
+        if asr_text:
+            text_parts.append(f"[语音转文字] {asr_text}")
+        if raw.get("file_text"):
+            text_parts.append(f"[文件内容] {raw['file_text']}")
+        
+        extracted_text = "\n".join(text_parts) if text_parts else ""
+        combined_query = raw.get("query", "") or extracted_text
+        
+        return {
+            "extracted_text": extracted_text,
+            "image_description": image_description,
+            "asr_text": asr_text,
+            "combined_query": combined_query,
+            "file_text": raw.get("file_text", ""),
+        }
+    
+    except Exception as e:
+        return {
+            "error_message": f"输入处理失败: {str(e)}",
+            "extracted_text": "",
+        }
+
+# @track_function_calls
+def route_to_workflow(state: MainAgentState) -> Dict[str, Any]:
+    """
+    🤖分析意图，调用相关智能体
+    """
+    route_flag = state.get("route_flag", "问题排查")
+    
+    if route_flag not in VALID_ROUTE_FLAGS:
+        route_flag = "问题排查"
+    
+    query_text = state.get("extracted_text", "")
+    combined_query = state.get("combined_query", "") or query_text
+    
+    raw_input = state.get("raw_input", {})
+    history_message = raw_input.get("history_message", "")
+    
+    route_params = {
+        "extracted_text": query_text,
+        "combined_query": combined_query,
+        "history_message": history_message,
+    }
+    
+    if raw_input.get("file_text"):
+        route_params["file_text"] = raw_input["file_text"]
+    
+    print(f"[主脑Agent] 路由到: {route_flag}")
+    print(f"[主脑Agent] 参数: combined_query={combined_query[:100]}...")
+
+    if route_flag == "问题排查":
+        detail = "正在调用问题排查智能体"
+    elif route_flag == "百科问答":
+        detail = "正在调用百科问答智能体"
+    elif route_flag == "操作使用":
+        detail = "正在调用操作使用智能体"
+    elif route_flag == "统计":
+        detail = "正在调用统计智能体"
+    else:
+        detail = "正在调用通用智能体"
+    
+    return {
+        "route_flag": route_flag,
+        "route_params": route_params,
+        "__detail__": detail,
+    }
+
+
+def create_main_agent():
+    """创建主脑Agent"""
+    workflow = StateGraph(MainAgentState)
+    
+    workflow.add_node("detect_input_type", detect_input_type_node)
+    workflow.add_node("process_input", process_multimodal_input)
+    workflow.add_node("route", route_to_workflow)
+    
+    workflow.add_edge(START, "detect_input_type")
+    workflow.add_edge("detect_input_type", "process_input")
+    workflow.add_edge("process_input", "route")
+    workflow.add_edge("route", END)
+    
+    return workflow.compile()
+
+
+from workflows.history_manager import filter_image_urls
+
+
+async def extract_conversation_title(content: str) -> str:
+    """提取对话标题"""
+    try:
+        prompt = f"请从以下对话内容中提取一个简短的标题（不超过20字）：\n\n{content[:500]}"
+        response = await OpenaiAPI.open_api_chat_without_thinking(
+            query=prompt,
+            model=None,
+            system_prompt="你是一个标题提取助手，请从对话内容中提取一个简短的标题。",
+            messages=[]
+        )
+        title = response.strip() if response else "新对话"
+        if len(title) > 30:
+            title = title[:30] + "..."
+        return title
+    except Exception:
+        return "新对话"
+
+
+if __name__ == "__main__":
+    import asyncio
+    
+    async def test():
+        print("测试主脑Agent")
+        agent = create_main_agent()
+        result = await agent.ainvoke({
+            "raw_input": {"query": "101舰主机发生异响"},
+            "route_flag": "故障排查与修理"
+        })
+        print(f"结果: {result}")
+    
+    asyncio.run(test())
+

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 openai import AsyncOpenAI
 
@@ -216,6 +216,66 @@ class OpenaiAPI:
             print(error_msg)
             return f"Error: {str(e)}"
 
+    @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
     def get_embeddings(embedding_text: str):
         embedding_base_url = EMBEDDING_CONFIG["base_url"]

modelsAPI/model_api.py.bak

@@ -0,0 +1,368 @@
+import asyncio
+import httpx
+from typing import List, Union
+import os
+import numpy as np
+from openai import OpenAI
+from config import LLM_CONFIG, EMBEDDING_CONFIG
+
+from openai import AsyncOpenAI
+
+# 模块级单例客户端，避免每次调用都创建新连接
+_async_client: AsyncOpenAI = None
+
+
+def _get_async_client() -> AsyncOpenAI:
+    """获取或创建 AsyncOpenAI 单例客户端"""
+    global _async_client
+    if _async_client is None:
+        _async_client = AsyncOpenAI(
+            api_key=LLM_CONFIG["api_key"],
+            base_url=LLM_CONFIG["base_url"],
+        )
+    return _async_client
+
+
+class OpenaiAPI:
+    @staticmethod
+    async def open_api_chat_without_thinking(
+        query: str = None,
+        model: str = None,
+        json_output: bool = False,
+        system_prompt: str = None,
+        messages: list = None,
+        enable_thinking: bool = True,
+        temperature: float = 0.1
+    ) -> str:
+        if model is None:
+            model = LLM_CONFIG["model"]
+        if system_prompt is None:
+            system_prompt = "我是一个专业的AI助手，请用中文回答问题。请直接给出最终答案，不要包含任何推理步骤、思考过程、解释性文字或前缀。"
+
+
+        # 构建消息
+        message_list = [{"role": "system", "content": system_prompt}]
+    
+        if messages is not None:
+            message_list.extend(messages)
+        if query is not None:
+            message_list.append({"role": "user", "content": query})
+
+        client = _get_async_client()
+
+        kwargs = {
+            "model": model,
+            "messages": message_list,
+            "temperature": temperature,
+            "stream": False,
+            "max_tokens":LLM_CONFIG["max_tokens"]
+        }
+
+        kwargs["extra_body"] = {"chat_template_kwargs": {"enable_thinking": False}}
+
+        if json_output:
+            kwargs["response_format"] = {"type": "json_object"}
+        
+        print("model API. history message", message_list)
+        response = await client.chat.completions.create(**kwargs)
+        return response.choices[0].message.content
+    
+    @staticmethod
+    async def open_api_chat_stream(
+        query: str = None,
+        model: str = None,
+        json_output: bool = False,
+        system_prompt: str = None,
+        messages: list = None,
+        enable_thinking: bool = True,
+        temperature: float = 0.1
+    ):
+        """
+        流式输出大模型响应
+        
+        Yields:
+            str: 流式内容片段
+        """
+        if model is None:
+            model = LLM_CONFIG["model"]
+        if system_prompt is None:
+            system_prompt = "我是一个专业的AI助手，请用中文回答问题。请直接给出最终答案，不要包含任何推理步骤、思考过程、解释性文字或前缀。"
+
+        # 构建消息
+        message_list = [{"role": "system", "content": system_prompt}]
+    
+        if messages is not None:
+            message_list.extend(messages)
+        if query is not None:
+            message_list.append({"role": "user", "content": query})
+
+        client = _get_async_client()
+
+        kwargs = {
+            "model": model,
+            "messages": message_list,
+            "temperature": temperature,
+            "stream": True,
+            "max_tokens":LLM_CONFIG["max_tokens"]
+        }
+
+        kwargs["extra_body"] = {"chat_template_kwargs": {"enable_thinking": False}}
+
+        if json_output:
+            kwargs["response_format"] = {"type": "json_object"}
+        
+        print("model API stream. history message", message_list)
+        
+        stream = await client.chat.completions.create(**kwargs)
+        async for chunk in stream:
+            if chunk.choices and len(chunk.choices) > 0:
+                delta = chunk.choices[0].delta
+                if delta.content:
+                    yield delta.content
+
+    @staticmethod   
+    async def open_api_vl_without_thinking(
+        image_url: str
+    ) -> str:
+        """
+        专门用于从图片中提取【设备名称】和【故障现象】。
+        自动过滤思考过程，仅返回核心结果。
+        
+        参数:
+            image_url: Base64 格式的图片数据 (data:image/...;base64,...)
+            model: 模型名称，默认为配置中的模型
+            custom_instruction: 额外的特定指令 (可选)
+        """
+        
+        # 1. 确定模型名称
+        model = LLM_CONFIG.get("model")
+        
+        # 2. 构建强约束的 System Prompt
+        # 核心目标：禁止思考标签，禁止废话，只给结果
+        system_prompt = (
+            "你是一个工业视觉分析专家。你的任务是从图片中识别设备并诊断故障。\n"
+            "【严格约束】\n"
+            "1. 绝对禁止输出 <think>, <think>, <reasoning> 等任何思考过程标签。\n"
+            "2. 绝对禁止输出'好的'、'根据图片'、'分析如下'等开场白或结束语。\n"
+            "3. 直接输出最终结论，格式必须严格遵守下面的模板。"
+        )
+
+        # 3. 构建针对性的 User Prompt
+        default_task = (
+            "请分析这张图片，描述图片内容，重点关注以下信息\n"
+            "1. 设备识别：图中是什么设备？型号或标签是否可见？\n"
+            "2. 状态评估：仪表盘数值是否在正常区间？指示灯颜色是什么？\n"
+            "3. 异常检测：是否存在漏油、生锈、电缆破损、螺丝松动或烟雾？\n"
+            "4. 环境风险：周围是否有杂物堆放或安全隐患？\n"
+            "请用专业、客观、简短的语言描述。"
+        )
+        
+        final_user_prompt = default_task
+
+        # 4. 初始化客户端
+        client = _get_async_client()
+
+        # 5. 构建请求参数
+        kwargs = {
+            "model": model,
+            "messages": [
+                {"role": "system", "content": system_prompt},
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": final_user_prompt},
+                        {"type": "image_url", "image_url": {"url": image_url}}
+                    ]
+                }
+            ],
+            "temperature": 0.1,  # 低温度以保证事实准确性
+            "stream": False,
+            "max_tokens": LLM_CONFIG.get("max_tokens"),
+        }
+
+        # 尝试通过参数关闭思考 (取决于后端支持情况)
+        kwargs["extra_body"] = {"chat_template_kwargs": {"enable_thinking": False}}
+
+        try:
+            print(f"🚀 正在调用 {model} 进行设备故障分析...")
+            response = await client.chat.completions.create(**kwargs)
+            raw_content = response.choices[0].message.content or ""
+
+            print(f"✅ 分析完成:\n{raw_content}")
+            return raw_content
+
+        except Exception as e:
+            import traceback
+            error_msg = f"❌ 设备故障分析失败：{str(e)}\n{traceback.format_exc()}"
+            print(error_msg)
+            return f"Error: {str(e)}"
+
+    @staticmethod
+    def get_embeddings(embedding_text: str):
+        embedding_base_url = EMBEDDING_CONFIG["base_url"]
+
+        with httpx.Client(timeout=60) as client:
+            response = client.post(
+            embedding_base_url,
+            json={"model": EMBEDDING_CONFIG["model"], "input": embedding_text},
+            headers={"Authorization": EMBEDDING_CONFIG["api_key"]},
+        )
+            embedding_text = response.json()["data"][0]["embedding"]
+        return embedding_text
+
+    @staticmethod
+    async def get_embeddings_async(embedding_text: str) -> List[float]:
+        embedding_base_url = EMBEDDING_CONFIG["base_url"]
+
+        async with httpx.AsyncClient(timeout=60) as client:
+            response = await client.post(
+                embedding_base_url,
+                json={"model": EMBEDDING_CONFIG["model"], "input": embedding_text},
+                headers={"Authorization": EMBEDDING_CONFIG["api_key"]},
+            )
+            embedding = response.json()["data"][0]["embedding"]
+        return embedding
+
+    @staticmethod
+    async def get_embeddings_batch_async(texts: List[str]) -> List[List[float]]:
+        embedding_base_url = EMBEDDING_CONFIG["base_url"]
+
+        async with httpx.AsyncClient(timeout=120) as client:
+            response = await client.post(
+                embedding_base_url,
+                json={"model": EMBEDDING_CONFIG["model"], "input": texts},
+                headers={"Authorization": EMBEDDING_CONFIG["api_key"]},
+            )
+            data = response.json()["data"]
+            sorted_data = sorted(data, key=lambda x: x["index"])
+            embeddings = [item["embedding"] for item in sorted_data]
+        return embeddings
+
+    @staticmethod
+    def cosine_similarity(vec1: Union[List[float], np.ndarray], vec2: Union[List[float], np.ndarray]) -> float:
+        v1 = np.array(vec1) if not isinstance(vec1, np.ndarray) else vec1
+        v2 = np.array(vec2) if not isinstance(vec2, np.ndarray) else vec2
+        
+        norm1 = np.linalg.norm(v1)
+        norm2 = np.linalg.norm(v2)
+        
+        if norm1 == 0 or norm2 == 0:
+            return 0.0
+        
+        return float(np.dot(v1, v2) / (norm1 * norm2))
+
+    @staticmethod
+    def cosine_similarity_batch(query_vec: Union[List[float], np.ndarray], 
+                                  candidates_vecs: List[Union[List[float], np.ndarray]]) -> List[float]:
+        query = np.array(query_vec) if not isinstance(query_vec, np.ndarray) else query_vec
+        candidates = [np.array(v) if not isinstance(v, np.ndarray) else v for v in candidates_vecs]
+        
+        query_norm = np.linalg.norm(query)
+        if query_norm == 0:
+            return [0.0] * len(candidates)
+        
+        similarities = []
+        for candidate in candidates:
+            cand_norm = np.linalg.norm(candidate)
+            if cand_norm == 0:
+                similarities.append(0.0)
+            else:
+                sim = float(np.dot(query, candidate) / (query_norm * cand_norm))
+                similarities.append(sim)
+        
+        return similarities
+
+    @staticmethod
+    async def hybrid_match_with_embeddings(
+        query_text: str,
+        candidates: List[dict],
+        name_key: str = "display_name",
+        embedding_key: str = "embedding",
+        text_weight: float = 0.3,
+        semantic_weight: float = 0.7,
+        text_threshold: float = 0.3,
+        semantic_threshold: float = 0.5
+    ) -> tuple:
+        if not candidates:
+            return None, 0.0
+        
+        query_embedding = await OpenaiAPI.get_embeddings_async(query_text)
+        
+        scored_candidates = []
+        
+        for idx, candidate in enumerate(candidates):
+            name = candidate.get(name_key, "")
+            if not name:
+                props = candidate.get("props", {})
+                name = props.get("name") or props.get("名称") or props.get("设备名称") or ""
+            
+            text_score = OpenaiAPI._compute_text_similarity(query_text, name)
+            
+            props = candidate.get("props", {})
+            candidate_embedding = props.get(embedding_key) or candidate.get(embedding_key)
+            
+            if candidate_embedding:
+                semantic_score = OpenaiAPI.cosine_similarity(query_embedding, candidate_embedding)
+            else:
+                semantic_score = 0.0
+            
+            combined_score = text_weight * text_score + semantic_weight * semantic_score
+            
+            scored_candidates.append({
+                "candidate": candidate,
+                "text_score": text_score,
+                "semantic_score": semantic_score,
+                "combined_score": combined_score,
+                "index": idx
+            })
+        
+        scored_candidates.sort(key=lambda x: x["combined_score"], reverse=True)
+        
+        best = scored_candidates[0]
+        best_candidate = best["candidate"]
+        best_score = best["combined_score"]
+        
+        min_threshold = max(text_threshold * text_weight + semantic_threshold * semantic_weight, 0.3)
+        
+        if best_score < min_threshold:
+            return None, best_score
+        
+        return best_candidate, best_score
+
+    @staticmethod
+    def _compute_text_similarity(a: str, b: str) -> float:
+        if not a or not b:
+            return 0.0
+        a = str(a).strip().lower()
+        b = str(b).strip().lower()
+        if not a or not b:
+            return 0.0
+        if a == b:
+            return 1.0
+        if a in b or b in a:
+            return 0.9
+        
+        set_a, set_b = set(a), set(b)
+        inter = len(set_a & set_b)
+        union = len(set_a | set_b) or 1
+        jaccard = inter / union
+        
+        len_diff = abs(len(a) - len(b))
+        len_penalty = max(0.0, 1.0 - len_diff / max(len(a), len(b), 1))
+        
+        return 0.5 * jaccard + 0.5 * len_penalty
+
+
+    
+
+if __name__ == "__main__":
+    answer = asyncio.run(OpenaiAPI.open_api_chat_without_thinking(
+        query="你好",
+        json_output=True,
+        system_prompt="你是一个助手，请用中文回答问题。请直接给出最终答案，不要包含任何推理步骤、思考过程、解释性文字或前缀。",
+        enable_thinking=True
+    ))
+    print(answer)
+        
+
+

prompts.py

@@ -470,7 +470,7 @@ BAIKE_PROMPTS = {
     # 输入变量: question, answer
     # ------------------------------------------------------------------
     "generate_suggested_replies": (
-        "你是一个专业的公文问答助手。根据以下问答内容，生成两个用户可能关心的后续问题。\n"
+        "你是一个专业的工业设备问答助手。根据以下问答内容，生成两个用户可能关心的后续问题。\n"
         "\n"
         "【用户问题】\n"
         "{question}\n"
@@ -498,7 +498,13 @@ BAIKE_PROMPTS = {
     # 输入变量: 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."
     ),
 
     # ------------------------------------------------------------------
@@ -506,18 +512,20 @@ 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. **重要**：如果参考资料中包含图片链接（格式为 `![图片](URL)` 或 `images/xxx.jpg`），请务必在相关位置自然展示使用，如果参考资料中没有图片，**不要输出任何关于图片的说明**\n"
+        "1. 先直接回答用户问题，再补充必要依据。\n"
-        "2. 如果参考资料包含 LaTeX 公式、连接等，请在对应位置正常使用，不得修改或删除\n"
+        "2. 每个关键事实都必须能在 Retrieved evidence 中找到依据；不要补充资料外结论。\n"
-        "3. 请专注于内容的准确性、完整性和逻辑性，不要担心格式的美观性\n"
+        "3. 如果资料只能说明部分内容，请明确说“资料中只能确认...”和“资料中未提供...”。\n"
-        "4. 可以适当使用简单的段落分隔，但不要使用复杂的Markdown格式\n"
+        "4. LaTeX 格式必须严格区分：变量或很短的表达式使用 $...$；包含等号、求和、积分、分式或较长的完整方程必须使用独立段落，且 $$ 必须各自单独占一行。两条独立公式必须分别使用两个公式块，两个公式块之间保留一个空行，不能写进同一个公式块或紧挨在一起。不要把完整方程嵌在列表正文中。\n"
-        "5. 如果参考资料包含表格，请不要输出表格，而要直接描述表格的内容\n"
+        "5. 参考资料中真实存在的图片可在对应段落附近展示；URL 必须逐字来自资料。\n"
+        "6. 表格内容请转成自然语言描述，不要输出复杂表格。\n"
         "\n"
         "请给出回复："
     ),
@@ -541,6 +549,7 @@ BAIKE_PROMPTS = {
         "1. 请根据您的专业知识提供准确、完整的回答\n"
         "2. 可以适当使用简单的段落分隔，但不要使用复杂的Markdown格式\n"
         "3. 请专注于内容的准确性和专业性\n"
+        "4. LaTeX 格式必须严格区分：变量或很短的表达式使用 $...$；包含等号、求和、积分、分式或较长的完整方程必须使用独立段落，且 $$ 必须各自单独占一行。两条独立公式必须分别使用两个公式块，两个公式块之间保留一个空行，不能写进同一个公式块或紧挨在一起。不要把完整方程嵌在列表正文中。\n"
         "\n"
         "请给出回复："
     ),
@@ -1228,4 +1237,3 @@ OPERATE_PROMPTS = {
         "请给出一般性的操作指导建议："
     ),
 }
-

workflows/__init__.py

@@ -2,7 +2,7 @@
 工作流模块
 包含所有工作流Agent
 """
-from .workflow_fault_diagnosis import run_fault_diagnosis_workflow
+from .workflow_baike import run_baike_workflow
 
 # 其他工作流暂未实现，待后续添加
 # from .workflow_repair import run_repair_workflow
@@ -11,10 +11,11 @@ from .workflow_fault_diagnosis import run_fault_diagnosis_workflow
 # from .workflow_report import run_report_workflow
 
 __all__ = [
-    "run_fault_diagnosis_workflow",
+    "run_baike_workflow",
     # "run_repair_workflow",
     # "run_statistics_workflow",
     # "run_qa_workflow",
     # "run_report_workflow"
 ]
 
+

workflows/history_manager.py

@@ -7,14 +7,17 @@ import json
 from typing import Any, Dict, List, Optional
 
 
+DEFAULT_HISTORY_MAX_MESSAGES = 12
+DEFAULT_ASSISTANT_TRUNCATE = 1000
+
+
 class HistoryManager:
     """统一的历史对话管理器，提供解析、截断、格式化、过滤等功能"""
-    ALLOWED_HISTORY_ROLES = {"user", "assistant"}
 
     def __init__(
         self,
-        max_messages: int = 10,
+        max_messages: int = DEFAULT_HISTORY_MAX_MESSAGES,
-        max_assistant_chars: int = 300,
+        max_assistant_chars: int = DEFAULT_ASSISTANT_TRUNCATE,
     ):
         """
         Args:
@@ -35,18 +38,7 @@ class HistoryManager:
             return []
         try:
             data = json.loads(history_message) if isinstance(history_message, str) else history_message
-            if not isinstance(data, list):
+            return data[-self.max_messages:] if isinstance(data, list) else []
-                return []
-
-            messages = []
-            for message in data:
-                if not isinstance(message, dict):
-                    continue
-                role = message.get("role")
-                content = message.get("content")
-                if role in self.ALLOWED_HISTORY_ROLES and content:
-                    messages.append({"role": role, "content": content})
-            return messages[-self.max_messages:]
         except Exception:
             return []
 
@@ -96,11 +88,7 @@ class HistoryManager:
             return []
         result = []
         for m in messages:
-            if (
+            if isinstance(m, dict) and m.get("role") and m.get("content"):
-                isinstance(m, dict)
-                and m.get("role") in self.ALLOWED_HISTORY_ROLES
-                and m.get("content")
-            ):
                 result.append({"role": m["role"], "content": m["content"]})
         return result
 
@@ -208,8 +196,8 @@ def parse_history(history_message: Any) -> List[Dict[str, Any]]:
 
 def build_history_str(
     history_messages: List[Dict[str, Any]],
-    recent_count: int = 10,
+    recent_count: int = DEFAULT_HISTORY_MAX_MESSAGES,
-    assistant_truncate: int = 300,
+    assistant_truncate: int = DEFAULT_ASSISTANT_TRUNCATE,
 ) -> str:
     """兼容旧调用的模块级函数"""
     mgr = HistoryManager(max_messages=recent_count, max_assistant_chars=assistant_truncate)

workflows/history_manager.py.bak

@@ -0,0 +1,217 @@
+"""
+统一历史对话管理器
+整合所有分散的 parse_history / _parse_history / build_history_str / filter_image_urls 逻辑
+"""
+
+import json
+from typing import Any, Dict, List, Optional
+
+
+class HistoryManager:
+    """统一的历史对话管理器，提供解析、截断、格式化、过滤等功能"""
+
+    def __init__(
+        self,
+        max_messages: int = 10,
+        max_assistant_chars: int = 300,
+    ):
+        """
+        Args:
+            max_messages: 保留的最近消息条数
+            max_assistant_chars: assistant 角色内容截断字符数
+        """
+        self.max_messages = max_messages
+        self.max_assistant_chars = max_assistant_chars
+
+    # ==================== 解析 ====================
+
+    def parse(self, history_message: Any) -> List[Dict[str, Any]]:
+        """
+        解析历史消息为列表，保留最近 max_messages 条。
+        兼容字符串和列表输入，替代所有 parse_history / _parse_history。
+        """
+        if not history_message:
+            return []
+        try:
+            data = json.loads(history_message) if isinstance(history_message, str) else history_message
+            return data[-self.max_messages:] if isinstance(data, list) else []
+        except Exception:
+            return []
+
+    # ==================== 截断 ====================
+
+    def truncate(self, messages: List[Dict[str, Any]], keep_recent: Optional[int] = None) -> List[Dict[str, Any]]:
+        """按数量截断，保留最近 keep_recent 条"""
+        if not messages:
+            return []
+        n = keep_recent if keep_recent is not None else self.max_messages
+        return messages[-n:]
+
+    # ==================== 格式化 ====================
+
+    def to_prompt_str(
+        self,
+        messages: List[Dict[str, Any]],
+        recent_count: Optional[int] = None,
+        assistant_truncate: Optional[int] = None,
+    ) -> str:
+        """
+        格式化为 "用户: xxx\\n助手: xxx" 纯文本字符串，用于 prompt 注入。
+        替代 build_history_str。
+        """
+        if not messages:
+            return ""
+        n = recent_count if recent_count is not None else self.max_messages
+        trunc = assistant_truncate if assistant_truncate is not None else self.max_assistant_chars
+        recent = messages[-n:]
+        parts = []
+        for m in recent:
+            if isinstance(m, dict):
+                role = m.get("role", "")
+                content = (m.get("content") or "").strip()
+                if role == "user":
+                    parts.append(f"用户: {content}")
+                elif role == "assistant":
+                    parts.append(f"助手: {content[:trunc]}")
+        return "\n".join(parts) if parts else ""
+
+    def to_openai_messages(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """
+        格式化为 OpenAI messages 列表，直接传入 LLM。
+        用于百科问答等需要完整消息格式的场景。
+        """
+        if not messages:
+            return []
+        result = []
+        for m in messages:
+            if isinstance(m, dict) and m.get("role") and m.get("content"):
+                result.append({"role": m["role"], "content": m["content"]})
+        return result
+
+    # ==================== 请求预处理 ====================
+
+    def preprocess_from_request(self, history_message: Any, exclude_last: bool = True) -> str:
+        """
+        处理前端传入的历史消息：解析、移除当前用户消息、序列化。
+        对应 app.py 中 history_message 预处理逻辑。
+
+        Args:
+            history_message: 前端传入的历史消息，支持 str 或 list
+            exclude_last: 是否移除最后一条（当前用户消息），默认 True
+
+        Returns:
+            处理后的 JSON 字符串
+        """
+        if not history_message:
+            return ""
+        try:
+            data = json.loads(history_message) if isinstance(history_message, str) else history_message
+        except (json.JSONDecodeError, TypeError):
+            return ""
+        if not isinstance(data, list):
+            return ""
+        if exclude_last and data:
+            data = data[:-1]
+        return json.dumps(data, ensure_ascii=False)
+
+    # ==================== 状态初始化 ====================
+
+    def init_history(self, history_message: Any) -> Dict[str, Any]:
+        """
+        一步完成历史消息解析，返回 history_message(str) + history_messages(list) 双字段。
+        用于各工作流入口初始化 state，消除重复的 parse_history 调用。
+
+        Args:
+            history_message: 历史消息，支持 str 或 list
+
+        Returns:
+            {"history_message": str, "history_messages": list}
+        """
+        parsed = self.parse(history_message)
+        raw = json.dumps(parsed, ensure_ascii=False) if parsed else ""
+        return {"history_message": raw, "history_messages": parsed}
+
+    # ==================== 过滤 ====================
+
+    @staticmethod
+    def filter_image_urls(history_message: Any) -> Any:
+        """
+        过滤历史消息中的图片URL，将多模态内容转为纯文本。
+        替代 main_agent.filter_image_urls。
+        """
+        if not history_message:
+            return history_message
+
+        try:
+            history_messages = json.loads(history_message) if isinstance(history_message, str) else history_message
+        except (json.JSONDecodeError, TypeError):
+            return history_message
+
+        if not isinstance(history_messages, list):
+            return history_messages
+
+        filtered_messages = []
+        for msg in history_messages:
+            if not isinstance(msg, dict):
+                filtered_messages.append(msg)
+                continue
+
+            new_msg = msg.copy()
+            content = msg.get("content")
+
+            if isinstance(content, list):
+                text_parts = []
+                for content_item in content:
+                    if isinstance(content_item, dict):
+                        if content_item.get("type") == "text":
+                            text_value = content_item.get("text", "")
+                            if text_value:
+                                text_parts.append(text_value)
+                    elif isinstance(content_item, str):
+                        text_parts.append(content_item)
+
+                if text_parts:
+                    new_msg["content"] = "".join(text_parts)
+                    filtered_messages.append(new_msg)
+            else:
+                filtered_messages.append(new_msg)
+
+        return filtered_messages
+
+
+# ==================== 模块级便捷实例 ====================
+# 提供与旧函数签名兼容的模块级函数，方便渐进式迁移
+
+_default_manager = HistoryManager()
+
+
+def parse_history(history_message: Any) -> List[Dict[str, Any]]:
+    """兼容旧调用的模块级函数"""
+    return _default_manager.parse(history_message)
+
+
+def build_history_str(
+    history_messages: List[Dict[str, Any]],
+    recent_count: int = 10,
+    assistant_truncate: int = 300,
+) -> str:
+    """兼容旧调用的模块级函数"""
+    mgr = HistoryManager(max_messages=recent_count, max_assistant_chars=assistant_truncate)
+    return mgr.to_prompt_str(history_messages, recent_count=recent_count, assistant_truncate=assistant_truncate)
+
+
+def filter_image_urls(history_message: Any) -> Any:
+    """兼容旧调用的模块级函数"""
+    return HistoryManager.filter_image_urls(history_message)
+
+
+def preprocess_from_request(history_message: Any, exclude_last: bool = True) -> str:
+    """兼容旧调用的模块级函数"""
+    return _default_manager.preprocess_from_request(history_message, exclude_last=exclude_last)
+
+
+def init_history(history_message: Any) -> Dict[str, Any]:
+    """兼容旧调用的模块级函数"""
+    return _default_manager.init_history(history_message)
+
+

workflows/history_manager.py_0723

@@ -0,0 +1,233 @@
+"""
+统一历史对话管理器
+整合所有分散的 parse_history / _parse_history / build_history_str / filter_image_urls 逻辑
+"""
+
+import json
+from typing import Any, Dict, List, Optional
+
+
+class HistoryManager:
+    """统一的历史对话管理器，提供解析、截断、格式化、过滤等功能"""
+    ALLOWED_HISTORY_ROLES = {"user", "assistant"}
+
+    def __init__(
+        self,
+        max_messages: int = 10,
+        max_assistant_chars: int = 300,
+    ):
+        """
+        Args:
+            max_messages: 保留的最近消息条数
+            max_assistant_chars: assistant 角色内容截断字符数
+        """
+        self.max_messages = max_messages
+        self.max_assistant_chars = max_assistant_chars
+
+    # ==================== 解析 ====================
+
+    def parse(self, history_message: Any) -> List[Dict[str, Any]]:
+        """
+        解析历史消息为列表，保留最近 max_messages 条。
+        兼容字符串和列表输入，替代所有 parse_history / _parse_history。
+        """
+        if not history_message:
+            return []
+        try:
+            data = json.loads(history_message) if isinstance(history_message, str) else history_message
+            if not isinstance(data, list):
+                return []
+
+            messages = []
+            for message in data:
+                if not isinstance(message, dict):
+                    continue
+                role = message.get("role")
+                content = message.get("content")
+                if role in self.ALLOWED_HISTORY_ROLES and content:
+                    messages.append({"role": role, "content": content})
+            return messages[-self.max_messages:]
+        except Exception:
+            return []
+
+    # ==================== 截断 ====================
+
+    def truncate(self, messages: List[Dict[str, Any]], keep_recent: Optional[int] = None) -> List[Dict[str, Any]]:
+        """按数量截断，保留最近 keep_recent 条"""
+        if not messages:
+            return []
+        n = keep_recent if keep_recent is not None else self.max_messages
+        return messages[-n:]
+
+    # ==================== 格式化 ====================
+
+    def to_prompt_str(
+        self,
+        messages: List[Dict[str, Any]],
+        recent_count: Optional[int] = None,
+        assistant_truncate: Optional[int] = None,
+    ) -> str:
+        """
+        格式化为 "用户: xxx\\n助手: xxx" 纯文本字符串，用于 prompt 注入。
+        替代 build_history_str。
+        """
+        if not messages:
+            return ""
+        n = recent_count if recent_count is not None else self.max_messages
+        trunc = assistant_truncate if assistant_truncate is not None else self.max_assistant_chars
+        recent = messages[-n:]
+        parts = []
+        for m in recent:
+            if isinstance(m, dict):
+                role = m.get("role", "")
+                content = (m.get("content") or "").strip()
+                if role == "user":
+                    parts.append(f"用户: {content}")
+                elif role == "assistant":
+                    parts.append(f"助手: {content[:trunc]}")
+        return "\n".join(parts) if parts else ""
+
+    def to_openai_messages(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """
+        格式化为 OpenAI messages 列表，直接传入 LLM。
+        用于百科问答等需要完整消息格式的场景。
+        """
+        if not messages:
+            return []
+        result = []
+        for m in messages:
+            if (
+                isinstance(m, dict)
+                and m.get("role") in self.ALLOWED_HISTORY_ROLES
+                and m.get("content")
+            ):
+                result.append({"role": m["role"], "content": m["content"]})
+        return result
+
+    # ==================== 请求预处理 ====================
+
+    def preprocess_from_request(self, history_message: Any, exclude_last: bool = True) -> str:
+        """
+        处理前端传入的历史消息：解析、移除当前用户消息、序列化。
+        对应 app.py 中 history_message 预处理逻辑。
+
+        Args:
+            history_message: 前端传入的历史消息，支持 str 或 list
+            exclude_last: 是否移除最后一条（当前用户消息），默认 True
+
+        Returns:
+            处理后的 JSON 字符串
+        """
+        if not history_message:
+            return ""
+        try:
+            data = json.loads(history_message) if isinstance(history_message, str) else history_message
+        except (json.JSONDecodeError, TypeError):
+            return ""
+        if not isinstance(data, list):
+            return ""
+        if exclude_last and data:
+            data = data[:-1]
+        return json.dumps(data, ensure_ascii=False)
+
+    # ==================== 状态初始化 ====================
+
+    def init_history(self, history_message: Any) -> Dict[str, Any]:
+        """
+        一步完成历史消息解析，返回 history_message(str) + history_messages(list) 双字段。
+        用于各工作流入口初始化 state，消除重复的 parse_history 调用。
+
+        Args:
+            history_message: 历史消息，支持 str 或 list
+
+        Returns:
+            {"history_message": str, "history_messages": list}
+        """
+        parsed = self.parse(history_message)
+        raw = json.dumps(parsed, ensure_ascii=False) if parsed else ""
+        return {"history_message": raw, "history_messages": parsed}
+
+    # ==================== 过滤 ====================
+
+    @staticmethod
+    def filter_image_urls(history_message: Any) -> Any:
+        """
+        过滤历史消息中的图片URL，将多模态内容转为纯文本。
+        替代 main_agent.filter_image_urls。
+        """
+        if not history_message:
+            return history_message
+
+        try:
+            history_messages = json.loads(history_message) if isinstance(history_message, str) else history_message
+        except (json.JSONDecodeError, TypeError):
+            return history_message
+
+        if not isinstance(history_messages, list):
+            return history_messages
+
+        filtered_messages = []
+        for msg in history_messages:
+            if not isinstance(msg, dict):
+                filtered_messages.append(msg)
+                continue
+
+            new_msg = msg.copy()
+            content = msg.get("content")
+
+            if isinstance(content, list):
+                text_parts = []
+                for content_item in content:
+                    if isinstance(content_item, dict):
+                        if content_item.get("type") == "text":
+                            text_value = content_item.get("text", "")
+                            if text_value:
+                                text_parts.append(text_value)
+                    elif isinstance(content_item, str):
+                        text_parts.append(content_item)
+
+                if text_parts:
+                    new_msg["content"] = "".join(text_parts)
+                    filtered_messages.append(new_msg)
+            else:
+                filtered_messages.append(new_msg)
+
+        return filtered_messages
+
+
+# ==================== 模块级便捷实例 ====================
+# 提供与旧函数签名兼容的模块级函数，方便渐进式迁移
+
+_default_manager = HistoryManager()
+
+
+def parse_history(history_message: Any) -> List[Dict[str, Any]]:
+    """兼容旧调用的模块级函数"""
+    return _default_manager.parse(history_message)
+
+
+def build_history_str(
+    history_messages: List[Dict[str, Any]],
+    recent_count: int = 10,
+    assistant_truncate: int = 300,
+) -> str:
+    """兼容旧调用的模块级函数"""
+    mgr = HistoryManager(max_messages=recent_count, max_assistant_chars=assistant_truncate)
+    return mgr.to_prompt_str(history_messages, recent_count=recent_count, assistant_truncate=assistant_truncate)
+
+
+def filter_image_urls(history_message: Any) -> Any:
+    """兼容旧调用的模块级函数"""
+    return HistoryManager.filter_image_urls(history_message)
+
+
+def preprocess_from_request(history_message: Any, exclude_last: bool = True) -> str:
+    """兼容旧调用的模块级函数"""
+    return _default_manager.preprocess_from_request(history_message, exclude_last=exclude_last)
+
+
+def init_history(history_message: Any) -> Dict[str, Any]:
+    """兼容旧调用的模块级函数"""
+    return _default_manager.init_history(history_message)
+
+

workflows/workflow_baike.py_0721

@@ -0,0 +1,597 @@
+"""
+工作流：普通问答Agent（使用LangGraph）
+输入：文本（转化后和原文本query）、图片描述、语音文本
+步骤：
+1. 使用RAG和GraphRAG检索相关信息
+2. 生成回答
+使用 OpenaiAPI 方法调用大模型，尽可能使用异步
+"""
+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
+
+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_current_stream_handler, get_all_callbacks
+from utils.image_utils import extract_image_paths, find_closest_image_path, normalize_markdown_images, get_image_prompt_guidance
+from prompts import BAIKE_PROMPTS, COMMON_PROMPTS
+import asyncio
+import json
+import re
+import random
+
+
+# =============== 定义状态 ===============
+class QAState(TypedDict):
+    """普通问答工作流状态"""
+    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]  # 是否需要检索
+
+    # 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)
+
+
+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
+    )
+    
+    need_retrieval = True
+    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
+
+
+
+    print(f"是否需要检索: {need_retrieval}")
+    return {"need_retrieval": need_retrieval}
+
+
+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[-8:] 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（如果有）
+    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
+    prompt = BAIKE_PROMPTS["rewrite_query_with_history"].format(
+        last_user_query=last_user_query or '（无）',
+        history_str=history_str,
+        original_query=original_query
+    )
+
+    rewritten_query = original_query
+    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
+
+    print("原始检索query:", original_query)
+    print("重写后检索query:", last_user_query+rewritten_query)
+
+    return {"retrieval_query":last_user_query+rewritten_query}
+
+
+@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]
+
+    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=2
+                )
+            else:
+                return await rag_search(
+                    query=extracted_text,
+                    top_k=2
+                )
+
+        async def _do_graph_search():
+            return await graph_rag_search.ainvoke({
+                "query": query,
+                "top_k": 3
+            })
+
+        rag_result, graph_results = await asyncio.gather(
+            _do_rag_search(),
+            _do_graph_search()
+        )
+
+        # 处理 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)
+                    })
+
+        # 只保留前 4 条数据
+#        rag_search_result = rag_search_result[:2]
+        rag_search_result = sorted(rag_search_result,key=lambda x:float(x.get("socre") or 0), reverse=True)[:4]
+
+        return {
+            "rag_search_result": rag_search_result,
+            "graph_search_result": graph_results if isinstance(graph_results, str) else ""
+        }
+    except Exception as e:
+        return {
+            "rag_search_result": [],
+            "graph_search_result": "",
+            "error_message": f"检索失败: {str(e)}"
+        }
+
+
+
+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]
+    )
+    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)
+        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: 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 []
+    need_retrieval = state.get("need_retrieval", True)
+    print("history_message(qa)", history_messages)
+    print("是否使用检索结果:", need_retrieval)
+
+    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
+
+        query_desc = extracted_text if extracted_text 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
+            )
+        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)
+
+        content_messages = []
+        if history_messages:
+            content_messages.extend(history_messages)
+        content_messages.append({"role": "user", "content": content_user_content})
+
+        # 第一步：生成内容
+        raw_content = await OpenaiAPI.open_api_chat_without_thinking(
+            model=None,
+            system_prompt=content_system_prompt,
+            messages=content_messages,
+            temperature=0.5
+        )
+
+        # 过滤掉异常标签（如果有）
+        raw_content = re.sub(r'ynchroneg>.*?ost switching>', '', raw_content, flags=re.DOTALL)
+        raw_content = raw_content.strip() if raw_content else "抱歉，无法生成回答。"
+
+        title_options = ["🤖 诊断分析中", "🤖 故障预检中"]
+        start_event = {
+            "type": "function_execution",
+            "title": random.choice(title_options),
+            "details": raw_content[:30] + "..."
+        }
+        for callback in get_all_callbacks():
+            try:
+                callback(start_event)
+            except Exception:
+                pass
+
+
+        # ========== 第二步：专注于规范格式 ==========
+        # 构建第二步的提示词：只关注格式规范，不修改内容
+        format_system_prompt = COMMON_PROMPTS["format_system"]
+        
+        format_user_content = COMMON_PROMPTS["format_user_with_image_examples"].format(raw_content=raw_content)
+
+        # 第二步：规范格式（流式输出）
+        stream_handler = get_current_stream_handler()
+        answer = ""
+        accumulated_content = ""  # 用于逐步发送给前端的内容
+        
+        # 使用 async for 遍历流式生成器
+        async for chunk in OpenaiAPI.open_api_chat_stream(
+            model=None,
+            system_prompt=format_system_prompt,
+            messages=[{"role": "user", "content": format_user_content}],
+            temperature=0.1
+        ):
+            answer += chunk
+            accumulated_content += chunk
+            # 每收到新的 chunk 就发送一次更新
+            if stream_handler:
+                stream_handler.send_stream_content(accumulated_content)
+
+        answer = answer.strip() if answer else raw_content
+
+        # LaTeX 公式空格规范化
+        def normalize_latex_spacing(text: str) -> str:
+            placeholder = "\x00DOUBLE_DOLLAR\x00"
+            text = text.replace("$$", placeholder)
+            text = re.sub(r'(?<!\s)\$', r' $', text)
+            text = re.sub(r'\$(?!\s)', r'$ ', text)
+            text = text.replace(placeholder, "$$")
+            text = re.sub(r'(?<!\s)\$\$', r' $$', text)
+            text = re.sub(r'\$\$(?!\s)', r'$$ ', text)
+            return text
+
+        answer = normalize_latex_spacing(answer)
+
+        # 使用增强的图片路径匹配与规范化（仅在有检索结果时）
+        if need_retrieval and original_image_paths:
+            answer = normalize_markdown_images(answer, original_paths=original_image_paths)
+        print("llm output", answer)
+
+        # 生成建议回复问题
+        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"
+
+
+# =============== 构建工作流 ===============
+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"
+        }
+    )
+    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 = ""
+) -> 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": "",
+        "route_flag": route_flag,
+        "history_message": history_message,
+        "history_messages": [],
+        "need_retrieval": None,
+        "rag_search_result": None,
+        "graph_search_result": None,
+        "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": "qa",
+            "suggestedReplies": final_state.get("suggestedReplies", [])
+        }
+    
+    except Exception as e:
+        return {
+            "response": f"普通问答工作流执行失败: {str(e)}",
+            "actions": [],
+            "result_tag": "qa",
+            "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())
+
+
+
+
+