From 10e1cafb491bb122404edba1f1376482af1dad51 Mon Sep 17 00:00:00 2001 From: shaohuzhang1 Date: Mon, 19 Oct 2026 19:36:23 +0800 Subject: [PATCH] feat: Add LLM node --- apps/application/serializers/common.py | 37 +- apps/application/workflow/content_type.py | 1 + .../message/aggregator/aggregator_factory.py | 4 +- .../message/aggregator/impl/__init__.py | 3 +- .../aggregator/impl/html_aggregator.py | 28 ++ .../workflow/message/struct/html_content.py | 22 ++ .../workflow/nodes/llm_node/__init__.py | 10 + .../workflow/nodes/llm_node/llm_node.py | 357 +++++++++++++++++ .../nodes/llm_node/message_converter.py | 209 ++++++++++ .../workflow/nodes/reply_node/reply_node.py | 90 ++++- apps/chat/serializers/chat.py | 33 +- .../impl/response/openai_to_response.py | 4 +- .../impl/response/system_to_response.py | 2 +- apps/common/utils/messages_util.py | 4 +- apps/locales/en_US/LC_MESSAGES/django.po | 6 + apps/locales/zh_CN/LC_MESSAGES/django.po | 6 + apps/locales/zh_Hant/LC_MESSAGES/django.po | 6 + .../0009_toolworkflowversion_publish_desc.py | 15 + apps/tools/models/tool_workflow.py | 1 + apps/tools/serializers/tool_import.py | 140 +++++++ apps/tools/serializers/tool_version.py | 90 +++-- apps/tools/serializers/tool_workflow.py | 17 +- apps/tools/views/tool_workflow.py | 11 +- .../components/content-list/index.vue | 24 +- .../components/content/index.vue | 8 +- .../items/group-tool/components/ToolItem.vue | 161 ++++++++ .../content/items/group-tool/index.vue | 122 ++++++ .../components/content/items/group/index.vue | 155 ++++++++ .../components/content/items/html.vue | 25 ++ .../content/items/tool/CreateFile.vue | 205 ++++++++++ .../content/items/tool/DownloadSkills.vue | 157 ++++++++ .../components/content/items/tool/Glob.vue | 192 +++++++++ .../components/content/items/tool/Grep.vue | 156 ++++++++ .../components/content/items/tool/ListDir.vue | 159 ++++++++ .../content/items/tool/ListSkills.vue | 150 +++++++ .../components/content/items/tool/Patch.vue | 326 ++++++++++++++++ .../content/items/tool/ReadFile.vue | 206 ++++++++++ .../content/items/tool/RunSkill.vue | 175 +++++++++ .../content/items/tool/Terminal.vue | 170 ++++++++ .../items/tool/components/ScrollDiff.vue | 40 ++ .../items/tool/extract-partial-json.ts | 367 ++++++++++++++++++ .../components/content/items/tool/index.vue | 51 +++ .../content/items/tool/patch/FileDiff.vue | 164 ++++++++ .../content/items/tool/patch/parsePatch.ts | 281 ++++++++++++++ .../content/items/tool/patch/patchTypes.ts | 23 ++ ui/src/conversation-panel/core/aggregators.ts | 31 +- ui/src/workflow-canvas/config/node-data.ts | 21 + ui/src/workflow-canvas/config/node-mapping.ts | 1 + .../workflow-canvas/icons/llm-node-icon.vue | 6 + ui/src/workflow-canvas/index.vue | 8 +- ui/src/workflow-canvas/node-menu/menu.ts | 8 +- .../llm-node/component/ToolFunctionDialog.vue | 213 ++++++++++ .../nodes/llm-node/component/ToolsSetting.vue | 174 +++++++++ .../nodes/llm-node/details/index.vue | 72 ++++ .../workflow-canvas/nodes/llm-node/index.ts | 17 + .../workflow-canvas/nodes/llm-node/index.vue | 198 ++++++++++ .../workflow-canvas/nodes/llm-node/types.ts | 46 +++ ui/src/workflow-canvas/plugins/dagre.ts | 5 +- ui/src/workflow-canvas/types.ts | 1 + ui/vite.config.ts | 19 +- 60 files changed, 5136 insertions(+), 97 deletions(-) create mode 100644 apps/application/workflow/message/aggregator/impl/html_aggregator.py create mode 100644 apps/application/workflow/message/struct/html_content.py create mode 100644 apps/application/workflow/nodes/llm_node/__init__.py create mode 100644 apps/application/workflow/nodes/llm_node/llm_node.py create mode 100644 apps/application/workflow/nodes/llm_node/message_converter.py create mode 100644 apps/tools/migrations/0009_toolworkflowversion_publish_desc.py create mode 100644 apps/tools/serializers/tool_import.py create mode 100644 ui/src/conversation-panel/components/content/items/group-tool/components/ToolItem.vue create mode 100644 ui/src/conversation-panel/components/content/items/group-tool/index.vue create mode 100644 ui/src/conversation-panel/components/content/items/group/index.vue create mode 100644 ui/src/conversation-panel/components/content/items/html.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/CreateFile.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/DownloadSkills.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/Glob.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/Grep.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/ListDir.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/ListSkills.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/Patch.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/ReadFile.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/RunSkill.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/Terminal.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/components/ScrollDiff.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/extract-partial-json.ts create mode 100644 ui/src/conversation-panel/components/content/items/tool/index.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/patch/FileDiff.vue create mode 100644 ui/src/conversation-panel/components/content/items/tool/patch/parsePatch.ts create mode 100644 ui/src/conversation-panel/components/content/items/tool/patch/patchTypes.ts create mode 100644 ui/src/workflow-canvas/icons/llm-node-icon.vue create mode 100644 ui/src/workflow-canvas/nodes/llm-node/component/ToolFunctionDialog.vue create mode 100644 ui/src/workflow-canvas/nodes/llm-node/component/ToolsSetting.vue create mode 100644 ui/src/workflow-canvas/nodes/llm-node/details/index.vue create mode 100644 ui/src/workflow-canvas/nodes/llm-node/index.ts create mode 100644 ui/src/workflow-canvas/nodes/llm-node/index.vue create mode 100644 ui/src/workflow-canvas/nodes/llm-node/types.ts diff --git a/apps/application/serializers/common.py b/apps/application/serializers/common.py index c1da528f3dc..e1c93da2ab1 100644 --- a/apps/application/serializers/common.py +++ b/apps/application/serializers/common.py @@ -164,8 +164,8 @@ def resolve_chat_user(chat_user_id, chat_user_type, asker=None): if asker: if isinstance(asker, dict): return asker - return {"username": asker} - return {"username": "游客"} + return {"username": asker, "id": str(chat_user_id)} + return {"username": "游客", "id": str(chat_user_id)} def resolve_chat_user_group(chat_user): @@ -390,33 +390,16 @@ def save_chat(self): ).save() def set_chat_variable(self, chat_context): - if not self.debug: - chat = QuerySet(Chat).filter(id=self.chat_id).first() - if chat: - chat.meta = {**(chat.meta if isinstance(chat.meta, dict) else {}), **chat_context} - chat.save() - else: - cache.set( - Cache_Version.CHAT_VARIABLE.get_key(key=self.chat_id), - chat_context, - version=Cache_Version.CHAT_VARIABLE.get_version(), - timeout=60 * 30, - ) + chat = QuerySet(Chat).filter(id=self.chat_id).first() + if chat: + chat.meta = {**(chat.meta if isinstance(chat.meta, dict) else {}), **chat_context} + chat.save() def get_chat_variable(self): - if not self.debug: - chat = QuerySet(Chat).filter(id=self.chat_id).first() - if chat: - return chat.meta - return {} - else: - return ( - cache.get( - Cache_Version.CHAT_VARIABLE.get_key(key=self.chat_id), - version=Cache_Version.CHAT_VARIABLE.get_version(), - ) - or {} - ) + chat = QuerySet(Chat).filter(id=self.chat_id).first() + if chat: + return chat.meta + return {} def append_chat_record(self, chat_record: ChatRecord): chat_record.problem_text = chat_record.problem_text[0:10240] if chat_record.problem_text is not None else "" diff --git a/apps/application/workflow/content_type.py b/apps/application/workflow/content_type.py index 193cf713cd2..84f22571b26 100644 --- a/apps/application/workflow/content_type.py +++ b/apps/application/workflow/content_type.py @@ -12,6 +12,7 @@ class ContentType(Enum): TEXT = "TEXT" + HTML = "HTML" REASONING = "REASONING" FAILURE = "FAILURE" TOOL = "TOOL" diff --git a/apps/application/workflow/message/aggregator/aggregator_factory.py b/apps/application/workflow/message/aggregator/aggregator_factory.py index 7f79d6ee1fc..b0d582a0924 100644 --- a/apps/application/workflow/message/aggregator/aggregator_factory.py +++ b/apps/application/workflow/message/aggregator/aggregator_factory.py @@ -9,12 +9,13 @@ from typing import Dict, Type, Optional from application.workflow.message.aggregator.content_aggregator import ContentAggregator -from application.workflow.message.aggregator.impl import FormAggregator +from application.workflow.message.aggregator.impl import FormAggregator, HtmlAggregator from application.workflow.message.aggregator.impl.reasoning_aggregator import ReasoningAggregator from application.workflow.message.aggregator.impl.text_aggregator import TextAggregator from application.workflow.message.aggregator.impl.tool_aggregator import ToolAggregator from application.workflow.message.struct.content import Content from application.workflow.message.struct.form_content import FormContent +from application.workflow.message.struct.html_content import HtmlContent from application.workflow.message.struct.reasoning_content import ReasoningContent from application.workflow.message.struct.text_content import TextContent from application.workflow.message.struct.tool_content import ToolContent @@ -31,6 +32,7 @@ class AggregatorFactory: ReasoningContent: ReasoningAggregator(), ToolContent: ToolAggregator(), FormContent: FormAggregator(), + HtmlContent: HtmlAggregator(), } @classmethod diff --git a/apps/application/workflow/message/aggregator/impl/__init__.py b/apps/application/workflow/message/aggregator/impl/__init__.py index 34a83e07003..66969800a6b 100644 --- a/apps/application/workflow/message/aggregator/impl/__init__.py +++ b/apps/application/workflow/message/aggregator/impl/__init__.py @@ -10,5 +10,6 @@ from application.workflow.message.aggregator.impl.reasoning_aggregator import ReasoningAggregator from application.workflow.message.aggregator.impl.tool_aggregator import ToolAggregator from application.workflow.message.aggregator.impl.form_aggregator import FormAggregator +from application.workflow.message.aggregator.impl.html_aggregator import HtmlAggregator -__all__ = ["TextAggregator", "ReasoningAggregator", "ToolAggregator", "FormAggregator"] +__all__ = ["TextAggregator", "ReasoningAggregator", "ToolAggregator", "FormAggregator", "HtmlAggregator"] diff --git a/apps/application/workflow/message/aggregator/impl/html_aggregator.py b/apps/application/workflow/message/aggregator/impl/html_aggregator.py new file mode 100644 index 00000000000..1e33f1ddf42 --- /dev/null +++ b/apps/application/workflow/message/aggregator/impl/html_aggregator.py @@ -0,0 +1,28 @@ +# coding=utf-8 +""" +@project: MaxKB +@file： html_aggregator.py +@date：2026/10/16 +@desc: HtmlContent 聚合器 +""" + +from application.workflow.message.aggregator.content_aggregator import ContentAggregator +from application.workflow.message.struct.html_content import HtmlContent + + +class HtmlAggregator(ContentAggregator[HtmlContent]): + """ + HTML 内容聚合器 + 与文本一致,按顺序拼接内容块 + """ + + def aggregate(self, prev: HtmlContent, chunk: HtmlContent) -> HtmlContent: + if prev is None: + return chunk + return HtmlContent( + chunk.id or prev.id, + (prev.content or "") + (chunk.content or ""), + chunk.status or prev.status, + chunk.node_info or prev.node_info, + chunk.position or prev.position, + ) diff --git a/apps/application/workflow/message/struct/html_content.py b/apps/application/workflow/message/struct/html_content.py new file mode 100644 index 00000000000..3bdf9d40884 --- /dev/null +++ b/apps/application/workflow/message/struct/html_content.py @@ -0,0 +1,22 @@ +# coding=utf-8 +""" +@project: MaxKB +@file： html_content.py +@date：2026/10/16 +@desc: HTML 内容,前端净化后渲染 +""" + +from application.workflow.content_type import ContentType +from application.workflow.message.struct.content import Content, NodeInfo, Position +from application.workflow.status import Status + + +class HtmlContent(Content): + def __init__(self, _id, content: str, status: Status, node_info: NodeInfo, position: Position, **kwargs): + self.content = content + super().__init__(_id, status, ContentType.HTML, node_info, position, **kwargs) + + def to_dict(self): + result = super().to_dict() + result["content"] = self.content + return result diff --git a/apps/application/workflow/nodes/llm_node/__init__.py b/apps/application/workflow/nodes/llm_node/__init__.py new file mode 100644 index 00000000000..dcdebceb1bd --- /dev/null +++ b/apps/application/workflow/nodes/llm_node/__init__.py @@ -0,0 +1,10 @@ +# coding=utf-8 +""" +@project: MaxKB +@Author：虎虎虎 +@file： __init__.py +@date：2026/10/16 +@desc: +""" + +from .llm_node import LLMNode diff --git a/apps/application/workflow/nodes/llm_node/llm_node.py b/apps/application/workflow/nodes/llm_node/llm_node.py new file mode 100644 index 00000000000..abc5a3f2496 --- /dev/null +++ b/apps/application/workflow/nodes/llm_node/llm_node.py @@ -0,0 +1,357 @@ +# coding=utf-8 +""" +@project: MaxKB +@Author：虎虎虎 +@file： llm_node.py +@date：2026/10/16 +@desc: LLM 节点:单次调用大模型(不执行工具),输出 content/reasoning_content/refusal/is_refusal/usage/tool_calls/finish_reason +""" + +import contextlib +import json + +import uuid_utils.compat as uuid +from django.db.models import QuerySet +from django.utils.translation import gettext_lazy as _ +from langchain_core.messages import HumanMessage, SystemMessage +from rest_framework import serializers + +from application.workflow.common import AsyncRunner, CancelledException, WorkflowType +from application.workflow.i_node import INode +from application.workflow.message.aggregator import AggregationManager +from application.workflow.message.struct.content import Content, NodeInfo, Position +from application.workflow.message.struct.reasoning_content import ReasoningContent +from application.workflow.message.struct.text_content import TextContent +from application.workflow.message.struct.tool_content import ToolContent +from application.workflow.nodes.llm_node.message_converter import to_messages +from application.workflow.status import Status +from common.exception.app_exception import AppApiException +from common.utils.count_tokens import count_local_tokens +from models_provider.models import Model +from models_provider.tools import get_model_credential, get_model_instance_by_model_workspace_id + + +class LLMNodeSerializer(serializers.Serializer): + model_id = serializers.CharField(required=False, allow_blank=True, allow_null=True, label=_("Model id")) + model_id_type = serializers.CharField(required=False, default="custom", label=_("Model id type")) + model_id_reference = serializers.ListField( + required=False, child=serializers.CharField(), allow_empty=True, label=_("Reference Field") + ) + model_params_setting = serializers.DictField(required=False, label=_("Model parameter settings")) + system = serializers.CharField(required=False, allow_blank=True, allow_null=True, label=_("Role Setting")) + prompt = serializers.CharField(required=False, allow_blank=True, allow_null=True, label=_("Prompt word")) + # 上下文:enable_context 为真时从 context_variable 引用的变量取消息列表,否则不带上下文 + enable_context = serializers.BooleanField(required=False, default=False, label=_("Custom context")) + context_variable = serializers.ListField( + required=False, child=serializers.CharField(), allow_empty=True, label=_("Context variable") + ) + # 工具定义(OpenAI function 格式),只交给模型做决策,不在本节点执行 + tools = serializers.DictField(required=False, allow_null=True, label=_("Tools")) + is_result = serializers.BooleanField(required=False, default=False, label=_("Whether to return content")) + + def is_valid(self, *, raise_exception=False): + super().is_valid(raise_exception=True) + # reference / default 在运行时才解析,此处只校验自定义模型 + if (self.data.get("model_id_type") or "custom") in ("reference", "default"): + return + model_id = self.data.get("model_id") + if not model_id or not QuerySet(Model).filter(id=model_id).exists(): + raise AppApiException(500, _("The model of the node does not exist")) + + +def _get_default_model_params_setting(model_id): + model = QuerySet(Model).filter(id=model_id).first() + credential = get_model_credential(model.provider, model.model_type, model.model_name) + return credential.get_model_params_setting_form(model.model_name).get_default_form_data() + + +class ToolCallAccumulator: + """ + 流式工具调用累积:为每个调用分配一个 uuid7,流式 ToolContent 与 tool_calls 输出共用,原始 id 另存为 original_id。 + 响应 id 只出现在调用的首个分片:按 index 识别同一调用,无 index 时按原始 id,两者都没有时归属最近一次调用。 + 不依赖 LangChain 的分片合并(无 index 时其会把后续分片误判为新调用)。 + """ + + def __init__(self): + self._key_to_id = {} + self._calls = {} + self._last_id = None + + def append(self, index, original_id, name, args): + """累积一个分片,返回该调用的 uuid7;无法归属时返回 None""" + key = index if index is not None else original_id + if key is None: + tool_call_id = self._last_id + else: + if key not in self._key_to_id: + self._key_to_id[key] = str(uuid.uuid7()) + self._calls[self._key_to_id[key]] = {"original_id": None, "name": "", "arguments": ""} + tool_call_id = self._key_to_id[key] + if tool_call_id is None: + return None + call = self._calls[tool_call_id] + call["original_id"] = call["original_id"] or original_id + call["name"] = call["name"] or name or "" + call["arguments"] += args or "" + self._last_id = tool_call_id + return tool_call_id + + def name_of(self, tool_call_id): + """调用名只在首个分片返回,后续分片输出时用累积到的名称补齐""" + return self._calls[tool_call_id]["name"] + + def tool_calls(self): + return [ + { + "index": index, + "id": tool_call_id, + "original_id": call["original_id"], + "type": "function", + "name": call["name"], + "arguments": call["arguments"] or "{}", + } + for index, (tool_call_id, call) in enumerate(self._calls.items()) + ] + + +def _message_to_dict(message): + """执行详情中展示的消息,保留工具调用信息便于排查上下文""" + result = {"content": message.content, "role": message.type} + if getattr(message, "tool_calls", None): + result["tool_calls"] = [ + { + "id": tc.get("id"), + "name": tc.get("name"), + "arguments": json.dumps(tc.get("args") or {}, ensure_ascii=False), + } + for tc in message.tool_calls + ] + if getattr(message, "tool_call_id", None): + result["tool_call_id"] = message.tool_call_id + return result + + +def _to_openai_tool(tool): + function = tool.get("function") or {} + result = {"name": function.get("name"), "description": function.get("description") or ""} + if function.get("parameters"): + result["parameters"] = function.get("parameters") + return {"type": tool.get("type") or "function", "function": result} + + +class LLMNode(INode): + serializer_class = LLMNodeSerializer + supported_workflow_type_list = [WorkflowType.APPLICATION, WorkflowType.KNOWLEDGE, WorkflowType.TOOL] + type = "llm-node" + + def __init__(self, node, workflow_manage, get_node_parameters): + super().__init__(node, workflow_manage, get_node_parameters) + self.async_runner = AsyncRunner() + self.is_result = False + + def cancel(self): + self.async_runner.cancel() + super().cancel() + + def write(self, message: Content): + super().write(message) + self.data.setdefault("messages", []).append(message) + + def _output(self, message: Content): + """仅在开启返回内容时输出给用户""" + if self.is_result: + self.write(message) + + def execute(self): + workflow_params = self.get_workflow_parameters() + node_params = self.get_parameters() + self.is_result = node_params.get("is_result", False) + + model_id = node_params.get("model_id") + model_params_setting = node_params.get("model_params_setting") + model_id_type = node_params.get("model_id_type", "custom") + model_id_reference = node_params.get("model_id_reference") + if model_id_type == "reference" and model_id_reference: + reference_data = self.workflow_manage.get_reference_field(model_id_reference[0], model_id_reference[1:]) + if reference_data and isinstance(reference_data, dict): + model_id = reference_data.get("model_id", model_id) + model_params_setting = reference_data.get("model_params_setting") + if model_id_type == "default": + default_model_setting = (workflow_params.get("default_model_setting") or {}).get("LLM") or {} + if isinstance(default_model_setting, dict): + model_id = default_model_setting.get("model_id", model_id) + model_params_setting = default_model_setting.get("model_params_setting") + if not model_id: + raise Exception(_("Model is not allowed to be empty")) + if model_params_setting is None: + model_params_setting = _get_default_model_params_setting(model_id) + chat_model = get_model_instance_by_model_workspace_id( + model_id, workflow_params.get("workspace_id"), **(model_params_setting or {}) + ) + + message_list = self._build_messages(node_params) + self.write_context("message_list", [_message_to_dict(m) for m in message_list]) + + tools = self._get_tools(node_params.get("tools")) + model = chat_model.bind_tools(tools) if tools else chat_model + + self._check_cancelled() + self.async_runner.run_async(lambda: self._stream_response(model, chat_model, message_list)) + + def _build_messages(self, node_params): + if node_params.get("enable_context") and node_params.get("context_variable"): + context_variable = node_params.get("context_variable") + messages = to_messages(self.workflow_manage.get_reference_field(context_variable[0], context_variable[1:])) + else: + messages = [] + + system = self.workflow_manage.generate_prompt(node_params.get("system") or "") + if system: + # 上下文首条已是 system 时合并为一条,避免出现多条 system(部分模型要求 system 唯一) + if messages and isinstance(messages[0], SystemMessage): + existing = messages[0].content if isinstance(messages[0].content, str) else "" + messages[0] = SystemMessage(f"{system}\n\n{existing}" if existing.strip() else system) + else: + messages.insert(0, SystemMessage(system)) + + prompt = self.workflow_manage.generate_prompt(node_params.get("prompt") or "") + if prompt: + messages.append(HumanMessage(prompt)) + return messages + + def _get_tools(self, tools_config): + if not tools_config: + return [] + if tools_config.get("location") == "reference": + reference = tools_config.get("reference") or [] + tools = self.workflow_manage.get_reference_field(reference[0], reference[1:]) if reference else None + else: + tools = tools_config.get("tools") + if isinstance(tools, str): + tools = json.loads(tools) + return [_to_openai_tool(t) for t in (tools or []) if isinstance(t, dict)] + + async def _stream_response(self, model, chat_model, message_list): + node_info = NodeInfo(self.get_node_id(), self.get_node_name(), Status.RUNNING) + position = Position(self.get_node_id()) + reasoning_content_id = str(uuid.uuid7()) + text_content_id = str(uuid.uuid7()) + tool_call_accumulator = ToolCallAccumulator() + reasoning_content = "" + reasoning_end = False + full = None + finish_reason = "error" + response = model.astream(message_list) + try: + async for chunk in response: + self._check_cancelled() + full = chunk if full is None else full + chunk + reasoning_chunk = chunk.additional_kwargs.get("reasoning_content") + if reasoning_chunk and not reasoning_end: + reasoning_content += reasoning_chunk + self._output( + ReasoningContent(reasoning_content_id, reasoning_chunk, Status.RUNNING, node_info, position) + ) + if isinstance(chunk.content, str) and chunk.content: + if reasoning_content and not reasoning_end: + reasoning_end = True + self._output(ReasoningContent(reasoning_content_id, "", Status.SUCCESS, node_info, position)) + self._output(TextContent(text_content_id, chunk.content, Status.RUNNING, node_info, position)) + for tc in chunk.tool_call_chunks or []: + tool_call_id = tool_call_accumulator.append( + tc.get("index"), tc.get("id"), tc.get("name"), tc.get("args") + ) + if not tool_call_id: + continue + if tc.get("name") or tc.get("args"): + self._output( + ToolContent( + tool_call_id, + tool_call_accumulator.name_of(tool_call_id), + tc.get("args") or "", + "", + Status.RUNNING, + node_info, + position, + ) + ) + finish_reason = None + except CancelledException: + finish_reason = "cancelled" + raise + finally: + self._output( + TextContent( + text_content_id, + "", + Status.SUCCESS, + NodeInfo(self.get_node_id(), self.get_node_name(), Status.SUCCESS), + position, + ) + ) + # 无论正常结束、抛异常还是被取消中断,都落盘已累积内容并关闭 LLM 流式响应底层连接 + self._write_final_context( + chat_model, message_list, full, reasoning_content, finish_reason, tool_call_accumulator.tool_calls() + ) + with contextlib.suppress(Exception): + await response.aclose() + + def _write_final_context(self, chat_model, message_list, message, reasoning_content, finish_reason, tool_calls): + content = message.content if message is not None and isinstance(message.content, str) else "" + additional_kwargs = message.additional_kwargs if message is not None else {} + response_metadata = message.response_metadata if message is not None else {} + refusal = additional_kwargs.get("refusal") + + usage_metadata = message.usage_metadata if message is not None else None + if usage_metadata: + prompt_tokens = usage_metadata.get("input_tokens", 0) + completion_tokens = usage_metadata.get("output_tokens", 0) + elif self.async_runner.cancel_event.is_set(): + # 中断时模型已不可用，只能走本地 tokenizer 计算 + prompt_tokens, completion_tokens = count_local_tokens(message_list, content) + else: + prompt_tokens = chat_model.get_num_tokens_from_messages(message_list) + completion_tokens = chat_model.get_num_tokens(content) + + if finish_reason is None: + finish_reason = response_metadata.get("finish_reason") or response_metadata.get("stop_reason") + if not finish_reason: + finish_reason = "tool_calls" if tool_calls else "stop" + self.write_context("content", content) + self.write_context("reasoning_content", reasoning_content) + self.write_context("refusal", refusal) + self.write_context("is_refusal", bool(refusal)) + self.write_context( + "usage", + { + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "total_tokens": prompt_tokens + completion_tokens, + }, + ) + self.write_context("tool_calls", tool_calls) + self.write_context("finish_reason", finish_reason) + self.data["message_tokens"] = prompt_tokens + self.data["answer_tokens"] = completion_tokens + + def get_details(self, index: int = 0, position: dict = None, old_details: dict = None, **kwargs): + details = super().get_details(index, position, old_details, **kwargs) + aggregation = AggregationManager() + for m in self.data.get("messages") or []: + aggregation.aggregate(m) + details.update( + { + "message_list": self.get_context("message_list"), + "content": self.get_context("content"), + "reasoning_content": self.get_context("reasoning_content"), + "refusal": self.get_context("refusal"), + "is_refusal": self.get_context("is_refusal"), + "usage": self.get_context("usage"), + "tool_calls": self.get_context("tool_calls"), + "finish_reason": self.get_context("finish_reason"), + "message_tokens": self.data.get("message_tokens"), + "answer_tokens": self.data.get("answer_tokens"), + "messages": aggregation.get_contents(), + } + ) + return details diff --git a/apps/application/workflow/nodes/llm_node/message_converter.py b/apps/application/workflow/nodes/llm_node/message_converter.py new file mode 100644 index 00000000000..c26933c4755 --- /dev/null +++ b/apps/application/workflow/nodes/llm_node/message_converter.py @@ -0,0 +1,209 @@ +# coding=utf-8 +""" +@project: MaxKB +@Author：虎虎虎 +@file： message_converter.py +@date：2026/10/16 +@desc: 把上下文变量中的消息列表转换为 LangChain 消息 + +支持两种元素格式,可混用: +1. 内容项,按 type 区分(其他 type 忽略): + - SYSTEM {"type": "SYSTEM", "content": "..."} + - USER {"type": "USER", "content": "...", "images": [...], "videos": [...], "files": [...]} + - ASSISTANT {"type": "ASSISTANT", "content": "..."} 模型回答 + - TOOL {"type": "TOOL", "id": "调用id", "name": "函数名", "arguments": "{...}", "content": "工具结果", + "ai_id": "同一次模型回复的id", "reasoning": "...", "reasoning_key": "reasoning_content"} + type 为 function(LLM 节点 tool_calls 输出)或缺省 type 但带 name / arguments 的元素同样视为 TOOL。 + 相邻同类型内容合并为一条消息;相邻且 ai_id 相同的 TOOL 合并为一条带 tool_calls 的 assistant 消息, + 随后每个 TOOL 生成一条 tool 消息。TOOL 字段同时兼容 toolName / functionArguments / aiId / reasoningKey 写法。 +2. OpenAI 消息:{"role": "system|user|assistant|tool", "content": ..., "tool_calls": [...], "tool_call_id": "..."} + 以及字符串(视为 user 消息)和 LangChain BaseMessage。 +""" + +import json + +from langchain_core.messages import ( + AIMessage, + BaseMessage, + HumanMessage, + SystemMessage, + ToolMessage, + convert_to_messages, +) + +from application.workflow.nodes.ai_chat_node.ai_chat_node import _process_images + +CONTENT_TYPES = {"SYSTEM", "USER", "ASSISTANT", "TOOL"} +# OpenAI 消息中需要透传给 LangChain 的字段 +OPENAI_MESSAGE_KEYS = ("role", "content", "name", "tool_calls", "tool_call_id") +TOOL_NAME_KEYS = ("name", "tool_name", "toolName", "function_name") +TOOL_ARGUMENT_KEYS = ("arguments", "function_arguments", "functionArguments") + + +def _content_type(item: dict): + """识别内容项类型;非内容项返回 None""" + _type = str(item.get("type") or "").upper() + if _type in CONTENT_TYPES: + return _type + # LLM 节点输出的 tool_calls(type=function)及未声明 type 的工具调用 + if _type == "FUNCTION" or (not _type and any(k in item for k in TOOL_NAME_KEYS + TOOL_ARGUMENT_KEYS)): + return "TOOL" + return None + + +def _first(item: dict, *keys, default=None): + for key in keys: + value = item.get(key) + if value is not None: + return value + return default + + +def _text(value) -> str: + if value is None: + return "" + return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False) + + +def _parse_arguments(arguments) -> dict: + if isinstance(arguments, dict): + return arguments + try: + parsed = json.loads(arguments or "{}") + return parsed if isinstance(parsed, dict) else {} + except (TypeError, ValueError): + return {} + + +def _user_parts(item: dict) -> list: + parts = [] + images = item.get("images") or [] + parts.extend(_process_images([{"url": i} if isinstance(i, str) else i for i in images])) + for video in item.get("videos") or []: + url = video if isinstance(video, str) else (video or {}).get("url") + if url: + parts.append({"type": "video_url", "video_url": {"url": url}}) + file_names = [ + (f if isinstance(f, str) else (f or {}).get("name") or (f or {}).get("file_name") or "unknown") + for f in item.get("files") or [] + ] + if file_names: + parts.append({"type": "text", "text": "Files: " + ", ".join(file_names)}) + return parts + + +class _Converter: + def __init__(self): + self.result: list[BaseMessage] = [] + self.system = "" + self.user = "" + self.user_parts: list = [] + self.assistant = "" + self.tool_group: list[dict] = [] + + def flush_system(self): + if self.system: + self.result.append(SystemMessage(self.system)) + self.system = "" + + def flush_user(self): + if not self.user and not self.user_parts: + return + if self.user_parts: + text = [{"type": "text", "text": self.user}] if self.user else [] + self.result.append(HumanMessage(content=[*text, *self.user_parts])) + else: + self.result.append(HumanMessage(self.user)) + self.user = "" + self.user_parts = [] + + def flush_assistant(self): + if self.assistant: + self.result.append(AIMessage(self.assistant)) + self.assistant = "" + + def flush_tools(self): + if not self.tool_group: + return + tool_calls = [ + { + "id": _text(item.get("id")), + "name": _text(_first(item, *TOOL_NAME_KEYS)), + "args": _parse_arguments(_first(item, *TOOL_ARGUMENT_KEYS)), + "type": "tool_call", + } + for item in self.tool_group + ] + additional_kwargs = {} + # 同一次模型回复共享 reasoning,只要声明了 reasoning_key 就保留(空字符串也保留) + for item in self.tool_group: + reasoning_key = _first(item, "reasoning_key", "reasoningKey") + if reasoning_key: + additional_kwargs[reasoning_key] = item.get("reasoning") or "" + break + self.result.append(AIMessage(content="", tool_calls=tool_calls, additional_kwargs=additional_kwargs)) + for item, tool_call in zip(self.tool_group, tool_calls): + self.result.append(ToolMessage(content=_text(item.get("content")), tool_call_id=tool_call["id"])) + self.tool_group = [] + + def flush_all(self): + self.flush_tools() + self.flush_system() + self.flush_user() + self.flush_assistant() + + def add_content(self, item: dict, _type: str): + if _type == "TOOL": + self.flush_system() + self.flush_user() + self.flush_assistant() + ai_id = _first(item, "ai_id", "aiId") + if self.tool_group and not (ai_id and ai_id == _first(self.tool_group[0], "ai_id", "aiId")): + self.flush_tools() + self.tool_group.append(item) + return + content = _text(item.get("content")) + self.flush_tools() + if _type == "SYSTEM": + self.flush_user() + self.flush_assistant() + self.system += content + elif _type == "USER": + self.flush_system() + self.flush_assistant() + self.user += content + self.user_parts.extend(_user_parts(item)) + elif _type == "ASSISTANT": + self.flush_system() + self.flush_user() + self.assistant += content + + def add_message(self, message: BaseMessage): + self.flush_all() + self.result.append(message) + + def convert(self, value) -> list[BaseMessage]: + for item in value: + if isinstance(item, BaseMessage): + self.add_message(item) + elif isinstance(item, str): + self.add_message(HumanMessage(item)) + elif isinstance(item, dict): + if "role" in item: + self.add_message(convert_to_messages([{k: item[k] for k in OPENAI_MESSAGE_KEYS if k in item}])[0]) + elif _type := _content_type(item): + self.add_content(item, _type) + self.flush_all() + return self.result + + +def to_messages(value) -> list[BaseMessage]: + """把引用变量里的消息列表转换为 LangChain 消息;非数组或 JSON 字符串解析失败时返回空列表""" + if isinstance(value, str): + try: + value = json.loads(value) + except ValueError: + return [] + if not isinstance(value, list): + return [] + return _Converter().convert(value) diff --git a/apps/application/workflow/nodes/reply_node/reply_node.py b/apps/application/workflow/nodes/reply_node/reply_node.py index f40b8f53373..1eea729f69b 100644 --- a/apps/application/workflow/nodes/reply_node/reply_node.py +++ b/apps/application/workflow/nodes/reply_node/reply_node.py @@ -7,7 +7,9 @@ @desc: """ +import json from typing import List + import uuid_utils.compat as uuid from django.utils.translation import gettext_lazy as _ from rest_framework import serializers @@ -15,17 +17,68 @@ from application.workflow.common import WorkflowType from application.workflow.i_node import INode from application.workflow.message.struct.content import NodeInfo, Position +from application.workflow.message.struct.html_content import HtmlContent from application.workflow.message.struct.text_content import TextContent +from application.workflow.message.struct.tool_content import ToolContent from application.workflow.status import Status class ReplyNodeSerializer(serializers.Serializer): reply_type = serializers.CharField(required=True, label=_("Response Type")) + # 输出内容类型:TEXT 文本(默认) / HTML / TOOL 工具调用 + content_type = serializers.ChoiceField( + required=False, default="TEXT", choices=["TEXT", "HTML", "TOOL"], label=_("Content Type") + ) fields = serializers.ListField(required=False, label=_("Reference Field")) content = serializers.CharField(required=False, allow_blank=True, allow_null=True, label=_("Direct answer content")) is_result = serializers.BooleanField(required=False, label=_("Whether to return content")) +def _to_text(value): + return "" if value is None else str(value) + + +def _to_json_text(value): + if value is None: + return "" + return value if isinstance(value, str) else json.dumps(value, ensure_ascii=False) + + +def _first(item: dict, *keys): + for key in keys: + if item.get(key) is not None: + return item.get(key) + return None + + +def _to_tool_list(value): + """ + 把工具回复内容归一化为工具调用列表。 + 支持对象或对象数组(或其 JSON 字符串),元素字段: + id 调用id(缺省自动生成) / name 函数名 / arguments 参数 / content 工具结果; + 同时兼容 toolName、functionArguments、function_name、function_arguments、result 写法。 + """ + if isinstance(value, str): + try: + value = json.loads(value) if value.strip() else [] + except ValueError: + raise Exception(_("Tool reply content must be a JSON object or array")) + if isinstance(value, dict): + value = [value] + if not isinstance(value, list): + raise Exception(_("Tool reply content must be a JSON object or array")) + return [ + { + "id": _to_text(item.get("id")) or str(uuid.uuid7()), + "name": _to_text(_first(item, "name", "toolName", "tool_name", "function_name")), + "arguments": _to_json_text(_first(item, "arguments", "functionArguments", "function_arguments")), + "content": _to_json_text(_first(item, "content", "result")), + } + for item in value + if isinstance(item, dict) + ] + + class ReplyNode(INode): serializer_class = ReplyNodeSerializer supported_workflow_type_list = [WorkflowType.APPLICATION, WorkflowType.KNOWLEDGE, WorkflowType.TOOL] @@ -38,27 +91,49 @@ def execute(self): reply_type = node_params.get("reply_type") fields = node_params.get("fields") content = node_params.get("content") + content_type = node_params.get("content_type") or "TEXT" is_result = node_params.get("is_result", False) + self.data["content_type"] = content_type if reply_type == "referencing": - result = self._get_reference_content(fields) + value = self._get_reference_value(fields) else: - result = self._generate_reply_content(content) + value = self._generate_reply_content(content) - self.write_context("answer", result) + node_info = NodeInfo(self.get_node_id(), self.get_node_name(), Status.SUCCESS) + position = Position(self.get_node_id()) + if content_type == "TOOL": + tool_list = _to_tool_list(value) + self.write_context("answer", tool_list) + if is_result: + for tool in tool_list: + self.write( + ToolContent( + tool["id"], + tool["name"], + tool["arguments"], + tool["content"], + Status.SUCCESS, + node_info, + position, + ) + ) + return + result = _to_text(value) + self.write_context("answer", result) if is_result: - node_info = NodeInfo(self.get_node_id(), self.get_node_name(), Status.SUCCESS) - self.write(TextContent(str(chunk_id), result, Status.SUCCESS, node_info, Position(self.get_node_id()))) + content_class = HtmlContent if content_type == "HTML" else TextContent + self.write(content_class(str(chunk_id), result, Status.SUCCESS, node_info, position)) def _generate_reply_content(self, prompt): if prompt is None: return "" return self.workflow_manage.generate_prompt(prompt) - def _get_reference_content(self, fields: List[str]): + def _get_reference_value(self, fields: List[str]): if fields and len(fields) >= 2: - return str(self.workflow_manage.get_reference_field(fields[0], fields[1:])) + return self.workflow_manage.get_reference_field(fields[0], fields[1:]) return "" def get_details(self, index: int = 0, position: dict = None, old_details: dict = None, **kwargs): @@ -66,6 +141,7 @@ def get_details(self, index: int = 0, position: dict = None, old_details: dict = details.update( { "answer": self.get_context("answer"), + "content_type": self.data.get("content_type", "TEXT"), } ) return details diff --git a/apps/chat/serializers/chat.py b/apps/chat/serializers/chat.py index 4a3e851e477..3db779a4a4c 100644 --- a/apps/chat/serializers/chat.py +++ b/apps/chat/serializers/chat.py @@ -16,6 +16,7 @@ import uuid_utils import uuid_utils.compat as uuid from django.db.models import QuerySet +from django.db.models.expressions import RawSQL from django.http import StreamingHttpResponse from django.utils import timezone from django.utils.translation import gettext_lazy as _ @@ -42,6 +43,7 @@ from application.workflow.message.struct.failure_content import FailureContent from application.workflow.message_queue import get_message_queue from application.workflow.nodes import get_start_node +from application.workflow.nodes.start_node.start_node import get_default_global_variable from application.workflow.workflow_manage import WorkflowManage, CallBack from application.workflow.workflow_run_registry import WorkflowRunRegistry from knowledge.services.retrieval_access import identity_from_server @@ -64,6 +66,12 @@ def get_work_flow(application): return build_workflow(application) +def init_chat_variable(workflow_manage, workflow, chat_variable): + base_node = workflow.get_node("base-node") + chat_input_field_list = base_node.properties.get("chat_input_field_list", []) if base_node else [] + workflow_manage.context["chat"] = {**get_default_global_variable(chat_input_field_list), **chat_variable} + + class ChatMessageSerializers(serializers.Serializer): """新流程的对话入参（去掉旧工作流调试字段 node_id/runtime_node_id/node_data/child_node）。""" @@ -223,6 +231,24 @@ def _usage_from_details(details): completion_tokens = sum((d.get("answer_tokens") or 0) for d in details if isinstance(d, dict)) return {"prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens} + @staticmethod + def update_chat_variable(chat_id, initial_chat_variable, chat_variable): + """ + 会话变量落库：只把本次运行改过的 key 合并进 Chat.meta（PG jsonb ||）， + 不 SELECT、不整行 save，同一会话并发时也不会互相覆盖未改动的 key。 + """ + if not isinstance(chat_variable, dict): + return + changed = {k: v for k, v in chat_variable.items() if initial_chat_variable.get(k) != v} + if not changed: + return + try: + QuerySet(Chat).filter(id=chat_id).update( + meta=RawSQL("COALESCE(meta, '{}'::jsonb) || %s::jsonb", [json.dumps(changed, ensure_ascii=False)]) + ) + except Exception as e: + maxkb_logger.error(f"update chat variable error [{chat_id}]: {e}") + @staticmethod def update_chat_record(chat_user_id, chat_record_id, workflow_context, messages, details, run_time): usage = ChatSerializers._usage_from_details(details) @@ -272,7 +298,9 @@ def chat_work_flow(self, application, instance: dict, base_to_response): workflow = new_instance(work_flow, WorkflowType.APPLICATION) chat_record_id_str = str(uuid.uuid7()) if chat_record_id is None else str(chat_record_id) - self.ensure_chat_row(message, chat_user) + chat = self.ensure_chat_row(message, chat_user) + # 会话变量：复用归属校验时已查出的 Chat 行，不额外查询 + chat_variable = dict(chat.meta) if isinstance(chat.meta, dict) else {} if chat_record_id is None: ChatRecord(id=chat_record_id_str, **self.get_defaults_record(message_dict)).save(force_insert=True) @@ -302,6 +330,7 @@ def chat_work_flow(self, application, instance: dict, base_to_response): "video_list": video_list or [], "other_list": other_list or [], "default_model_setting": default_model_setting, + "chat_variable": chat_variable, } result_queue = queue.Queue() @@ -334,6 +363,7 @@ def on_complete(wf_manage, error): details = wf_manage.get_details(position=position, old_details=old_details) run_time = time.time() - wf_manage.start_time self.update_chat_record(chat_user_id, chat_record_id_str, wf_manage.context, messages, details, run_time) + self.update_chat_variable(chat_id, chat_variable, wf_manage.context.get("chat")) ChatCountSerializer(data={"chat_id": chat_id}).update_chat() # 表单续跑时 message_dict.content 为空;保留原记录里的用户问题,避免 WORKFLOW 历史丢问题 question = chat_record.question if (chat_record and chat_record.question) else message_dict @@ -374,6 +404,7 @@ def get_start_node_fn(wf, wm): workflow, parameters, WorkflowType.APPLICATION, call_back, get_start_node_fn ) + init_chat_variable(work_flow_manage, workflow, chat_variable) work_flow_manage.start_node.workflow_manage = work_flow_manage WorkflowRunRegistry.register(chat_record_id_str, str(chat_id), work_flow_manage) diff --git a/apps/common/handle/impl/response/openai_to_response.py b/apps/common/handle/impl/response/openai_to_response.py index 98023aef364..35cdea1179a 100644 --- a/apps/common/handle/impl/response/openai_to_response.py +++ b/apps/common/handle/impl/response/openai_to_response.py @@ -44,7 +44,7 @@ def _to_tool_call_delta(self, block: dict) -> dict: def to_stream(self, chat_id, chat_record_id, block: dict): block_type = block.get("type") delta_kwargs = {"chat_id": chat_id} - if block_type == "TEXT": + if block_type in ("TEXT", "HTML"): delta_kwargs["content"] = block.get("content", "") elif block_type == "REASONING": delta_kwargs["reasoning_content"] = block.get("content", "") @@ -82,7 +82,7 @@ def to_stream_end(self, chat_id, chat_record_id, usage: dict = None): def to_block(self, chat_id, chat_record_id, contents: list, usage: dict = None, _status=status.HTTP_200_OK): usage = usage or {} - answer = "".join(c.get("content", "") for c in (contents or []) if c.get("type") == "TEXT") + answer = "".join(c.get("content", "") for c in (contents or []) if c.get("type") in ("TEXT", "HTML")) tool_calls = [] for c in contents or []: if c.get("type") != "TOOL": diff --git a/apps/common/handle/impl/response/system_to_response.py b/apps/common/handle/impl/response/system_to_response.py index daabc6cc1b5..47a338878a9 100644 --- a/apps/common/handle/impl/response/system_to_response.py +++ b/apps/common/handle/impl/response/system_to_response.py @@ -34,7 +34,7 @@ def to_stream_end(self, chat_id, chat_record_id, usage: dict = None): def to_block(self, chat_id, chat_record_id, contents: list, usage: dict = None, _status=status.HTTP_200_OK): usage = usage or {} - answer = "".join(c.get("content", "") for c in (contents or []) if c.get("type") == "TEXT") + answer = "".join(c.get("content", "") for c in (contents or []) if c.get("type") in ("TEXT", "HTML")) return result.success( { "chat_id": str(chat_id), diff --git a/apps/common/utils/messages_util.py b/apps/common/utils/messages_util.py index b29645df1d3..3827dd86cdd 100644 --- a/apps/common/utils/messages_util.py +++ b/apps/common/utils/messages_util.py @@ -25,7 +25,7 @@ def to_human_message_list(question): def to_ai_message_list(messages): """ - 将 messages 中的 TEXT / TOOL 内容块转换为 LangChain 消息列表。 + 将 messages 中的 TEXT / HTML / TOOL 内容块转换为 LangChain 消息列表。 REASONING/FORM/FAILURE 不进历史；按顺序保留交错。 注意：type 用字面量，避免对 application.workflow.ContentType 产生反向依赖。 """ @@ -34,7 +34,7 @@ def to_ai_message_list(messages): if not isinstance(m, dict): continue m_type = m.get("type") - if m_type == "TEXT": + if m_type in ("TEXT", "HTML"): if m.get("content"): ai_message_list.append(AIMessage(content=m.get("content"))) elif m_type == "TOOL": diff --git a/apps/locales/en_US/LC_MESSAGES/django.po b/apps/locales/en_US/LC_MESSAGES/django.po index 448be2f7a8e..29f48236f04 100644 --- a/apps/locales/en_US/LC_MESSAGES/django.po +++ b/apps/locales/en_US/LC_MESSAGES/django.po @@ -541,6 +541,12 @@ msgstr "" msgid "Response Type" msgstr "" +msgid "Content Type" +msgstr "" + +msgid "Tool reply content must be a JSON object or array" +msgstr "" + #: apps/application/flow/step_node/direct_reply_node/i_reply_node.py:21 #: apps/application/flow/step_node/variable_assign_node/i_variable_assign_node.py:13 msgid "Reference Field" diff --git a/apps/locales/zh_CN/LC_MESSAGES/django.po b/apps/locales/zh_CN/LC_MESSAGES/django.po index dcdc196b9f5..0f9a148abbf 100644 --- a/apps/locales/zh_CN/LC_MESSAGES/django.po +++ b/apps/locales/zh_CN/LC_MESSAGES/django.po @@ -553,6 +553,12 @@ msgstr "条件 或|与" msgid "Response Type" msgstr "响应类型" +msgid "Content Type" +msgstr "内容类型" + +msgid "Tool reply content must be a JSON object or array" +msgstr "工具回复内容必须是 JSON 对象或数组" + #: apps/application/flow/step_node/direct_reply_node/i_reply_node.py:21 #: apps/application/flow/step_node/variable_assign_node/i_variable_assign_node.py:13 msgid "Reference Field" diff --git a/apps/locales/zh_Hant/LC_MESSAGES/django.po b/apps/locales/zh_Hant/LC_MESSAGES/django.po index ef537cc0f81..0e3daee6ee4 100644 --- a/apps/locales/zh_Hant/LC_MESSAGES/django.po +++ b/apps/locales/zh_Hant/LC_MESSAGES/django.po @@ -553,6 +553,12 @@ msgstr "條件 或|與" msgid "Response Type" msgstr "響應類型" +msgid "Content Type" +msgstr "內容類型" + +msgid "Tool reply content must be a JSON object or array" +msgstr "工具回覆內容必須是 JSON 物件或陣列" + #: apps/application/flow/step_node/direct_reply_node/i_reply_node.py:21 #: apps/application/flow/step_node/variable_assign_node/i_variable_assign_node.py:13 msgid "Reference Field" diff --git a/apps/tools/migrations/0009_toolworkflowversion_publish_desc.py b/apps/tools/migrations/0009_toolworkflowversion_publish_desc.py new file mode 100644 index 00000000000..7a22caaa98e --- /dev/null +++ b/apps/tools/migrations/0009_toolworkflowversion_publish_desc.py @@ -0,0 +1,15 @@ +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("tools", "0008_toolworkflow_default_model_setting_and_more"), + ] + + operations = [ + migrations.AddField( + model_name="toolworkflowversion", + name="publish_desc", + field=models.CharField(default="", max_length=1024, verbose_name="更新说明"), + ), + ] diff --git a/apps/tools/models/tool_workflow.py b/apps/tools/models/tool_workflow.py index 8ca45c8c549..39d6fc332cd 100644 --- a/apps/tools/models/tool_workflow.py +++ b/apps/tools/models/tool_workflow.py @@ -43,6 +43,7 @@ class ToolWorkflowVersion(AppModelMixin): tool = models.ForeignKey(Tool, on_delete=models.CASCADE, verbose_name="工具", db_constraint=False) workspace_id = models.CharField(max_length=64, verbose_name="工作空间id", default="default", db_index=True) name = models.CharField(verbose_name="版本名称", max_length=128, default="") + publish_desc = models.CharField(verbose_name="更新说明", max_length=1024, default="") work_flow = models.JSONField(verbose_name="工作流数据", default=dict) publish_user_id = models.UUIDField(verbose_name="发布者id", max_length=128, default=None, null=True) publish_user_name = models.CharField(verbose_name="发布者名称", max_length=128, default="") diff --git a/apps/tools/serializers/tool_import.py b/apps/tools/serializers/tool_import.py new file mode 100644 index 00000000000..6096b28b9de --- /dev/null +++ b/apps/tools/serializers/tool_import.py @@ -0,0 +1,140 @@ +# -*- coding: utf-8 -*- +""" +导入(应用 / 知识库工作流 / 工具工作流)时,包内工具的复用、新建与引用权限校验 +""" + +from typing import Dict, List, Tuple + +import uuid_utils.compat as uuid +from common.exception.app_exception import AppApiException +from common.utils.common import generate_uuid +from django.db.models import QuerySet +from django.utils.translation import gettext_lazy as _ +from rest_framework.utils.formatting import lazy_format +from system_manage.models import AuthTargetType, WorkspaceUserResourcePermission +from tools.models import Tool +from users.serializers.user import is_workspace_manage + + +def _walk_workflow_nodes(work_flow, collector): + """遍历工作流节点(含 loop-node 嵌套),对每个节点的 node_data 调用 collector。""" + if not work_flow: + return + for node in work_flow.get("nodes", []) or []: + node_data = (node.get("properties") or {}).get("node_data") or {} + collector(node_data) + if node.get("type") == "loop-node": + _walk_workflow_nodes(node_data.get("loop_body"), collector) + + +def get_bound_tool_ids(instance: Dict) -> List[str]: + """ + 收集应用配置(含工作流节点)中引用的所有工具id,用于绑定前的权限校验 + """ + tool_ids = set() + for key in ("tool_ids", "skill_tool_ids", "mcp_tool_ids"): + for tool_id in instance.get(key) or []: + tool_ids.add(str(tool_id)) + if instance.get("mcp_tool_id"): + tool_ids.add(str(instance.get("mcp_tool_id"))) + + def collect(node_data): + for key in ("tool_lib_id", "mcp_tool_id"): + if node_data.get(key): + tool_ids.add(str(node_data.get(key))) + for key in ("mcp_tool_ids", "tool_ids", "skill_tool_ids"): + for tool_id in node_data.get(key) or []: + tool_ids.add(str(tool_id)) + + _walk_workflow_nodes(instance.get("work_flow"), collect) + return list(tool_ids) + + +def get_authorized_tool_ids(user_id: str, workspace_id: str, tool_ids: List[str]) -> List[str]: + """ + 返回 tool_ids 中当前用户被授权绑定/使用的工具id。 + 工作空间管理员默认拥有全部工具权限;其他用户必须在 workspace_user_resource_permission + 中存在针对该工具的显式授权记录(默认拒绝)。 + """ + if not tool_ids: + return [] + tool_ids = list({str(t) for t in tool_ids}) + if is_workspace_manage(user_id, workspace_id): + return tool_ids + granted_tool_ids = { + str(permission.target) + for permission in QuerySet(WorkspaceUserResourcePermission).filter( + workspace_id=workspace_id, + user_id=user_id, + auth_target_type=AuthTargetType.TOOL.value, + target__in=tool_ids, + ) + if "VIEW" in permission.permission_list or "ROLE" in permission.permission_list + } + return [tool_id for tool_id in tool_ids if tool_id in granted_tool_ids] + + +def resolve_import_tools(tool_list: List[Dict], workspace_id: str, user_id: str, force_new=False) -> Tuple[Dict, List]: + """ + 处理导入包内的工具:工作空间已存在且导入者有权限的直接复用,否则以新id新建 + + @param tool_list: 导入包内的工具列表 + @param workspace_id: 工作空间id + @param user_id: 导入者id + @param force_new: 是否全部新建,不复用已存在的工具 + @return: (update_tool_map, new_tool_list) + update_tool_map: 包内旧id -> 实际绑定id(复用的已存在id 或 新建id),用于改写工作流引用 + new_tool_list: 需要新建的工具,id 已替换为新id + """ + tool_list = [tool for tool in {str(t.get("id")): t for t in tool_list if t.get("id")}.values()] + if len(tool_list) == 0: + return {}, [] + + # 旧id -> 候选的已存在id(优先旧id本身,其次历史导入生成的确定性id) + candidate_map = {} + if not force_new: + candidate_id_map = { + str(tool.get("id")): [str(tool.get("id")), str(generate_uuid(str(tool.get("id")) + str(workspace_id)))] + for tool in tool_list + } + exists_tool_id_set = { + str(tool.id) + for tool in QuerySet(Tool).filter( + id__in=[i for ids in candidate_id_map.values() for i in ids], workspace_id=workspace_id + ) + } + authorized_tool_id_set = set(get_authorized_tool_ids(user_id, workspace_id, list(exists_tool_id_set))) + for old_id, candidate_ids in candidate_id_map.items(): + existing_id = next((i for i in candidate_ids if i in exists_tool_id_set), None) + if existing_id is not None and existing_id in authorized_tool_id_set: + candidate_map[old_id] = existing_id + + update_tool_map = {} + new_tool_list = [] + for tool in tool_list: + old_id = str(tool.get("id")) + if old_id in candidate_map: + update_tool_map[old_id] = candidate_map[old_id] + else: + new_id = str(uuid.uuid7()) + update_tool_map[old_id] = new_id + new_tool_list.append({**tool, "id": new_id}) + return update_tool_map, new_tool_list + + +def validate_unpackaged_tool_refs(instance_list: List[Dict], update_tool_map: Dict, user_id: str, workspace_id: str): + """ + 校验不在导入包内的工具引用:无法随包新建,只能绑定已存在的工具,需导入者有使用权限 + + @param instance_list: 含工具引用的配置(应用 / 工作流 / 新建的工作流工具),须为改写引用前的原始数据 + @param update_tool_map: resolve_import_tools 返回的 id 映射 + """ + tool_ids = {tool_id for instance in instance_list for tool_id in get_bound_tool_ids(instance)} + unpackaged_tool_ids = [tool_id for tool_id in tool_ids if tool_id not in update_tool_map] + if not unpackaged_tool_ids: + return + authorized_tool_ids = set(get_authorized_tool_ids(user_id, workspace_id, unpackaged_tool_ids)) + unauthorized_tool_ids = [tool_id for tool_id in unpackaged_tool_ids if tool_id not in authorized_tool_ids] + if unauthorized_tool_ids: + message = lazy_format(_("No permission to use tool(s): {tool_ids}"), tool_ids=", ".join(unauthorized_tool_ids)) + raise AppApiException(403, str(message)) diff --git a/apps/tools/serializers/tool_version.py b/apps/tools/serializers/tool_version.py index e986e6b177d..d68b9fadc33 100644 --- a/apps/tools/serializers/tool_version.py +++ b/apps/tools/serializers/tool_version.py @@ -1,11 +1,12 @@ # coding=utf-8 """ - @project: MaxKB - @Author：虎虎 - @file： KnowledgeVersionSerializer.py - @date：2025/11/28 18:00 - @desc: +@project: MaxKB +@Author：虎虎 +@file： KnowledgeVersionSerializer.py +@date：2025/11/28 18:00 +@desc: """ + from typing import Dict from django.db.models import QuerySet @@ -18,22 +19,34 @@ class ToolWorkflowVersionEditSerializer(serializers.Serializer): - name = serializers.CharField(required=False, max_length=128, allow_null=True, allow_blank=True, - label=_("Version Name")) + name = serializers.CharField( + required=False, max_length=128, allow_null=True, allow_blank=True, label=_("Version Name") + ) + publish_desc = serializers.CharField( + required=False, max_length=1024, allow_null=True, allow_blank=True, label=_("Publish Desc") + ) class ToolVersionModelSerializer(serializers.ModelSerializer): class Meta: model = ToolWorkflowVersion - fields = ['id', 'name', 'workspace_id', 'tool_id', 'work_flow', 'publish_user_id', 'publish_user_name', - 'create_time', - 'update_time'] + fields = [ + "id", + "name", + "publish_desc", + "workspace_id", + "tool_id", + "work_flow", + "publish_user_id", + "publish_user_name", + "create_time", + "update_time", + ] class ToolWorkflowVersionQuerySerializer(serializers.Serializer): tool_id = serializers.UUIDField(required=True, label=_("Tool ID")) - name = serializers.CharField(required=False, allow_null=True, allow_blank=True, - label=_("summary")) + name = serializers.CharField(required=False, allow_null=True, allow_blank=True, label=_("summary")) class ToolWorkflowVersionSerializer(serializers.Serializer): @@ -43,11 +56,11 @@ class Query(serializers.Serializer): workspace_id = serializers.CharField(required=False, allow_null=True, allow_blank=True, label=_("Workspace ID")) def get_query_set(self, query): - query_set = QuerySet(ToolWorkflowVersion).filter(tool_id=query.get('tool_id')) - if 'name' in query and query.get('name') is not None: - query_set = query_set.filter(name__contains=query.get('name')) - if 'workspace_id' in self.data and self.data.get('workspace_id') is not None: - query_set = query_set.filter(workspace_id=self.data.get('workspace_id')) + query_set = QuerySet(ToolWorkflowVersion).filter(tool_id=query.get("tool_id")) + if "name" in query and query.get("name") is not None: + query_set = query_set.filter(name__contains=query.get("name")) + if "workspace_id" in self.data and self.data.get("workspace_id") is not None: + query_set = query_set.filter(workspace_id=self.data.get("workspace_id")) return query_set.order_by("-create_time") def list(self, query, with_valid=True): @@ -60,48 +73,57 @@ def list(self, query, with_valid=True): def page(self, query, current_page, page_size, with_valid=True): if with_valid: self.is_valid(raise_exception=True) - return page_search(current_page, page_size, - self.get_query_set(query), - post_records_handler=lambda v: ToolVersionModelSerializer(v).data) + return page_search( + current_page, + page_size, + self.get_query_set(query), + post_records_handler=lambda v: ToolVersionModelSerializer(v).data, + ) class Operate(serializers.Serializer): workspace_id = serializers.CharField(required=False, allow_null=True, allow_blank=True, label=_("Workspace ID")) tool_id = serializers.UUIDField(required=True, label=_("Tool ID")) - tool_version_id = serializers.UUIDField(required=True, - label=_("Tool version ID")) + tool_version_id = serializers.UUIDField(required=True, label=_("Tool version ID")) def is_valid(self, *, raise_exception=False): super().is_valid(raise_exception=True) - workspace_id = self.data.get('workspace_id') - query_set = QuerySet(Tool).filter(id=self.data.get('tool_id')) + workspace_id = self.data.get("workspace_id") + query_set = QuerySet(Tool).filter(id=self.data.get("tool_id")) if workspace_id: query_set = query_set.filter(workspace_id=workspace_id) if not query_set.exists(): - raise AppApiException(500, _('Tool id does not exist')) + raise AppApiException(500, _("Tool id does not exist")) def one(self, with_valid=True): if with_valid: self.is_valid(raise_exception=True) - tool_version = QuerySet(ToolWorkflowVersion).filter(tool_id=self.data.get('tool_id'), - id=self.data.get( - 'tool_version_id')).first() + tool_version = ( + QuerySet(ToolWorkflowVersion) + .filter(tool_id=self.data.get("tool_id"), id=self.data.get("tool_version_id")) + .first() + ) if tool_version is not None: return ToolVersionModelSerializer(tool_version).data else: - raise AppApiException(500, _('Workflow version does not exist')) + raise AppApiException(500, _("Workflow version does not exist")) def edit(self, instance: Dict, with_valid=True): if with_valid: self.is_valid(raise_exception=True) ToolWorkflowVersionEditSerializer(data=instance).is_valid(raise_exception=True) - tool_version = QuerySet(ToolWorkflowVersion).filter(tool_id=self.data.get('tool_id'), - id=self.data.get( - 'tool_version_id')).first() + tool_version = ( + QuerySet(ToolWorkflowVersion) + .filter(tool_id=self.data.get("tool_id"), id=self.data.get("tool_version_id")) + .first() + ) if tool_version is not None: - name = instance.get('name', None) + name = instance.get("name", None) if name is not None and len(name) > 0: tool_version.name = name + publish_desc = instance.get("publish_desc", None) + if publish_desc is not None: + tool_version.publish_desc = publish_desc tool_version.save() return ToolVersionModelSerializer(tool_version).data else: - raise AppApiException(500, _('Workflow version does not exist')) + raise AppApiException(500, _("Workflow version does not exist")) diff --git a/apps/tools/serializers/tool_workflow.py b/apps/tools/serializers/tool_workflow.py index 3212b3315a8..11cae7d5d0e 100644 --- a/apps/tools/serializers/tool_workflow.py +++ b/apps/tools/serializers/tool_workflow.py @@ -150,6 +150,15 @@ def get_tool_list(self): return self.tool_list or [] +class ToolWorkflowPublishSerializer(serializers.Serializer): + publish_name = serializers.CharField( + required=False, max_length=128, allow_null=True, allow_blank=True, label=_("Publish Name") + ) + publish_desc = serializers.CharField( + required=False, max_length=1024, allow_null=True, allow_blank=True, label=_("Publish Desc") + ) + + class ToolWorkflowSerializer(serializers.Serializer): class Operate(serializers.Serializer): user_id = serializers.UUIDField(required=True, label=_("user id")) @@ -343,10 +352,13 @@ def compute_tool_state(details, error): has_fail = any((d or {}).get("status") == Status.FAIL.value for d in (details or [])) return State.FAILURE if has_fail else State.SUCCESS - def publish(self, with_valid=True): + def publish(self, instance=None, with_valid=True): if with_valid: self.is_valid() + ToolWorkflowPublishSerializer(data=instance or {}).is_valid(raise_exception=True) user_id = self.data.get("user_id") + publish_name = str((instance or {}).get("publish_name") or "").strip() + publish_desc = (instance or {}).get("publish_desc") or "" user = QuerySet(User).filter(id=user_id).first() tool_workflow = QuerySet(ToolWorkflow).filter(tool_id=self.data.get("tool_id")).first() @@ -354,7 +366,8 @@ def publish(self, with_valid=True): work_flow_version = ToolWorkflowVersion( work_flow=tool_workflow.work_flow, tool_id=self.data.get("tool_id"), - name=timezone.localtime(timezone.now()).strftime("%Y-%m-%d %H:%M:%S"), + name=publish_name or timezone.localtime(timezone.now()).strftime("%Y-%m-%d %H:%M:%S"), + publish_desc=publish_desc, publish_user_id=user_id, publish_user_name=user.username, workspace_id=workspace_id, diff --git a/apps/tools/views/tool_workflow.py b/apps/tools/views/tool_workflow.py index d0e38441b4b..6eda5e0a6bc 100644 --- a/apps/tools/views/tool_workflow.py +++ b/apps/tools/views/tool_workflow.py @@ -17,7 +17,12 @@ from tools.api.tool import GetInternalToolAPI from tools.api.tool_workflow import ToolWorkflowApi -from tools.serializers.tool_workflow import StoreToolWorkflow, ToolWorkflowMcpSerializer, ToolWorkflowSerializer +from tools.serializers.tool_workflow import ( + StoreToolWorkflow, + ToolWorkflowMcpSerializer, + ToolWorkflowPublishSerializer, + ToolWorkflowSerializer, +) from tools.views import get_tool_operation_object @@ -33,7 +38,7 @@ class Publish(APIView): summary=_("Publishing an tool"), operation_id=_("Publishing an tool"), # type: ignore parameters=ToolWorkflowApi.get_parameters(), - request=None, + request=ToolWorkflowPublishSerializer, responses=DefaultResultSerializer, tags=[_("Tool")], # type: ignore ) @@ -60,7 +65,7 @@ def put(self, request: Request, workspace_id: str, tool_id: str): "user_id": request.user.id, "workspace_id": workspace_id, } - ).publish() + ).publish(request.data) ) class Operate(APIView): diff --git a/ui/src/conversation-panel/components/content-list/index.vue b/ui/src/conversation-panel/components/content-list/index.vue index ee3739bea22..23570ddcf20 100644 --- a/ui/src/conversation-panel/components/content-list/index.vue +++ b/ui/src/conversation-panel/components/content-list/index.vue @@ -1,6 +1,6 @@