diff --git a/apps/application/workflow/nodes/application_node/__init__.py b/apps/application/workflow/nodes/application_node/__init__.py new file mode 100644 index 00000000000..fb099b7fed4 --- /dev/null +++ b/apps/application/workflow/nodes/application_node/__init__.py @@ -0,0 +1,4 @@ +# coding=utf-8 +from .application_node import ApplicationNode + +__all__ = ["ApplicationNode"] diff --git a/apps/application/workflow/nodes/application_node/application_node.py b/apps/application/workflow/nodes/application_node/application_node.py new file mode 100644 index 00000000000..85a0b6c7454 --- /dev/null +++ b/apps/application/workflow/nodes/application_node/application_node.py @@ -0,0 +1,283 @@ +# coding=utf-8 +""" +@project: MaxKB +@file: application_node.py +@date:2026/9/3 +@desc: 智能体节点 +""" + +import uuid_utils.compat as uuid +from django.db.models import QuerySet +from django.utils.translation import gettext_lazy as _ +from rest_framework import serializers + +from application.models import Application, ApplicationVersion, Chat, ChatRecord, ChatSourceChoices +from application.workflow.common import WorkflowType, new_instance +from application.workflow.content_type import ContentType +from application.workflow.i_node import INode, Signal +from application.workflow.message.struct.content import Position +from application.workflow.status import Status +from application.workflow.workflow_manage import WorkflowManage, CallBack +from chat.serializers.chat_history import ChatHistory + + +def string_to_uuid(input_str): + return str(uuid.uuid5(uuid.NAMESPACE_DNS, input_str)) + + +class ApplicationNodeSerializer(serializers.Serializer): + application_id = serializers.CharField(required=True, label=_("Application ID")) + question_reference_address = serializers.ListField(required=True, label=_("User Questions")) + api_input_field_list = serializers.ListField(required=False, label=_("API Input Fields")) + user_input_field_list = serializers.ListField(required=False, label=_("User Input Fields")) + image_list = serializers.ListField(required=False, label=_("picture")) + document_list = serializers.ListField(required=False, label=_("document")) + audio_list = serializers.ListField(required=False, label=_("Audio")) + video_list = serializers.ListField(required=False, label=_("Video")) + node_data = serializers.DictField(required=False, allow_null=True, label=_("Form Data")) + + +class ApplicationNode(INode): + serializer_class = ApplicationNodeSerializer + supported_workflow_type_list = [WorkflowType.APPLICATION] + type = "application-node" + + def _run(self): + # 完成时机由子应用 on_complete 回调驱动,这里不自动 complete + self.execute() + + def execute(self): + node_params = self.get_parameters() + workflow_params = self.get_workflow_parameters() + + application_id = node_params.get("application_id") + chat_id = workflow_params.get("chat_id") + chat_user_id = workflow_params.get("chat_user_id") + chat_user_type = workflow_params.get("chat_user_type") + ip_address = workflow_params.get("ip_address") or "-" + source = workflow_params.get("source") or {"type": ChatSourceChoices.ONLINE.value} + debug = workflow_params.get("debug", False) + + # 父工作流 position 指向本节点 → 子工作流表单提交,需恢复续跑 + position = workflow_params.get("position") or {} + is_submit = position.get("id") == self.get_node_id() + sub_chat_record_id = self.get_context("sub_chat_record_id") + # 表单暂停点:前端回传的 position 是一个嵌套链(id=本节点, children=子工作流暂停点)。 + # 恢复时用 position.children 逐级下传,才能把深层锚点(子子工作流表单)完整带下去。 + sub_position = position.get("children") or self.get_context("sub_position") + + # 自引用守卫 + if application_id == workflow_params.get("application_id"): + raise Exception(_("The sub application cannot use the current node")) + + # 解析用户问题 + question_address = node_params.get("question_reference_address") or [] + if question_address: + question = self.workflow_manage.get_reference_field(question_address[0], question_address[1:]) + else: + question = "" + question = str(question or "") + self.write_context("question", question) + + # 解析 api 输入 / 用户输入 → form_data + form_data = {} + for api_input_field in node_params.get("api_input_field_list", []): + value = api_input_field.get("value", [""])[0] if api_input_field.get("value") else "" + form_data[api_input_field["variable"]] = ( + self.workflow_manage.get_reference_field(value, api_input_field["value"][1:]) if value != "" else "" + ) + for user_input_field in node_params.get("user_input_field_list", []): + value = user_input_field.get("value", [""])[0] if user_input_field.get("value") else "" + form_data[user_input_field["field"]] = ( + self.workflow_manage.get_reference_field(value, user_input_field["value"][1:]) if value != "" else "" + ) + + # 解析文件列表(校验 file_id) + app_document_list = self._resolve_file_list(node_params.get("document_list", []), "document") + app_image_list = self._resolve_file_list(node_params.get("image_list", []), "image") + app_audio_list = self._resolve_file_list(node_params.get("audio_list", []), "audio") + app_video_list = self._resolve_file_list(node_params.get("video_list", []), "video") + + # 派生子应用聊天 id + current_chat_id = string_to_uuid(chat_id + application_id) + Chat.objects.get_or_create( + id=current_chat_id, + defaults={ + "application_id": application_id, + "abstract": question[0:1024], + "chat_user_id": chat_user_id, + "chat_user_type": chat_user_type, + "ip_address": ip_address, + "source": source, + "asker": self._get_chat_asker(workflow_params), + }, + ) + + # 解析子应用工作流(debug 取本体,否则取最新发布版本),与 chat_work_flow 的 get_application 一致 + if debug: + sub_application = QuerySet(Application).filter(id=application_id).first() + else: + sub_application = ( + QuerySet(ApplicationVersion).filter(application_id=application_id).order_by("-create_time")[0:1].first() + ) + if sub_application is None: + raise Exception(_("The application has not been published. Please use it after publishing.")) + from chat.serializers.chat import get_work_flow + + sub_workflow = new_instance(get_work_flow(sub_application), WorkflowType.APPLICATION) + + # 首次运行:生成子应用记录 id 并建 ChatRecord;恢复时沿用已持久化的 id + if not is_submit: + sub_chat_record_id = str(uuid.uuid7()) + self.write_context("sub_chat_record_id", sub_chat_record_id) + self.write_context("sub_position", sub_position) + QuerySet(ChatRecord).create( + id=sub_chat_record_id, + chat_id=current_chat_id, + problem_text=question[0:1024], + answer_text="", + details={}, + message_tokens=0, + answer_tokens=0, + answer_text_list=[[]], + index=0, + ip_address=ip_address or "", + source=source, + workflow_context={}, + question={"content": question}, + messages=[], + ) + sub_position = None + + # 组装子应用参数(复制父工作流参数并覆盖子应用相关字段) + sub_parameters = dict(workflow_params) + sub_parameters.update( + { + "chat_id": current_chat_id, + "chat_record_id": sub_chat_record_id, + "application_id": application_id, + "question": question, + "stream": True, + "form_data": workflow_params.get("form_data") if is_submit else form_data, + "position": sub_position if is_submit else None, + "chunk_id": workflow_params.get("chunk_id"), + "history_chat_record": ChatHistory(current_chat_id).load(exclude_record_id=sub_chat_record_id), + "image_list": app_image_list, + "document_list": app_document_list, + "audio_list": app_audio_list, + "video_list": app_video_list, + } + ) + + # 内联回调:转发子应用输出、嵌套 position、传播表单暂停信号 + self._answer = "" + self._reasoning_content = "" + + def on_next(wf_manage, content): + if content.type == ContentType.FORM: + # 已提交表单的回显块不转发,避免前端出现重复的已填表单 + if content.is_submit: + return + # 记录子工作流表单节点位置(保留嵌套链),供父工作流恢复时透传回子工作流续跑 + self.write_context("sub_position", content.position.to_dict()) + content.position = Position(self.get_node_id(), None, content.position) + self.write(content) + return + content.position = Position(self.get_node_id(), None, content.position) + if content.type == ContentType.TEXT: + self._answer += content.content + elif content.type == ContentType.REASONING: + self._reasoning_content += content.content + self.write(content) + + def on_complete(wf_manage, error): + # 持久化子应用上下文,供后续 resume 的 from_context 读取 + QuerySet(ChatRecord).filter(id=sub_chat_record_id).update(workflow_context=wf_manage.context) + usage = self._usage_from_context(wf_manage.context) + self._write_final_context(self._answer, self._reasoning_content, usage) + if error: + self.complete(Status.FAIL, error=error) + return + # 子应用命中表单(Signal.FORM):向上传播中断,暂停父工作流,等用户提交后恢复 + if wf_manage.signal == Signal.FORM: + self.complete(Status.SUCCESS, signal=Signal.FORM) + return + self.complete(Status.SUCCESS) + + call_back = CallBack(on_next, on_complete) + + def get_start_node_fn(wf, wm): + from application.workflow.nodes import get_start_node + + return get_start_node(wf, wm, WorkflowType.APPLICATION, sub_position if is_submit else None) + + # 表单提交:从历史 context 恢复子应用;否则全新运行 + if is_submit: + sub_manage = WorkflowManage.from_context( + chat_record_id=sub_chat_record_id, + workflow=sub_workflow, + parameters=sub_parameters, + workflow_type=WorkflowType.APPLICATION, + call_back=call_back, + get_start_node=get_start_node_fn, + ) + if sub_manage is None: + sub_manage = WorkflowManage( + sub_workflow, sub_parameters, WorkflowType.APPLICATION, call_back, get_start_node_fn + ) + else: + sub_manage = WorkflowManage( + sub_workflow, sub_parameters, WorkflowType.APPLICATION, call_back, get_start_node_fn + ) + + sub_manage.start_node.workflow_manage = sub_manage + sub_manage.run() + + def _resolve_file_list(self, field_list, name): + if not field_list or len(field_list) == 0: + return [] + values = self.workflow_manage.get_reference_field(field_list[0], field_list[1:]) or [] + for item in values: + if "file_id" not in item: + raise ValueError( + _("Parameter value error: The uploaded {name} lacks file_id, and the {name} upload fails").format( + name=name + ) + ) + return list(values) + + def _get_chat_asker(self, workflow_params): + asker = (workflow_params.get("form_data") or {}).get("asker") + if asker: + return asker if isinstance(asker, dict) else {"username": asker} + return workflow_params.get("chat_user") + + @staticmethod + def _usage_from_context(workflow_context): + prompt_tokens = 0 + completion_tokens = 0 + for node_context in (workflow_context or {}).values(): + if isinstance(node_context, dict): + prompt_tokens += node_context.get("message_tokens", 0) or 0 + completion_tokens += node_context.get("answer_tokens", 0) or 0 + return {"prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens} + + def _write_final_context(self, answer, reasoning_content, usage): + self.write_context("answer", answer) + self.write_context("result", answer) + self.write_context("reasoning_content", reasoning_content) + self.write_context("message_tokens", usage.get("prompt_tokens", 0)) + self.write_context("answer_tokens", usage.get("completion_tokens", 0)) + + def get_details(self, index: int = 0, position: dict = None, old_details: dict = None, **kwargs): + details = super().get_details(index, position, old_details, **kwargs) + details.update( + { + "question": self.get_context("question"), + "answer": self.get_context("answer"), + "reasoning_content": self.get_context("reasoning_content"), + "message_tokens": self.get_context("message_tokens"), + "answer_tokens": self.get_context("answer_tokens"), + } + ) + return details