Skip to content

feat: Node supports knowledge base workflow #4394

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zhanweizhang7 merged 1 commit into from
Nov 27, 2025
Merged

feat: Node supports knowledge base workflow #4394

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion apps/application/flow/step_node/condition_node/i_condition_node.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,4 +38,4 @@ def get_node_params_serializer_class(self) -> Type[serializers.Serializer]:

type = 'condition-node'

support = [WorkflowMode.APPLICATION_LOOP]
support = [WorkflowMode.APPLICATION, WorkflowMode.APPLICATION_LOOP, WorkflowMode.KNOWLEDGE, WorkflowMode.KNOWLEDGE_LOOP]
13 changes: 9 additions & 4 deletions apps/application/flow/step_node/image_to_video_step_node/i_image_to_video_node.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,8 @@ class ImageToVideoNodeSerializer(serializers.Serializer):

class IImageToVideoNode(INode):
type = 'image-to-video-node'
support = [WorkflowMode.APPLICATION, WorkflowMode.APPLICATION_LOOP]

support = [WorkflowMode.APPLICATION, WorkflowMode.APPLICATION_LOOP, WorkflowMode.KNOWLEDGE,
WorkflowMode.KNOWLEDGE_LOOP]
def get_node_params_serializer_class(self) -> Type[serializers.Serializer]:
return ImageToVideoNodeSerializer

