优化

backend/api/chat_router.py

@@ -26,7 +26,7 @@ from core.llm_catalog import (
     resolve_to_api_model,
     validate_request_can_use_provider,
 )
-from core.database import get_db_pool, get_checkpointer
+from core.database import get_db_pool, get_checkpointer, reset_db_pools_on_connection_error
 from core.mcp_client import get_mcp_client
 from core.dependencies import get_current_user, get_moderation_service
 from core.exceptions import ModerationError
@@ -68,7 +68,7 @@ from services.chat_thread_service import (
     get_knowledge_graph_tool_flags,
 )
 from services.chat_message_file_service import ChatMessageFileService
-from services.chat_message_service import ChatMessageService  # 新增：消息保存服务
+from services.chat_message_service import ChatMessageService, normalize_message_content
 from utils.helpers import BaseResponse
 from logger.logging import get_logger
 
@@ -622,6 +622,7 @@ async def chat_completion(
                 logger.error(f"[V2] 保存消息到 chat_messages 表失败: {save_err}")
                     
         except Exception as e:
+            await reset_db_pools_on_connection_error(e)
             logger.exception(f"聊天接口错误: {e}")
             yield json.dumps({"error": str(e)}, ensure_ascii=False)
         
@@ -957,7 +958,7 @@ async def _save_messages_to_chat_messages_table(
                     continue
                 
                 msg_type = msg.type
-                msg_content = getattr(msg, 'content', '') or ''
+                msg_content = normalize_message_content(getattr(msg, 'content', '') or '')
                 
                 # 检查单条消息是否已存在
                 existing = await conn.fetchval(

backend/core/database.py

@@ -9,6 +9,7 @@ from typing import Optional
 import asyncio
 
 import asyncpg
+import psycopg
 from psycopg_pool import AsyncConnectionPool
 from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
 
@@ -24,23 +25,15 @@ _psycopg_pool: Optional[AsyncConnectionPool] = None
 _checkpointer: Optional[AsyncPostgresSaver] = None
 
 
-async def get_db_pool() -> asyncpg.Pool:
+async def _create_asyncpg_pool() -> asyncpg.Pool:
-    """
+    """创建 asyncpg 连接池，带指数退避重试。"""
-    获取或创建 asyncpg 数据库连接池
-    
-    用于一般的数据库 CRUD 操作。
-    """
-    global _asyncpg_pool
-    
-    if _asyncpg_pool is None:
-        logger.info(f"初始化 asyncpg 数据库连接池: {settings.db_user}@{settings.db_host}:{settings.db_port}/{settings.db_name}")
-        
     max_retries = 3
-        retry_delay = 2  # 秒
+    retry_delay = 2
 
     for attempt in range(max_retries):
+        pool: Optional[asyncpg.Pool] = None
         try:
-                _asyncpg_pool = await asyncpg.create_pool(
+            pool = await asyncpg.create_pool(
                 host=settings.db_host,
                 port=settings.db_port,
                 database=settings.db_name,
@@ -49,94 +42,166 @@ async def get_db_pool() -> asyncpg.Pool:
                 min_size=settings.db_pool_min_size,
                 max_size=settings.db_pool_max_size,
                 command_timeout=settings.db_command_timeout,
-                    timeout=30,  # 连接超时 30 秒
+                timeout=30,
+                max_inactive_connection_lifetime=300,  # 闲置 5 分钟后自动丢弃，防止连接腐烂
                 server_settings={
                     'application_name': 'huoyan-enterprise',
-                        'jit': 'off'  # 禁用 JIT 以提高稳定性
+                    'jit': 'off',
-                    }
+                },
             )
-                
+            async with pool.acquire() as conn:
-                # 测试连接
+                await conn.execute("SELECT 1")
-                async with _asyncpg_pool.acquire() as _conn:
+                await ensure_graph_metadata(conn)
-                    await _conn.execute("SELECT 1")
-                    await ensure_graph_metadata(_conn)
-                
             logger.info("asyncpg 数据库连接池初始化成功")
-                break
+            return pool
-                
         except Exception as e:
-                logger.error(f"asyncpg 数据库连接池初始化失败 (尝试 {attempt + 1}/{max_retries}): {e}")
+            logger.error(f"asyncpg 连接池初始化失败 (尝试 {attempt + 1}/{max_retries}): {e}")
-                
+            if pool is not None:
-                if _asyncpg_pool is not None:
                 try:
-                        await _asyncpg_pool.close()
+                    await pool.close()
-                    except:
+                except Exception:
                     pass
-                    _asyncpg_pool = None
-                
             if attempt < max_retries - 1:
                 logger.info(f"将在 {retry_delay} 秒后重试...")
                 await asyncio.sleep(retry_delay)
-                    retry_delay *= 2  # 指数退避
+                retry_delay *= 2
             else:
                 logger.error("数据库连接池初始化失败，已达到最大重试次数")
                 raise
 
+
+async def _discard_asyncpg_pool() -> None:
+    """关闭并清除 asyncpg 连接池全局变量，下次调用 get_db_pool 时会重建。"""
+    global _asyncpg_pool
+    if _asyncpg_pool is not None:
+        try:
+            await _asyncpg_pool.close()
+        except Exception:
+            pass
+        _asyncpg_pool = None
+        reset_graph_metadata()
+        logger.info("asyncpg 连接池已重置")
+
+
+async def get_db_pool() -> asyncpg.Pool:
+    """
+    获取或创建 asyncpg 数据库连接池。
+
+    若检测到连接池已关闭（如数据库重启后连接全部断开），
+    会自动丢弃旧池并重建，无需重启服务。
+    """
+    global _asyncpg_pool
+
+    # 检查现有 pool 是否已被关闭
+    if _asyncpg_pool is not None and _asyncpg_pool._closed:
+        logger.warning("检测到 asyncpg 连接池已关闭，将重新初始化")
+        _asyncpg_pool = None
+        reset_graph_metadata()
+
+    if _asyncpg_pool is None:
+        logger.info(
+            f"初始化 asyncpg 数据库连接池: "
+            f"{settings.db_user}@{settings.db_host}:{settings.db_port}/{settings.db_name}"
+        )
+        _asyncpg_pool = await _create_asyncpg_pool()
+
     return _asyncpg_pool
 
 
+async def _create_psycopg_checkpointer() -> tuple[AsyncConnectionPool, AsyncPostgresSaver]:
+    """创建 psycopg 连接池与 LangGraph Checkpointer，带指数退避重试。"""
+    max_retries = 3
+    retry_delay = 2
+
+    for attempt in range(max_retries):
+        pool: Optional[AsyncConnectionPool] = None
+        try:
+            pool = AsyncConnectionPool(
+                conninfo=settings.db_uri_psycopg,
+                max_size=settings.checkpointer_pool_max_size,
+                open=False,
+                timeout=30,
+                max_idle=300,  # 闲置 5 分钟后丢弃，避免复用已断开的连接
+                kwargs={
+                    "autocommit": True,
+                    "prepare_threshold": 0,
+                },
+            )
+            await pool.open()
+
+            checkpointer = AsyncPostgresSaver(pool)
+            await checkpointer.setup()
+
+            logger.info("Checkpointer 初始化成功")
+            return pool, checkpointer
+        except Exception as e:
+            logger.error(f"Checkpointer 初始化失败 (尝试 {attempt + 1}/{max_retries}): {e}")
+            if pool is not None:
+                try:
+                    await pool.close()
+                except Exception:
+                    pass
+            if attempt < max_retries - 1:
+                logger.info(f"将在 {retry_delay} 秒后重试...")
+                await asyncio.sleep(retry_delay)
+                retry_delay *= 2
+            else:
+                logger.error("Checkpointer 初始化失败，已达到最大重试次数")
+                raise
+
+
+async def _discard_psycopg_pool() -> None:
+    """关闭并清除 psycopg 连接池与 Checkpointer，下次调用 get_checkpointer 时会重建。"""
+    global _psycopg_pool, _checkpointer
+    if _psycopg_pool is not None:
+        try:
+            await _psycopg_pool.close()
+        except Exception:
+            pass
+    _psycopg_pool = None
+    _checkpointer = None
+    logger.info("psycopg 连接池与 Checkpointer 已重置")
+
+
+def is_db_connection_error(exc: BaseException) -> bool:
+    """判断异常是否由数据库连接断开引起。"""
+    if isinstance(
+        exc,
+        (
+            asyncpg.PostgresConnectionError,
+            asyncpg.TooManyConnectionsError,
+            asyncpg.InterfaceError,
+            psycopg.OperationalError,
+            psycopg.InterfaceError,
+            OSError,
+        ),
+    ):
+        return True
+    cause = exc.__cause__
+    return isinstance(cause, BaseException) and is_db_connection_error(cause)
+
+
+async def reset_db_pools_on_connection_error(exc: BaseException) -> None:
+    """连接异常时重置相关连接池，使后续请求可自动恢复。"""
+    if not is_db_connection_error(exc):
+        return
+    logger.error(f"检测到数据库连接异常，正在重置连接池: {exc}")
+    await _discard_asyncpg_pool()
+    await _discard_psycopg_pool()
+
+
 async def get_checkpointer() -> AsyncPostgresSaver:
     """
     获取或创建 LangGraph Checkpointer
 
     使用 psycopg AsyncConnectionPool，用于 LangGraph 的状态持久化。
+    若连接池已失效，会自动丢弃并重建，无需重启服务。
     """
     global _psycopg_pool, _checkpointer
 
     if _checkpointer is None:
         logger.info("初始化 psycopg 连接池和 Checkpointer...")
-        
+        _psycopg_pool, _checkpointer = await _create_psycopg_checkpointer()
-        max_retries = 3
-        retry_delay = 2  # 秒
-        
-        for attempt in range(max_retries):
-            try:
-                _psycopg_pool = AsyncConnectionPool(
-                    conninfo=settings.db_uri_psycopg,
-                    max_size=settings.checkpointer_pool_max_size,
-                    open=False,
-                    timeout=30,  # 连接超时 30 秒
-                    kwargs={
-                        "autocommit": True,
-                        "prepare_threshold": 0
-                    },
-                )
-                await _psycopg_pool.open()
-                
-                _checkpointer = AsyncPostgresSaver(_psycopg_pool)
-                await _checkpointer.setup()
-                
-                logger.info("Checkpointer 初始化成功")
-                break
-                
-            except Exception as e:
-                logger.error(f"Checkpointer 初始化失败 (尝试 {attempt + 1}/{max_retries}): {e}")
-                
-                if _psycopg_pool is not None:
-                    try:
-                        await _psycopg_pool.close()
-                    except:
-                        pass
-                    _psycopg_pool = None
-                    _checkpointer = None
-                
-                if attempt < max_retries - 1:
-                    logger.info(f"将在 {retry_delay} 秒后重试...")
-                    await asyncio.sleep(retry_delay)
-                    retry_delay *= 2  # 指数退避
-                else:
-                    logger.error("Checkpointer 初始化失败，已达到最大重试次数")
-                    raise
 
     return _checkpointer
 
@@ -145,7 +210,6 @@ async def close_db_pool():
     """关闭所有数据库连接池"""
     global _asyncpg_pool, _psycopg_pool, _checkpointer
 
-    # 关闭 asyncpg 连接池
     if _asyncpg_pool is not None:
         logger.info("关闭 asyncpg 数据库连接池...")
         await _asyncpg_pool.close()
@@ -153,7 +217,6 @@ async def close_db_pool():
         reset_graph_metadata()
         logger.info("asyncpg 数据库连接池已关闭")
 
-    # 关闭 psycopg 连接池
     if _psycopg_pool is not None:
         logger.info("关闭 psycopg 连接池...")
         await _psycopg_pool.close()
@@ -163,8 +226,23 @@ async def close_db_pool():
 
 
 async def get_db_connection():
-    """获取数据库连接（用于依赖注入）"""
+    """
+    获取数据库连接（用于 FastAPI 依赖注入）。
+
+    若 acquire 因连接断开而失败，会自动重置连接池，
+    下一次请求将重新建立连接，无需重启服务。
+    本次请求仍会返回 500，但不会永久卡死。
+    """
+    try:
         pool = await get_db_pool()
         async with pool.acquire() as connection:
             yield connection
+    except (
+        asyncpg.PostgresConnectionError,
+        asyncpg.TooManyConnectionsError,
+        asyncpg.InterfaceError,
+        OSError,
+    ) as e:
+        await reset_db_pools_on_connection_error(e)
+        raise
 

backend/services/chat_message_service.py

@@ -4,13 +4,38 @@
 用于保存和查询用户原始消息和AI响应，替代从 checkpoint 中解析
 """
 import json
-from typing import List, Dict, Any, Optional
+from typing import List, Dict, Any, Optional, Union
 import asyncpg
 from logger.logging import get_logger
 
 logger = get_logger(__name__)
 
 
+def normalize_message_content(content: Union[str, list, dict, None]) -> str:
+    """将 LangChain 消息 content（可能是 str 或 content blocks 列表）规范为可落库的字符串。"""
+    if content is None:
+        return ""
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts: list[str] = []
+        for block in content:
+            if isinstance(block, str):
+                if block:
+                    parts.append(block)
+            elif isinstance(block, dict):
+                text = block.get("text") or block.get("content")
+                if text is not None and str(text).strip():
+                    parts.append(str(text))
+            elif block is not None:
+                parts.append(str(block))
+        return "\n".join(parts)
+    if isinstance(content, dict):
+        text = content.get("text") or content.get("content")
+        return str(text) if text is not None else json.dumps(content, ensure_ascii=False)
+    return str(content)
+
+
 class ChatMessageService:
     """聊天消息服务类"""
     
@@ -59,8 +84,8 @@ class ChatMessageService:
                 checkpoint_id,
                 message_index,
                 'user',
-                content,
+                normalize_message_content(content),
-                injected_content,
+                normalize_message_content(injected_content) if injected_content else None,
                 has_files,
                 json.dumps(metadata) if metadata else None
             )
@@ -112,7 +137,7 @@ class ChatMessageService:
                 checkpoint_id,
                 message_index,
                 'assistant',
-                content,
+                normalize_message_content(content),
                 json.dumps(metadata) if metadata else None
             )
             
@@ -166,7 +191,7 @@ class ChatMessageService:
                 checkpoint_id,
                 message_index,
                 'tool',
-                content,
+                normalize_message_content(content),
                 name,
                 json.dumps(metadata) if metadata else None
             )

frontend/src/stores/chat.js

@@ -13,7 +13,7 @@ function generateUUID() {
 }
 
 /** LangChain 落库多为块数组 / 结构化 content，需压成可读字符串再给模板与 Markdown */
-function normalizeMessageContent(val) {
+export function normalizeMessageContent(val) {
   if (val == null) return ''
   if (typeof val === 'string') return val
   if (typeof val === 'number' || typeof val === 'boolean') return String(val)
@@ -184,6 +184,11 @@ export const useChatStore = defineStore('chat', () => {
       }
       
       const { step, content = '', reasoning_content = '', tool_calls = null, metadata = null, messageData = null } = stepData
+      const normalizedContent = normalizeMessageContent(content)
+      const uiTypeFromData = messageData?.type
+        ? normalizeStreamMessageUiType(messageData.type)
+        : null
+      const isToolStep = uiTypeFromData === 'tool'
       
       // 查找是否已存在该步骤
       let stepIndex = lastMessage.steps.findIndex(s => s.step === step)
@@ -204,9 +209,13 @@ export const useChatStore = defineStore('chat', () => {
       
       const currentStep = lastMessage.steps[stepIndex]
       
-      // 更新步骤内容（增量更新）
+      // 更新步骤内容（流式增量；tool 消息多为整块 content blocks，需先 normalize）
-      if (content) {
+      if (normalizedContent) {
-        currentStep.content = currentStep.content + content
+        if (isToolStep || currentStep.messageType === 'tool') {
+          currentStep.content = normalizedContent
+        } else {
+          currentStep.content = currentStep.content + normalizedContent
+        }
       }
       
       // 更新思考内容（增量更新）

frontend/src/views/Chat.vue

@@ -432,7 +432,7 @@
                         {{ tc.function?.name || tc.name }}
                       </div>
                       <div v-if="tc.function?.arguments || tc.args" class="tool-call-args">
-                        <code>{{ tc.function?.arguments || tc.args }}</code>
+                        <code>{{ formatToolArgs(tc.function?.arguments || tc.args) }}</code>
                       </div>
                     </div>
                   </div>
@@ -872,7 +872,7 @@
 import { ref, onMounted, onUnmounted, nextTick, watch, computed } from 'vue'
 import { useRouter, useRoute } from 'vue-router'
 import { useAuthStore } from '../stores/auth'
-import { useChatStore } from '../stores/chat'
+import { useChatStore, normalizeMessageContent } from '../stores/chat'
 import { useKnowledgeBaseStore } from '../stores/knowledgeBase'
 import { marked } from 'marked'
 import axios from 'axios'
@@ -1247,11 +1247,23 @@ function formatJSON(content) {
   }
 }
 
+// 将 tool call 的 args 字符串格式化为可读形式（流式未完成时允许非法 JSON）
+function formatToolArgs(raw) {
+  if (!raw) return ''
+  const s = typeof raw === 'string' ? raw : JSON.stringify(raw)
+  try {
+    return JSON.stringify(JSON.parse(s), null, 2)
+  } catch {
+    return s
+  }
+}
+
 // 渲染 JSON 为 HTML（带语法高亮）
 function renderJSON(content) {
-  if (!content) return ''
+  const text = normalizeMessageContent(content)
+  if (!text) return ''
   try {
-    const formatted = formatJSON(content)
+    const formatted = formatJSON(text)
     // 转义 HTML 特殊字符
     const escaped = formatted
       .replace(/&/g, '&amp;')
@@ -1270,7 +1282,7 @@ function renderJSON(content) {
     return `<pre class="json-content"><code>${highlighted}</code></pre>`
   } catch (e) {
     // 如果格式化失败，返回原内容（转义后）
-    const escaped = content
+    const escaped = text
       .replace(/&/g, '&amp;')
       .replace(/</g, '&lt;')
       .replace(/>/g, '&gt;')
@@ -1778,28 +1790,56 @@ async function sendMessage() {
               
               // 如果存在 langgraph_step，使用步骤方式更新
               if (langgraphStep !== undefined && langgraphStep !== null) {
-                // 获取内容（兼容旧格式和新格式）
+                // 获取内容（兼容 str / LangChain content blocks 数组）
-                const content = messageData.content || ''
+                const content = normalizeMessageContent(messageData.content)
                 
                 // 获取思考内容
                 const reasoning_content = messageData.additional_kwargs?.reasoning_content || ''
                 
-                // 获取工具调用（优先使用 tool_calls，如果没有则使用 tool_call_chunks）
+                // 获取当前步骤已积累的 tool_calls，用于增量合并
+                const existingStep = currentMessage.steps?.find(s => s.step === langgraphStep)
+                const prevToolCalls = existingStep?.tool_calls ? existingStep.tool_calls.map(tc => ({ ...tc, function: { ...tc.function } })) : []
+
+                // 收集本帧 args 增量：tool_call_chunks 始终包含完整的增量 args 片段，
+                // invalid_tool_calls 只是 LangChain 对不完整 JSON 的分类，数据与 tool_call_chunks 重复，不再额外合并
+                const argChunks = (messageData.tool_call_chunks || [])
+                const frameToolCalls = messageData.tool_calls || []
+                const hasNewData = frameToolCalls.some(tc => tc.name || tc.id) || argChunks.some(c => c.args || c.name)
+
                 let tool_calls = null
-                if (messageData.tool_calls && messageData.tool_calls.length > 0) {
+                if (hasNewData || prevToolCalls.length > 0) {
-                  tool_calls = messageData.tool_calls
+                  // 以前一帧积累的 tool_calls 为基础，逐步合并
-                } else if (messageData.tool_call_chunks && messageData.tool_call_chunks.length > 0) {
+                  const merged = prevToolCalls.map(tc => ({
-                  // 将 tool_call_chunks 转换为 tool_calls 格式
+                    ...tc,
-                  tool_calls = messageData.tool_call_chunks.map(chunk => ({
+                    function: { ...(tc.function || { name: '', arguments: '' }) }
-                    id: chunk.id || '',
-                    type: chunk.type || 'function',
-                    function: {
-                      name: chunk.name || '',
-                      arguments: chunk.args || ''
-                    },
-                    name: chunk.name,
-                    args: chunk.args
                   }))
+
+                  // 应用本帧 tool_calls 中非空的 name / id（保留已有值）
+                  frameToolCalls.forEach((tc, idx) => {
+                    if (!merged[idx]) {
+                      merged[idx] = { name: '', args: '', id: '', function: { name: '', arguments: '' } }
+                    }
+                    if (tc.name) { merged[idx].name = tc.name; merged[idx].function.name = tc.name }
+                    if (tc.id) { merged[idx].id = tc.id }
+                  })
+
+                  // 追加 args 增量（来自 tool_call_chunks 和 invalid_tool_calls）
+                  argChunks.forEach(chunk => {
+                    const idx = typeof chunk.index === 'number' ? chunk.index : 0
+                    if (!merged[idx]) {
+                      merged[idx] = { name: chunk.name || '', args: '', id: chunk.id || '', function: { name: chunk.name || '', arguments: '' } }
+                    }
+                    if (chunk.args) {
+                      merged[idx].args = (merged[idx].args || '') + chunk.args
+                      merged[idx].function.arguments = merged[idx].args
+                    }
+                    if (chunk.name && !merged[idx].name) {
+                      merged[idx].name = chunk.name
+                      merged[idx].function.name = chunk.name
+                    }
+                  })
+
+                  if (merged.length > 0) tool_calls = merged
                 }
                 
                 // 获取当前步骤的完整内容（包括之前的内容）
@@ -1845,7 +1885,7 @@ async function sendMessage() {
                 })
               } else {
                 // 兼容旧格式：没有 langgraph_step 的情况
-                const content = messageData.content || ''
+                const content = normalizeMessageContent(messageData.content)
                 const newContent = currentMessage.content + content
                 
                 // 合并 additional_kwargs（保留已有的，更新新的）