diff --git a/apps/application/models/application_chat.py b/apps/application/models/application_chat.py
index 5b87fb26f38..473628a76ff 100644
--- a/apps/application/models/application_chat.py
+++ b/apps/application/models/application_chat.py
@@ -10,12 +10,11 @@
 import uuid_utils.compat as uuid
 from django.contrib.postgres.fields import ArrayField
 from django.db import models
-from django.utils.translation import gettext as _
-from langchain_core.messages import HumanMessage, AIMessage
 
 from application.models import Application
 from common.encoder.encoder import SystemEncoder
 from common.mixins.app_model_mixin import AppModelMixin
+from common.utils.messages_util import to_ai_message_list, to_human_message_list
 from users.models import User
 
 
@@ -130,35 +129,10 @@ class ChatRecord(AppModelMixin):
     workflow_context = models.JSONField(verbose_name="工作流上下文", default=dict, null=True, blank=True)
 
     def get_human_message(self):
-        # 用户消息取自 question（{content, image_list, ...}），历史上下文用文本部分
-        question = self.question if isinstance(self.question, dict) else {"content": self.question or ""}
-        return [HumanMessage(content=question.get("content", "") or "")]
+        return to_human_message_list(self.question)
 
     def get_ai_message(self):
-        # 答案取自 messages 中的 TEXT / TOOL 内容块（REASONING/FORM/FAILURE 不进历史），按顺序保留交错。
-        # 注意：type 用字面量，避免 models 反向依赖 application.workflow.ContentType
-        ai_message_list = []
-        for m in self.messages or []:
-            if not isinstance(m, dict):
-                continue
-            m_type = m.get("type")
-            if m_type == "TEXT":
-                if m.get("content"):
-                    ai_message_list.append(AIMessage(content=m.get("content")))
-            elif m_type == "TOOL":
-                # 工具调用：名称 + 入参 + 结果 拼成一段
-                tool_parts = [str(p) for p in (m.get("content"), m.get("arguments"), m.get("result")) if p]
-                if tool_parts:
-                    ai_message_list.append(AIMessage(content="\n".join(tool_parts)))
-        if len(ai_message_list) == 0:
-            ai_message_list = [
-                AIMessage(
-                    content=_(
-                        "Sorry, no relevant content was found. Please re-describe your problem or provide more information. "
-                    )
-                )
-            ]
-        return ai_message_list
+        return to_ai_message_list(self.messages)
 
     def get_node_details_runtime_node_id(self, runtime_node_id):
         return self.details.get(runtime_node_id, None)
diff --git a/apps/common/utils/messages_util.py b/apps/common/utils/messages_util.py
new file mode 100644
index 00000000000..b29645df1d3
--- /dev/null
+++ b/apps/common/utils/messages_util.py
@@ -0,0 +1,72 @@
+# coding=utf-8
+"""
+@project: MaxKB
+@Author:  虎虎虎
+@file:    messages_util.py
+@date:    2023/9/11 11:45
+@desc:    ChatRecord 存储的 question / messages 与 LangChain 消息之间的转换工具
+"""
+
+import json
+
+import uuid_utils.compat as uuid
+from django.utils.translation import gettext as _
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+
+
+def to_human_message_list(question):
+    """
+    将用户消息转换为 HumanMessage 列表。
+    question 为 {content, image_list, ...} 结构，历史上下文只取文本部分。
+    """
+    question = question if isinstance(question, dict) else {"content": question or ""}
+    return [HumanMessage(content=question.get("content", "") or "")]
+
+
+def to_ai_message_list(messages):
+    """
+    将 messages 中的 TEXT / TOOL 内容块转换为 LangChain 消息列表。
+    REASONING/FORM/FAILURE 不进历史；按顺序保留交错。
+    注意：type 用字面量，避免对 application.workflow.ContentType 产生反向依赖。
+    """
+    ai_message_list = []
+    for m in messages or []:
+        if not isinstance(m, dict):
+            continue
+        m_type = m.get("type")
+        if m_type == "TEXT":
+            if m.get("content"):
+                ai_message_list.append(AIMessage(content=m.get("content")))
+        elif m_type == "TOOL":
+            # 工具调用按 OpenAI/LangChain 协议拆成两条消息：
+            # 1. AIMessage 携带 tool_calls（名称 + 入参）
+            # 2. ToolMessage 携带结果，通过 tool_call_id 与上一条对应
+            tool_name = m.get("content")
+            if not tool_name:
+                continue
+            # arguments 存储为 JSON 字符串，需还原为 dict 供 tool_calls 使用
+            raw_arguments = m.get("arguments")
+            try:
+                args = json.loads(raw_arguments) if isinstance(raw_arguments, str) and raw_arguments else {}
+            except (json.JSONDecodeError, ValueError):
+                args = {}
+            if not isinstance(args, dict):
+                args = {"arguments": args}
+            # tool_call_id 必须让 AIMessage 与 ToolMessage 一一对应，否则模型侧会报错
+            tool_call_id = m.get("id") or str(uuid.uuid7())
+            ai_message_list.append(
+                AIMessage(
+                    content="",
+                    tool_calls=[{"name": tool_name, "args": args, "id": tool_call_id, "type": "tool_call"}],
+                )
+            )
+            ai_message_list.append(ToolMessage(content=m.get("result") or "", tool_call_id=tool_call_id))
+    if len(ai_message_list) == 0:
+        ai_message_list = [
+            AIMessage(
+                content=_(
+                    "Sorry, no relevant content was found. Please re-describe your problem or provide more information. "
+                )
+            )
+        ]
+    return ai_message_list