Expand All @@ -55,10 +55,15 @@ def _run(self):
self.node_params_serializer.data.get('last_frame_url')[1:])
node_params_data = {k: v for k, v in self.node_params_serializer.data.items()
if k not in ['first_frame_url', 'last_frame_url']}
return self.execute(first_frame_url=first_frame_url, last_frame_url=last_frame_url,
if [WorkflowMode.KNOWLEDGE, WorkflowMode.KNOWLEDGE_LOOP].__contains__(
self.workflow_manage.flow.workflow_mode):
return self.execute(first_frame_url=first_frame_url, last_frame_url=last_frame_url, **node_params_data, **self.flow_params_serializer.data,
**{'history_chat_record': [], 'stream': True, 'chat_id': None, 'chat_record_id': None})
else:
return self.execute(first_frame_url=first_frame_url, last_frame_url=last_frame_url,
**node_params_data, **self.flow_params_serializer.data)

def execute(self, model_id, prompt, negative_prompt, dialogue_number, dialogue_type, history_chat_record, chat_id,
def execute(self, model_id, prompt, negative_prompt, dialogue_number, dialogue_type, history_chat_record,
model_params_setting,
chat_record_id,
first_frame_url, last_frame_url,
Expand Down
52 changes: 39 additions & 13 deletions apps/application/flow/step_node/image_to_video_step_node/impl/base_image_to_video_node.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from django.db.models import QuerySet
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

from application.flow.common import WorkflowMode
from application.flow.i_step_node import NodeResult
from application.flow.step_node.image_to_video_step_node.i_image_to_video_node import IImageToVideoNode
from common.utils.common import bytes_to_uploaded_file
Expand All @@ -23,12 +24,11 @@ def save_context(self, details, workflow_manage):
if self.node_params.get('is_result', False):
self.answer_text = details.get('answer')

def execute(self, model_id, prompt, negative_prompt, dialogue_number, dialogue_type, history_chat_record, chat_id,
def execute(self, model_id, prompt, negative_prompt, dialogue_number, dialogue_type, history_chat_record,
model_params_setting,
chat_record_id,
first_frame_url, last_frame_url=None,
**kwargs) -> NodeResult:
application = self.workflow_manage.work_flow_post_handler.chat_info.application
workspace_id = self.workflow_manage.get_body().get('workspace_id')
ttv_model = get_model_instance_by_model_workspace_id(model_id, workspace_id,
**model_params_setting)
Expand All @@ -54,17 +54,7 @@ def execute(self, model_id, prompt, negative_prompt, dialogue_number, dialogue_t
if isinstance(video_urls, str) and video_urls.startswith('http'):
video_urls = requests.get(video_urls).content
file = bytes_to_uploaded_file(video_urls, file_name)
meta = {
'debug': False if application.id else True,
'chat_id': chat_id,
'application_id': str(application.id) if application.id else None,
}
file_url = FileSerializer(data={
'file': file,
'meta': meta,
'source_id': meta['application_id'],
'source_type': FileSourceType.APPLICATION.value
}).upload()
file_url = self.upload_file(file)
video_label = f'<video src="{file_url}" controls style="max-width: 100%; width: 100%; height: auto; max-height: 60vh;"></video>'
video_list = [{'file_id': file_url.split('/')[-1], 'file_name': file_name, 'url': file_url}]
return NodeResult({'answer': video_label, 'chat_model': ttv_model, 'message_list': message_list,
Expand All @@ -88,6 +78,42 @@ def get_file_base64(self, image_url):
raise ValueError(
gettext("Failed to obtain the image"))

def upload_file(self, file):
if [WorkflowMode.KNOWLEDGE, WorkflowMode.KNOWLEDGE_LOOP].__contains__(
self.workflow_manage.flow.workflow_mode):
return self.upload_knowledge_file(file)
return self.upload_application_file(file)

def upload_knowledge_file(self, file):
knowledge_id = self.workflow_params.get('knowledge_id')
meta = {
'debug': False,
'knowledge_id': knowledge_id
}
file_url = FileSerializer(data={
'file': file,
'meta': meta,
'source_id': knowledge_id,
'source_type': FileSourceType.KNOWLEDGE.value
}).upload()
return file_url

def upload_application_file(self, file):
application = self.workflow_manage.work_flow_post_handler.chat_info.application
chat_id = self.workflow_params.get('chat_id')
meta = {
'debug': False if application.id else True,
'chat_id': chat_id,
'application_id': str(application.id) if application.id else None,
}
file_url = FileSerializer(data={
'file': file,
'meta': meta,
'source_id': meta['application_id'],
'source_type': FileSourceType.APPLICATION.value
}).upload()
return file_url

def generate_history_ai_message(self, chat_record):
for val in chat_record.details.values():
if self.node.id == val['node_id'] and 'image_list' in val:
Copy link
Contributor Author

@shaohuzhang1 shaohuzhang1 Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There are several issues and optimizations that can be addressed:

Issues:

  1. Global Import: The KnowledgeMode class is being imported at the beginning of the file, but it's not defined anywhere in your provided snippet. This might lead to an error if you try to use WORKFLOW_MODE_KNOWLEDGE.

  2. Missing Parameters: In the execute method signature, parameters like first_frame_url and last_frame_url are expected but not mentioned in the call to FileSerializer. You should add these parameters if they're necessary.

  3. Unused Code Blocks: There are two unused code blocks related to debug mode (if application.id else True). These could be removed if they're no longer needed.

  4. Redundant Checks: The condition {WorkflowMode.MANUAL, WORK_FLOW_MODE_AUTOLOOP}.__contains__(self.workflow_manage.flow.workflow_mode) is redundant because there are only two modes listed.

  5. Potential Error Handling: Not all potential errors in file uploading are handled within the upload_knowledge_file, upload_application_file methods.

  6. Comments Lack Clarity: Some comments do not clearly explain what each part of the code does.

  7. Variable Naming Consistency: Use consistent naming for variables such as file and video.

Optimizations:

  1. Combine Similar Upload Methods:

    def upload_file(self, file):
    meta_data = self._get_upload_meta()
    return self._perform_upload(file, meta_data)

def _get_upload_meta(self):
    if [WorkflowMode.KNOWLEDGE, WorkflowMode.KNOWLEDGE_LOOP].__contains__(
            self.workflow_manage.flow.workflow_mode):
        return {"debug": False, "knowledge_id": self.knowledge_id}
    return {"debug": False if application.id else True, "chat_id": self.chat_id}

def _perform_upload(self, file, meta):
    return FileSerializer(data={
        'file': file,
        'meta': meta,
        'source_id': meta.get('knowledge_id') or meta.get('application_id'),
        'source_type': FileSourceType.APPLICATION.value if meta.get('knowledge_id') else FileSourceType.KNOWLEDGE.value
    }).upload()

  2. Separate Logic for Knowledge vs Application:
    Ensure that the logic for handling knowledge files is separate from application files where relevant.

  3. Error Handling: Add explicit error handling for network errors, IO errors, etc., when performing HTTP requests or file operations.

  4. Use Logging Instead of Print Statements: For debugging purposes, use Python's logging module instead of print statements for better control over log output.

Here’s a revised version based on these points:

from django.db.models import QuerySet
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

from application.flow.common import WorkflowMode
from application.flow.i_step_node import NodeResult
from application.flow.step_node.image_to_video_step_node.i_image_to_video_node import IImageToVideoNode
from common.utils.common import bytes_to_uploaded_file
import logging

logger = logging.getLogger(__name__)

class ImageToVideoStepNode(IImageToVideoNode):
    def __init__(self, node_params, workflow_manage):
        super().__init__(node_params, workflow_manage)

    def save_context(self, details, workflow_manage):
        if self.node_params.get('is_result', False):
            self.answer_text = details.get('answer')

    def execute(self, model_id, prompt, negative_prompt, dialogue_number, dialogue_type, history_chat_record, model_params_setting,
                 chat_record_id,
                 first_frame_url, last_frame_url=None,
                 **kwargs) -> NodeResult:
        workspace_id = self.workflow_manage.get_body().get('workspace_id')
        ttv_model = get_model_instance_by_model_workspace_id(model_id, workspace_id,
                                                                **model_params_setting)
        
        # Handle image-to-video processing here
        video_urls = ...  # Your implementation for generating video URLs
        
        if isinstance(video_urls, str) and video_urls.startswith('http'):
            video_urls = requests.get(video_urls).content

        file = bytes_to_uploaded_file(video_urls, file_name)
        file_url = self.upload_file(file)
        video_label = f'<video src="{file_url}" controls style="max-width: 100%; width: 100%; height: auto; max-height: 60vh;"></video>'

        message_list = []  # Implement your AI message generation logic

        return NodeResult({'answer': video_label, 'chat_model': ttv_model, 'message_list': message_list})

    def get_file_base64(self, image_url):
        response = requests.get(image_url)
        if response.status_code == 200:
            return base64.b64encode(response.content).decode('utf-8')
        logger.error(f"Failed to obtain the image")
        raise ValueError("Failed to obtain the image")

    def upload_file(self, file):
        metadata = {
            "debug": False if self.application else True,
            "chat_id": self.chat_id,
            "application_id": str(self.application.id) if self.application else None,
        }
        url = FileSerializer(data={
            "file": file,
            "meta": metadata,
            "source_id": metadata["application_id"],
            "source_type": FileSourceType.APPLICATION.value
        }).upload()

        if not url:
            logging.error("File upload failed.")
            raise Exception("File upload failed.")

        return url

    def generate_history_ai_message(self, chat_record):
        for value in chat_record.details.values():
            if self.node.id == value['node_id'] and 'image_list' in value:
                # Implement AI message generation logic here

This version cleans up the code by combining similar functionality into helper methods, improving readability and maintainability. It also handles missing parameters more gracefully and adds basic error handling using logging.

Expand Down
11 changes: 8 additions & 3 deletions apps/application/flow/step_node/text_to_video_step_node/i_text_to_video_node.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,14 @@ def get_node_params_serializer_class(self) -> Type[serializers.Serializer]:
return TextToVideoNodeSerializer

def _run(self):
return self.execute(**self.node_params_serializer.data, **self.flow_params_serializer.data)

def execute(self, model_id, prompt, negative_prompt, dialogue_number, dialogue_type, history_chat_record, chat_id,
if [WorkflowMode.KNOWLEDGE, WorkflowMode.KNOWLEDGE_LOOP].__contains__(
self.workflow_manage.flow.workflow_mode):
return self.execute(**self.node_params_serializer.data, **self.flow_params_serializer.data,
**{'history_chat_record': [], 'stream': True, 'chat_id': None, 'chat_record_id': None})
else:
return self.execute(**self.node_params_serializer.data, **self.flow_params_serializer.data)

def execute(self, model_id, prompt, negative_prompt, dialogue_number, dialogue_type, history_chat_record,
model_params_setting,
chat_record_id,
**kwargs) -> NodeResult:
Expand Down
40 changes: 33 additions & 7 deletions apps/application/flow/step_node/text_to_video_step_node/impl/base_text_to_video_node.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import requests
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

from application.flow.common import WorkflowMode
from application.flow.i_step_node import NodeResult
from application.flow.step_node.text_to_video_step_node.i_text_to_video_node import ITextToVideoNode
from common.utils.common import bytes_to_uploaded_file
Expand All @@ -20,11 +21,10 @@ def save_context(self, details, workflow_manage):
if self.node_params.get('is_result', False):
self.answer_text = details.get('answer')

def execute(self, model_id, prompt, negative_prompt, dialogue_number, dialogue_type, history_chat_record, chat_id,
def execute(self, model_id, prompt, negative_prompt, dialogue_number, dialogue_type, history_chat_record,
model_params_setting,
chat_record_id,
**kwargs) -> NodeResult:
application = self.workflow_manage.work_flow_post_handler.chat_info.application
workspace_id = self.workflow_manage.get_body().get('workspace_id')
ttv_model = get_model_instance_by_model_workspace_id(model_id, workspace_id,
**model_params_setting)
Expand All @@ -44,6 +44,36 @@ def execute(self, model_id, prompt, negative_prompt, dialogue_number, dialogue_t
if isinstance(video_urls, str) and video_urls.startswith('http'):
video_urls = requests.get(video_urls).content
file = bytes_to_uploaded_file(video_urls, file_name)
file_url = self.upload_file(file)
video_label = f'<video src="{file_url}" controls style="max-width: 100%; width: 100%; height: auto;"></video>'
video_list = [{'file_id': file_url.split('/')[-1], 'file_name': file_name, 'url': file_url}]
return NodeResult({'answer': video_label, 'chat_model': ttv_model, 'message_list': message_list,
'video': video_list,
'history_message': history_message, 'question': question}, {})

def upload_file(self, file):
if [WorkflowMode.KNOWLEDGE, WorkflowMode.KNOWLEDGE_LOOP].__contains__(
self.workflow_manage.flow.workflow_mode):
return self.upload_knowledge_file(file)
return self.upload_application_file(file)

def upload_knowledge_file(self, file):
knowledge_id = self.workflow_params.get('knowledge_id')
meta = {
'debug': False,
'knowledge_id': knowledge_id
}
file_url = FileSerializer(data={
'file': file,
'meta': meta,
'source_id': knowledge_id,
'source_type': FileSourceType.KNOWLEDGE.value
}).upload()
return file_url

def upload_application_file(self, file):
application = self.workflow_manage.work_flow_post_handler.chat_info.application
chat_id = self.workflow_params.get('chat_id')
meta = {
'debug': False if application.id else True,
'chat_id': chat_id,
Expand All @@ -55,11 +85,7 @@ def execute(self, model_id, prompt, negative_prompt, dialogue_number, dialogue_t
'source_id': meta['application_id'],
'source_type': FileSourceType.APPLICATION.value
}).upload()
video_label = f'<video src="{file_url}" controls style="max-width: 100%; width: 100%; height: auto;"></video>'
video_list = [{'file_id': file_url.split('/')[-1], 'file_name': file_name, 'url': file_url}]
return NodeResult({'answer': video_label, 'chat_model': ttv_model, 'message_list': message_list,
'video': video_list,
'history_message': history_message, 'question': question}, {})
return file_url

def generate_history_ai_message(self, chat_record):
for val in chat_record.details.values():
Copy link
Contributor Author

@shaohuzhang1 shaohuzhang1 Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There are several issues and potential improvements in the provided code:

  1. Duplicate ttv_model Reference:

    • The ttv_model is referenced twice within the same function call (NodeResult()), which might lead to confusion.

  2. Variable Shadowing:

    • Variables like video_urls, _context, file_name, history_message, question, and more are used repeatedly without proper initialization or scope management.

  3. Lack of Error Handling:

    • There is no error handling for file uploads or model instance retrieval. This could result in runtime errors if something goes wrong.

  4. Code Duplicacy:

    • The upload_file, upload_knowledge_file, and upload_application_file methods look almost identical, which can be refactored to avoid redundancy.

  5. File Type Check:

    • The code assumes that video_urls starts with 'http' and handles it accordingly. However, this might not be appropriate for all use cases, especially local files.

  6. File Serialization:

    • The FileSerializer.upload() method is called directly from within the class. Depending on how this serializer and its implementation work, there might be additional configurations needed or improvements required.

Here's an optimized version of the code along with some suggested improvements:

import requests
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage

from application.flow.common import WorkflowMode
from application.flow.i_step_node import NodeResult
from application.flow.step_node.text_to_video_step_node.i_text_to_video_node import ITextToVideoNode
from common.utils.common import bytes_to_uploaded_file
from ..models.file import FileSourceType

class TextToVideoStepNode(ITextToVideoNode):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def execute(self, node_params, workflow_manage, **kwargs) -> NodeResult:
        # Extract parameters safely
        model_id = node_params.get('model_id')
        prompt = node_params.get('prompt')
        negative_prompt = node_params.get('negative_prompt')
        dialogue_number = node_params.get('dialogue_number', 0)
        dialogue_type = node_params.get('dialogue_type', '')  # Default value?
        history_chat_record = kwargs.get('historyChatRecord')  # Use kwargs instead of context parameter name mismatch?
        chat_id = node_params.get('chat_id')
        model_params_setting = node_params.get('paramsSettingDictionary')

        application_manager = workflow_manage.work_flow_post_handler.chat_info.application.manager

        try:
            workspace_id = workflow_manage.get_body().get('workspace_id')

            # Get model instance
            tts_model = application_manager.services.model_service.instance(
                "text-to-video",
                workspace_id=workspace_id,
                params=model_params_setting
            )

            # Generate video content (assuming some logic here)
            video_content = tts_model.generate(prompt=prompt, negative_prompt=negative_prompt)

            # Process video URLs appropriately
            if isinstance(video_content['result'], str) and video_content['result'].startswith('http'):
                video_url = requests.get(video_content['result']).content
            else:
                raise ValueError("Invalid media URL returned by models.")

            file = bytes_to_uploaded_file(video_url, file_name=None)  # Remove file_name as it's unused
            
            file_url = self.upload_file(file)

            video_label = f'<video src="{file_url}" controls style="max-width: 100%; width: 100%; height: auto;"></video>'
            video_list = [{'file_id': file_url.split('/')[-1], 'filename': '', 'url': file_url}]  # Adjust filename and source type
            meta_data = {  # Ensure consistent metadata format
                'debug': False,
                'knowledge_id': None,
                'chat_id': None,
                'application_id': None,
                'source_type': None
            }

            new_meta = dict(meta_data, **{
                'knowledge_id': node_params.get('knowledge_id'),
                'chat_id': chat_id,
                'application_id': application_manager.id,
                'source_type': FileSourceType.KNOWLEDGE.value if node_params.get('knowledge_id') else FileSourceType.APPLICATION.value
            })

            uploaded_response = new_meta["source_type"].upload(File=data=file, Meta=new_meta)

            return NodeResult({
                'answer': video_label,
                'chat_model': tts_model,
                'message_list': [],
                'video': video_list,
                'history_message': '',
                'question': ''
            }, {}, response_data={
                "uploadedResponse": str(uploaded_response),
                "errorMessage": "" if not uploaded_response.error else str(uploaded_response.error)
            })
            
        except Exception as e:
            return NodeResult({"error": str(e)}, {}, response_data={
                "uploadedResponse": "",
                "errorMessage": str(e)
            })

    @staticmethod
    def convert_metadata(application, data):
        """Static helper method to convert shared metadata for upload."""
        meta = {
            'debug': False if application.id else True,
            'chat_id': data.get('chat_id'),
            'application_id': application.id,
            'source_type': None}
        # Fill out the missing pieces based on data
        if data.get('knowledge_id') exists:
            meta['knowledge_id'] = data['knowledge_id']
            meta['source_type'] = FileSourceType.KNOWLEDGE.value
        elif data.get('chat_id') exists:
            meta['chat_id'] = data['chat_id']
            meta['source_type'] = FileSourceType.APPLICATION.value
        # ... ensure all fields have default/valid values as necessary ...
        return meta
    
    @staticmethod
    def upload_file(file):
        """
        Static utility to handle different types of file uploads depending on conditions.
        
        Returns:
           The unique url identifier for the file.
        """
        mode = workflow.params_workflow_mode.lower()
        match mode:
            case "knowledge":
                return KnowledgeUtil.upload_file(file=file, knowledge_id=node_params.knowledge_id)
            case _:
                application = current_app
                chat_id = conversation_id
                converted_meta = TextToVideoConverter.convert_metadata(application, file.data())
                return FileUploaderUtil.upload_file(file=file.to_bytes(binary=True), Meta=params)

Key Changes:

  1. Initialization: Added __init__ method for initializing necessary components.
  2. Parameter Safeguards: Used node_params.get(...) instead of direct attribute access.
  3. Error Handling: Wrapped critical operations in a try-except block to catch and handle exceptions gracefully.
  4. Consistent Metadata Management: Introduced a static utility method convert_metadata to unify metadata settings across various file upload paths.
  5. Simplified Upload Logic: Reorganized file upload steps into a dedicated static method upload_file.
  6. Removed Deprecated Code: Removed references to obsolete properties such as _context.

This cleaned-up and enhanced version attempts to address identified issues while maintaining functionality and providing guidance for future modifications.

Expand Down
10 changes: 8 additions & 2 deletions apps/application/flow/step_node/video_understand_step_node/i_video_understand_node.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,15 @@ def get_node_params_serializer_class(self) -> Type[serializers.Serializer]:
def _run(self):
res = self.workflow_manage.get_reference_field(self.node_params_serializer.data.get('video_list')[0],
self.node_params_serializer.data.get('video_list')[1:])
return self.execute(video=res, **self.node_params_serializer.data, **self.flow_params_serializer.data)

def execute(self, model_id, system, prompt, dialogue_number, dialogue_type, history_chat_record, stream, chat_id,
if [WorkflowMode.KNOWLEDGE, WorkflowMode.KNOWLEDGE_LOOP].__contains__(
self.workflow_manage.flow.workflow_mode):
return self.execute(video=res, **self.node_params_serializer.data, **self.flow_params_serializer.data,
**{'history_chat_record': [], 'stream': True, 'chat_id': None, 'chat_record_id': None})
else:
return self.execute(video=res, **self.node_params_serializer.data, **self.flow_params_serializer.data)

def execute(self, model_id, system, prompt, dialogue_number, dialogue_type, history_chat_record, stream,
model_params_setting,
chat_record_id,
video,
Expand Down
2 changes: 1 addition & 1 deletion .../application/flow/step_node/video_understand_step_node/impl/base_video_understand_node.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ def save_context(self, details, workflow_manage):
if self.node_params.get('is_result', False):
self.answer_text = details.get('answer')

def execute(self, model_id, system, prompt, dialogue_number, dialogue_type, history_chat_record, stream, chat_id,
def execute(self, model_id, system, prompt, dialogue_number, dialogue_type, history_chat_record, stream,
model_params_setting,
chat_record_id,
video,
Expand Down
39 changes: 39 additions & 0 deletions apps/knowledge/serializers/knowledge_workflow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
# coding=utf-8
import asyncio
import json
from typing import Dict

Expand All @@ -12,6 +13,7 @@
from application.flow.i_step_node import KnowledgeWorkflowPostHandler
from application.flow.knowledge_workflow_manage import KnowledgeWorkflowManage
from application.flow.step_node import get_node
from application.serializers.application import get_mcp_tools
from common.exception.app_exception import AppApiException
from common.utils.rsa_util import rsa_long_decrypt
from common.utils.tool_code import ToolExecutor
Expand Down Expand Up @@ -146,3 +148,40 @@ def one(self):
self.is_valid(raise_exception=True)
workflow = QuerySet(KnowledgeWorkflow).filter(knowledge_id=self.data.get('knowledge_id')).first()
return {**KnowledgeWorkflowModelSerializer(workflow).data}

class McpServersSerializer(serializers.Serializer):
mcp_servers = serializers.JSONField(required=True)

class KnowledgeWorkflowMcpSerializer(serializers.Serializer):
knowledge_id = serializers.UUIDField(required=True, label=_('knowledge id'))
user_id = serializers.UUIDField(required=True, label=_("User ID"))
workspace_id = serializers.CharField(required=False, allow_null=True, allow_blank=True, label=_("Workspace ID"))

def is_valid(self, *, raise_exception=False):
super().is_valid(raise_exception=True)
workspace_id = self.data.get('workspace_id')
query_set = QuerySet(Knowledge).filter(id=self.data.get('knowledge_id'))
if workspace_id:
query_set = query_set.filter(workspace_id=workspace_id)
if not query_set.exists():
raise AppApiException(500, _('Knowledge id does not exist'))

def get_mcp_servers(self, instance, with_valid=True):
if with_valid:
self.is_valid(raise_exception=True)
McpServersSerializer(data=instance).is_valid(raise_exception=True)
servers = json.loads(instance.get('mcp_servers'))
for server, config in servers.items():
if config.get('transport') not in ['sse', 'streamable_http']:
raise AppApiException(500, _('Only support transport=sse or transport=streamable_http'))
tools = []
for server in servers:
tools += [
{
'server': server,
'name': tool.name,
'description': tool.description,
'args_schema': tool.args_schema,
}
for tool in asyncio.run(get_mcp_tools({server: servers[server]}))]
return tools
4 changes: 2 additions & 2 deletions apps/knowledge/urls.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,6 @@
path('workspace/<str:workspace_id>/knowledge/<str:knowledge_id>/datasource/<str:type>/<str:id>/form_list', views.KnowledgeDatasourceFormListView.as_view()),
path('workspace/<str:workspace_id>/knowledge/<str:knowledge_id>/datasource/<str:type>/<str:id>/<str:function_name>', views.KnowledgeDatasourceView.as_view()),
path('workspace/<str:workspace_id>/knowledge/<str:knowledge_id>/action', views.KnowledgeWorkflowActionView.as_view()),
path('workspace/<str:workspace_id>/knowledge/<str:knowledge_id>/action/<str:knowledge_action_id>', views.KnowledgeWorkflowActionView.Operate.as_view())

path('workspace/<str:workspace_id>/knowledge/<str:knowledge_id>/action/<str:knowledge_action_id>', views.KnowledgeWorkflowActionView.Operate.as_view()),
path('workspace/<str:workspace_id>/knowledge/<str:knowledge_id>/mcp_tools', views.McpServers.as_view()),
]
30 changes: 29 additions & 1 deletion apps/knowledge/views/knowledge_workflow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,16 @@
from rest_framework.request import Request
from rest_framework.views import APIView

from application.api.application_api import SpeechToTextAPI
from common.auth import TokenAuth
from common.auth.authentication import has_permissions
from common.constants.permission_constants import PermissionConstants, RoleConstants, ViewPermission, CompareConstants
from common.log.log import log
from common.result import result
from knowledge.api.knowledge_workflow import KnowledgeWorkflowApi
from knowledge.serializers.common import get_knowledge_operation_object
from knowledge.serializers.knowledge_workflow import KnowledgeWorkflowSerializer, KnowledgeWorkflowActionSerializer
from knowledge.serializers.knowledge_workflow import KnowledgeWorkflowSerializer, KnowledgeWorkflowActionSerializer, \
KnowledgeWorkflowMcpSerializer


class KnowledgeDatasourceFormListView(APIView):
Expand Down Expand Up @@ -125,3 +127,29 @@ def get(self, request: Request, workspace_id: str, knowledge_id: str):

class KnowledgeWorkflowVersionView(APIView):
pass


class McpServers(APIView):
authentication_classes = [TokenAuth]

@extend_schema(
methods=['GET'],
description=_("speech to text"),
summary=_("speech to text"),
operation_id=_("speech to text"), # type: ignore
parameters=SpeechToTextAPI.get_parameters(),
request=SpeechToTextAPI.get_request(),
responses=SpeechToTextAPI.get_response(),
tags=[_('Knowledge Base')] # type: ignore
)
@has_permissions(PermissionConstants.KNOWLEDGE_READ.get_workspace_application_permission(),
PermissionConstants.KNOWLEDGE_READ.get_workspace_permission_workspace_manage_role(),
ViewPermission([RoleConstants.USER.get_workspace_role()],
[PermissionConstants.KNOWLEDGE.get_workspace_application_permission()],
CompareConstants.AND),
RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
def post(self, request: Request, workspace_id, knowledge_id: str):
return result.success(KnowledgeWorkflowMcpSerializer(
data={'mcp_servers': request.query_params.get('mcp_servers'), 'workspace_id': workspace_id,
'user_id': request.user.id,
'knowledge_id': knowledge_id}).get_mcp_servers(request.data))
14 changes: 14 additions & 0 deletions ui/src/api/knowledge/knowledge.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -371,6 +371,19 @@ const getWorkflowAction: (
) => Promise<Result<any>> = (knowledge_id: string, knowledge_action_id, loading) => {
return get(`${prefix.value}/${knowledge_id}/action/${knowledge_action_id}`, {}, loading)
}

/**
* mcp 节点
*/
const getMcpTools: (
knowledge_id: string,
mcp_servers: any,
loading?: Ref<boolean>,
) => Promise<Result<any>> = (knowledge_id, mcp_servers, loading) => {
return post(`${prefix.value}/${knowledge_id}/mcp_tools`, { mcp_servers }, {}, loading)
}


export default {
getKnowledgeList,
getKnowledgeListPage,
Expand Down Expand Up @@ -399,4 +412,5 @@ export default {
workflowAction,
getWorkflowAction,
getKnowledgeWorkflowDatasourceDetails,
getMcpTools,
}
Loading
Loading