Pipeline Knowledge Base Service API Implementation Documentation

[fatelei]

Self Checks

• I have searched for existing issues search for existing issues, including closed ones.
• I confirm that I am using English to submit this report (我已阅读并同意 Language Policy).
• [FOR CHINESE USERS] 请务必使用英文提交 Issue，否则会被关闭。谢谢！:)
• Please do not modify this template :) and fill in all the required fields.

Pipeline Knowledge Base Service API Implementation Documentation

1. Background and Requirements

1.1 Requirements

Add the following Service API endpoint support for Pipeline Knowledge Bases:

1. Data Preparation Endpoints (do NOT execute workflow):

   • Prepare document data structure for all datasource types (local_file, online_document, website_crawl, online_drive)
   • Support file upload, text content, or direct datasource_info_list input
   • Automatically inject credential_id from datasource node configuration
   • Return prepared data structure ready for PipelineGenerateService.generate()

2. Unified Endpoints (instead of separate text/file endpoints):

   • POST /v1/datasets/<dataset_id>/pipeline/documents - Prepare data for document creation
   • PATCH /v1/datasets/<dataset_id>/pipeline/documents/<document_id> - Prepare data for document update

3. Users execute workflow separately:

   • Use POST /v1/datasets/<dataset_id>/pipeline/run endpoint with prepared data
   • Or use PipelineGenerateService.generate() directly

2. Technical Analysis

Correct Approach:

• Data Preparation: Prepare data structure with datasource_info_list (automatically inject credential_id)
• Workflow Execution: Users call PipelineGenerateService.generate() or /pipeline/run endpoint separately
• Pipeline Workflow automatically creates and processes documents (datasource node → knowledge-index node)

2.3 Pipeline Workflow Document Processing Flow

User uploads file/text/datasource_info_list
    ↓
PipelineGenerateService.generate()
    ↓
PipelineGenerator.generate()
    ↓
Create Document (if invoke_from == PUBLISHED_PIPELINE)
    ↓
Run Pipeline Workflow
    ↓
Datasource Node (process data source)
    ↓
Knowledge-Index Node (index document)
    ↓
Document indexing completed

2.4 How Pipeline Handles online_document Datasource Type

When pipeline runs with online_document datasource type, the processing flow is:

1. DatasourceNode receives datasource_info_list with online_document type
    ↓
2. Extract datasource_info from datasource_info_list:
   - workspace_id: Workspace ID (e.g., Notion workspace)
   - page: Page information object
     - page_id: Page ID
     - page_name: Page name
     - type: Page type (page, database, etc.)
   - credential_id: Credential ID for authentication
    ↓
3. Get credentials from DatasourceProviderService using credential_id
    ↓
4. Call OnlineDocumentDatasourcePlugin.get_online_document_page_content():
   - Uses PluginDatasourceManager to communicate with plugin daemon
   - Plugin daemon calls actual provider (e.g., Notion API)
   - Fetches page content from online document service
    ↓
5. Receive DatasourceMessage stream from plugin:
   - TEXT: Page text content
   - IMAGE/BINARY: Images and files embedded in the page
   - JSON: Structured data
   - LINK: Links in the page
    ↓
6. Transform messages via _transform_message():
   - Extract text content
   - Download and store images/files as ToolFile
   - Convert to File objects for workflow
   - Store in variable pool for next node
    ↓
7. Pass transformed content to Knowledge-Index Node:
   - Text content is chunked and indexed
   - Files are processed and indexed
   - Document is created/updated in knowledge base

Key Points:

• Real-time fetching: Content is fetched from online document service (e.g., Notion) during pipeline execution, not pre-cached
• Credential-based authentication: Uses credential_id to authenticate with the online document service
• Streaming processing: Content is processed in a streaming manner, allowing large documents to be handled efficiently
• Multi-format support: Handles text, images, files, JSON, and links from the online document
• Plugin-based architecture: Uses plugin system to support multiple online document providers (Notion, etc.)

3. Implementation Plan

3.1 Design Decision

Create new endpoints instead of modifying existing ones

Reasons:

1. Clear separation of concerns: Pipeline and Vendor knowledge bases have different processing flows
2. Follows existing pattern: Other pipeline endpoints use /datasets/<id>/pipeline/... path
3. Explicit API design: Users clearly know they are using pipeline knowledge base
4. No breaking changes: Existing Vendor Knowledge Base endpoints remain unchanged

3.2 API Endpoint Design

Unified Data Preparation Endpoints (do NOT execute workflow):

Create Endpoint:

POST /v1/datasets/<uuid:dataset_id>/pipeline/documents

Update Endpoint:

PATCH /v1/datasets/<uuid:dataset_id>/pipeline/documents/<uuid:document_id>

Key Design Points:

1. Single unified endpoint for all datasource types (local_file, online_document, website_crawl, online_drive)
2. Data preparation only - does NOT execute PipelineGenerateService.generate()
3. Automatic datasource node detection: Automatically gets the first datasource node from workflow (start_node_id is optional)
4. Automatic datasource type and provider detection:
   • Datasource type (online_document vs local_file, etc.) is obtained from node_data.provider_type
   • Specific provider (Notion vs Google Drive) is obtained from node_data.provider_name and node_data.plugin_id
   • If pipeline has multiple datasource nodes, can specify via start_node_id parameter
5. Automatic credential_id injection from datasource node default credential
6. Differentiated responses:
   • For local_file and website_crawl: Returns datasource_info_list
   • For online_document and online_drive: Returns document list (datasource_info_list), users need to select before execution

3.3 File Organization

api/controllers/service_api/dataset/rag_pipeline/
├── __init__.py
├── rag_pipeline_workflow.py (existing)
└── rag_pipeline_document.py (new)

4. Implementation Details

4.1 Core Implementation Logic

4.1.1 Data Preparation Flow (No Execution)

def prepare_document_data(
    dataset_id: str,
    document_id: str | None,
    start_node_id: str | None,
    request_data: dict,
    files: dict,
    user: Account
):
    """Prepare data structure for PipelineGenerateService.generate() - does NOT execute workflow"""
    
    # 1. Validate it's a Pipeline Knowledge Base
    dataset = get_dataset(dataset_id)
    _validate_pipeline_knowledge_base(dataset)
    
    # 2. Get Pipeline and Workflow
    pipeline = get_pipeline(dataset.pipeline_id)
    workflow = rag_pipeline_service.get_published_workflow(pipeline)
    
    # 3. Automatically get datasource node (if start_node_id not provided, auto-select first)
    if not start_node_id:
        start_node_id = _get_first_datasource_node_id(workflow)
    datasource_type, node_data = _get_datasource_type_from_node(workflow, start_node_id)
    
    # 4. Build datasource_info_list (automatically injects credential_id)
    datasource_info_list = _build_datasource_info_list(
        datasource_type=datasource_type,
        request_data=request_data,
        files=files,
        node_data=node_data,
        tenant_id=pipeline.tenant_id
    )
    
    # 5. Prepare response
    inputs = request_data.get("inputs", {})
    
    # For online_document and online_drive, get document list and only return datasource_info_list
    if datasource_type in ["online_document", "online_drive"]:
        # Call get_online_document_pages() or similar method to get document list
        datasource_info_list = _get_online_document_list(
            datasource_type=datasource_type,
            node_data=node_data,
            tenant_id=pipeline.tenant_id,
            user_id=user.id
        )
        return {
            "start_node_id": start_node_id,
            "datasource_type": datasource_type,
            "datasource_info_list": datasource_info_list,
        }
    
    
    return {
        "start_node_id": start_node_id,
        "datasource_type": datasource_type,
        "datasource_info_list": datasource_info_list,
    }

4.2 Key Function Implementations

4.2.1 Pipeline Knowledge Base Validation

def _validate_pipeline_knowledge_base(dataset: Dataset) -> None:
    """Validate that dataset is a Pipeline Knowledge Base"""
    if dataset.runtime_mode != "rag_pipeline":
        raise ValueError(
            "This endpoint only supports Pipeline Knowledge Bases. "
            "For Vendor Knowledge Bases, use /datasets/<id>/document/create_by_text."
        )
    if not dataset.pipeline_id:
        raise ValueError("Pipeline Knowledge Base must have a pipeline_id.")

4.2.2 Get First Datasource Node ID

def _get_first_datasource_node_id(workflow: Workflow) -> str:
    """Get the first datasource node ID from workflow"""
    nodes = workflow.graph_dict.get("nodes", [])
    for node in nodes:
        node_data = node.get("data", {})
        if node_data.get("type") == "datasource":
            return node.get("id")
    raise ValueError("No datasource node found in workflow")

4.2.3 Get Datasource Type from Node

def _get_datasource_type_from_node(
    workflow: Workflow, 
    start_node_id: str
) -> tuple[str, dict]:
    """Get datasource type and node data from start_node_id
    
    From datasource node configuration, we can get:
    - provider_type: Datasource type (e.g., "online_document", "local_file", "website_crawl", "online_drive")
    - provider_name: Provider name (e.g., "notion", "google_drive")
    - plugin_id: Plugin ID (e.g., "langgenius/notion_datasource")
    
    This information is used to:
    1. Determine datasource type (online_document vs local_file)
    2. Determine specific provider (Notion vs Google Drive)
    3. Get corresponding credentials and configuration
    """
    nodes = workflow.graph_dict.get("nodes", [])
    for node in nodes:
        if node.get("id") == start_node_id:
            node_data = node.get("data", {})
            if node_data.get("type") != "datasource":
                raise ValueError(f"Node {start_node_id} is not a datasource node")
            datasource_type = node_data.get("provider_type")
            # node_data contains:
            # - provider_type: "online_document"
            # - provider_name: "notion" or "google_drive", etc.
            # - plugin_id: "langgenius/notion_datasource", etc.
            return datasource_type, node_data
    raise ValueError(f"Datasource node {start_node_id} not found")

4.2.4 Get Default Credential ID

def _get_default_credential_id(
    tenant_id: str,
    provider: str,
    plugin_id: str
) -> str | None:
    """Get default credential ID for datasource provider"""
    datasource_provider_service = DatasourceProviderService()
    credentials = datasource_provider_service.list_datasource_credentials(
        tenant_id=tenant_id,
        provider=provider,
        plugin_id=plugin_id,
    )
    # Find default credential
    for credential in credentials:
        if credential.get("is_default"):
            return credential.get("id")
    # If no default, return first available
    if credentials:
        return credentials[0].get("id")
    return None

4.2.5 Get Online Document List

def _get_online_document_list(
    datasource_type: str,
    node_data: dict,
    tenant_id: str,
    user_id: str
) -> list[dict]:
    """Get document list for online documents (e.g., Notion, Google Drive)
    
    Parameter explanation:
    - datasource_type: Obtained from node_data.provider_type (e.g., "online_document")
    - node_data.provider_name: Provider name (e.g., "notion", "google_drive")
    - node_data.plugin_id: Plugin ID (e.g., "langgenius/notion_datasource")
    
    The system automatically determines whether it's Notion or Google Drive based on this information
    and calls the corresponding plugin to get the document list.
    """
    from core.datasource.datasource_manager import DatasourceManager
    from core.datasource.online_document.online_document_plugin import OnlineDocumentDatasourcePlugin
    from core.datasource.online_drive.online_drive_plugin import OnlineDriveDatasourcePlugin
    
    # Get provider information from node_data
    provider_name = node_data.get('provider_name')  # e.g., "notion", "google_drive"
    plugin_id = node_data.get('plugin_id')  # e.g., "langgenius/notion_datasource"
    
    # Get datasource runtime (automatically determines Notion or Google Drive based on provider_name and plugin_id)
    datasource_runtime = DatasourceManager.get_datasource_runtime(
        provider_id=f"{plugin_id}/{provider_name}",
        datasource_name=node_data.get("datasource_name") or "",
        tenant_id=tenant_id,
        datasource_type=DatasourceProviderType.value_of(datasource_type),
    )
    
    # Get default credentials
    datasource_provider_service = DatasourceProviderService()
    default_credential_id = _get_default_credential_id(
        tenant_id=tenant_id,
        provider=node_data.get("provider_name"),
        plugin_id=node_data.get("plugin_id"),
    )
    credentials = None
    if default_credential_id:
        credentials = datasource_provider_service.get_datasource_credentials(
            tenant_id=tenant_id,
            provider=node_data.get("provider_name"),
            plugin_id=node_data.get("plugin_id"),
            credential_id=default_credential_id,
        )
    
    if credentials:
        datasource_runtime.runtime.credentials = credentials
    
    # Get document list
    if datasource_type == "online_document":
        datasource_runtime = cast(OnlineDocumentDatasourcePlugin, datasource_runtime)
        # Get datasource_parameters from node_data
        datasource_parameters = node_data.get("datasource_parameters", {})
        pages_generator = datasource_runtime.get_online_document_pages(
            user_id=user_id,
            datasource_parameters=datasource_parameters,
            provider_type=datasource_type,
        )
        # Collect all document lists
        result = []
        for message in pages_generator:
            for info in message.result:
                result.append({
                    "workspace_id": info.workspace_id,
                    "workspace_name": info.workspace_name,
                    "workspace_icon": info.workspace_icon,
                    "pages": [
                        {
                            "page_id": page.page_id,
                            "page_name": page.page_name,
                            "type": page.type,
                            "parent_id": page.parent_id,
                            "page_icon": page.page_icon,
                            "is_bound": False,
                        }
                        for page in info.pages
                    ],
                    "credential_id": default_credential_id,
                })
        return result
    elif datasource_type == "online_drive":
        # Similar handling for online_drive
        # TODO: Implement online_drive document list retrieval
        raise NotImplementedError("online_drive document list not implemented yet")
    else:
        raise ValueError(f"Unsupported datasource type for document list: {datasource_type}")

4.2.6 Build Datasource Info List

def _build_datasource_info_list(
    datasource_type: str,
    request_data: dict,
    files: dict,
    node_data: dict,
    tenant_id: str
) -> list[dict]:
    """Build datasource_info_list based on datasource type and input"""
    
    # Get default credential_id from node configuration
    default_credential_id = None
    if datasource_type != "local_file":
        default_credential_id = _get_default_credential_id(
            tenant_id=tenant_id,
            provider=node_data.get("provider_name"),
            plugin_id=node_data.get("plugin_id"),
        )
    
    if datasource_type == "local_file":
        # Option 1: File upload
        if "file" in files:
            file = files["file"]
            upload_file = FileService.upload_file(...)
            return [{
                "related_id": upload_file.id,
                "name": upload_file.name,
                "type": upload_file.type,
                "size": upload_file.size,
                "extension": upload_file.extension,
                "mime_type": upload_file.mime_type,
                "url": "",
                "transfer_method": "local_file",
            }]
        
        # Option 2: Text content
        elif "text" in request_data:
            upload_file = FileService.upload_text(...)
            return [{
                "related_id": upload_file.id,
                "name": upload_file.name,
                "type": upload_file.type,
                "size": upload_file.size,
                "extension": upload_file.extension,
                "mime_type": upload_file.mime_type,
                "url": "",
                "transfer_method": "local_file",
            }]
        
        # Option 3: Direct datasource_info_list
        elif "datasource_info_list" in request_data:
            return request_data["datasource_info_list"]
        else:
            raise ValueError("Must provide file, text, or datasource_info_list for local_file")
    
    # For other types, use datasource_info_list directly and inject credential_id
    elif datasource_type in ["online_document", "website_crawl", "online_drive"]:
        if "datasource_info_list" not in request_data:
            raise ValueError(f"{datasource_type} requires datasource_info_list")
        
        datasource_info_list = request_data["datasource_info_list"]
        # Automatically inject credential_id if not provided
        for info in datasource_info_list:
            if "credential_id" not in info or not info.get("credential_id"):
                if default_credential_id:
                    info["credential_id"] = default_credential_id
        return datasource_info_list
    
    else:
        raise ValueError(f"Unsupported datasource type: {datasource_type}")

5. Complete API Usage Examples

5.1 Create Document with Local File Upload

Step 1: Prepare Data (does NOT execute workflow)

Request:

POST /v1/datasets/550e8400-e29b-41d4-a716-446655440000/pipeline/documents
Content-Type: multipart/form-data

file: [binary file data]
inputs: {}

Note: start_node_id is optional. If not provided, the system will automatically select the first datasource node in the workflow.

Response:

{
  "start_node_id": "node_123",
  "datasource_type": "local_file",
  "datasource_info_list": [
    {
      "related_id": "file_550e8400-e29b-41d4-a716-446655440001",
      "name": "document.pdf",
      "type": "document",
      "size": 102400,
      "extension": "pdf",
      "mime_type": "application/pdf",
      "url": "",
      "transfer_method": "local_file"
    }
  ]
}

Step 2: Execute Workflow (user calls separately)

POST /v1/datasets/550e8400-e29b-41d4-a716-446655440000/pipeline/run
Content-Type: application/json

{
  "inputs": {},
  "start_node_id": "node_123",
  "datasource_type": "local_file",
  "datasource_info_list": [
    {
      "related_id": "file_550e8400-e29b-41d4-a716-446655440001",
      "name": "document.pdf",
      "type": "document",
      "size": 102400,
      "extension": "pdf",
      "mime_type": "application/pdf",
      "url": "",
      "transfer_method": "local_file"
    }
  ],
  "is_published": true,
  "response_mode": "blocking"
}

5.2 Create Document with Text Content

Step 1: Prepare Data

Request:

POST /v1/datasets/550e8400-e29b-41d4-a716-446655440000/pipeline/documents
Content-Type: application/json

{
  "text": "This is the document content that will be converted to a file.",
  "inputs": {}
}

Note: start_node_id is optional. If not provided, the system will automatically select the first datasource node in the workflow.

Response:

{
  "start_node_id": "node_123",
  "datasource_type": "local_file",
  "datasource_info_list": [
    {
      "related_id": "file_550e8400-e29b-41d4-a716-446655440002",
      "name": "text_document.txt",
      "type": "text",
      "size": 512,
      "extension": "txt",
      "mime_type": "text/plain",
      "url": "",
      "transfer_method": "local_file"
    }
  ]
}

Step 2: Execute Workflow (same as Step 2 in 5.1)

5.3 Create Document with Online Document (Notion)

Step 1: Get Document List

Request:

POST /v1/datasets/550e8400-e29b-41d4-a716-446655440000/pipeline/documents
Content-Type: application/json

{}

Or specify a specific datasource node (if pipeline has multiple datasource nodes):

POST /v1/datasets/550e8400-e29b-41d4-a716-446655440000/pipeline/documents
Content-Type: application/json

{
  "start_node_id": "node_456"
}

Response:

{
  "start_node_id": "node_456",
  "datasource_type": "online_document",
  "datasource_info_list": [
    {
      "workspace_id": "workspace_123",
      "workspace_name": "My Notion Workspace",
      "workspace_icon": "https://...",
      "pages": [
        {
          "page_id": "page_456",
          "page_name": "My Notion Page 1",
          "type": "page",
          "parent_id": "parent_789",
          "page_icon": null,
          "is_bound": false
        },
        {
          "page_id": "page_789",
          "page_name": "My Notion Page 2",
          "type": "page",
          "parent_id": "parent_789",
          "page_icon": null,
          "is_bound": false
        }
      ],
      "credential_id": "cred_default_123"
    }
  ]
}

Note:

• credential_id is automatically injected by the system
• Response only contains datasource_info_list (document list), does NOT contain pipeline_run_args
• Users need to select documents from the returned document list

Step 2: User Selects Documents and Executes Workflow

User selects documents from the document list returned in Step 1, then calls /pipeline/run:

POST /v1/datasets/550e8400-e29b-41d4-a716-446655440000/pipeline/run
Content-Type: application/json

{
  "inputs": {},
  "start_node_id": "node_456",
  "datasource_type": "online_document",
  "datasource_info_list": [
    {
      "workspace_id": "workspace_123",
      "page": {
        "page_id": "page_456",
        "page_name": "My Notion Page 1",
        "type": "page",
        "parent_id": "parent_789",
        "page_icon": null,
        "is_bound": false
      },
      "credential_id": "cred_default_123"
    }
  ],
  "is_published": true,
  "response_mode": "blocking"
}

5.4 Create Document with Website Crawl

Step 1: Prepare Data

Request:

POST /v1/datasets/550e8400-e29b-41d4-a716-446655440000/pipeline/documents
Content-Type: application/json

{
  "datasource_info_list": [
    {
      "source_url": "https://example.com/page"
    }
  ],
  "inputs": {}
}

Response:

{
  "start_node_id": "node_789",
  "datasource_type": "website_crawl",
  "datasource_info_list": [
    {
      "source_url": "https://example.com/page",
      "credential_id": "cred_default_456"
    }
  ]
}

Step 2: Execute Workflow (using returned data)

5.5 Update Document

Step 1: Prepare Data

Request:

PATCH /v1/datasets/550e8400-e29b-41d4-a716-446655440000/pipeline/documents/550e8400-e29b-41d4-a716-446655440099
Content-Type: multipart/form-data

file: [updated binary file data]
inputs: {}

Note: start_node_id is optional. If not provided, the system will automatically select the first datasource node in the workflow.

Response:

{
  "start_node_id": "node_123",
  "datasource_type": "local_file",
  "datasource_info_list": [
    {
      "related_id": "file_550e8400-e29b-41d4-a716-446655440003",
      "name": "updated_document.pdf",
      "type": "document",
      "size": 204800,
      "extension": "pdf",
      "mime_type": "application/pdf",
      "url": "",
      "transfer_method": "local_file"
    }
  ]
}

Step 2: Execute Workflow (using returned data with original_document_id)

6. Important Considerations

6.1 Key Design Principles

1. Data Preparation Only: Endpoints only prepare and validate data, do NOT execute PipelineGenerateService.generate()
2. Separation of Concerns: Data preparation is separated from workflow execution
3. Unified API: Single endpoint handles all datasource types (local_file, online_document, website_crawl, online_drive)
4. Differentiated Responses:
   • For local_file and website_crawl: Returns datasource_info_list ready for execution
   • For online_document and online_drive: Returns document list (datasource_info_list), users need to select documents before execution
5. Automatic Credential Injection: System automatically injects credential_id from datasource node default credential
6. Flexible Input: Supports direct datasource_info_list, file upload, or text input
7. User Controls Execution: Users explicitly call /pipeline/run or PipelineGenerateService.generate() with prepared data

6.2 Critical Limitations

1. Requires Published Workflow: Can only use published workflow, cannot use draft workflow
2. Requires Datasource Node: Pipeline workflow must contain at least one datasource node
3. Data Source Type Detection: Automatically determined from datasource node configuration (uses specified if start_node_id provided, otherwise uses first)
4. Workflow Execution: Users must separately call /pipeline/run endpoint or PipelineGenerateService.generate() to execute workflow

6.3 Error Handling

• If dataset is not a pipeline knowledge base, return clear error message
• If pipeline does not exist or is not published, return appropriate error
• If workflow has no datasource node, return error
• If provided start_node_id is not a datasource node, return error
• If datasource type doesn't match input method (e.g., file upload for online_document), return error
• File upload errors (size, type, etc.) need proper handling
• Missing required fields in datasource_info_list for non-local_file types

6.4 Performance Considerations

• File upload uses streaming read, supports large files
• Data preparation is synchronous and fast (no workflow execution)

6.5 Compatibility

• Maintain consistency with existing Pipeline API patterns (/pipeline/run, /pipeline/datasource/nodes/...)
• Response format provides ready-to-use data structure
• Does not affect existing Vendor Knowledge Base functionality
• Reuses existing /pipeline/run endpoint for workflow execution

7. Implementation Steps

7.1 Create New File

Create api/controllers/service_api/dataset/rag_pipeline/rag_pipeline_document.py

7.2 Implement Endpoints

1. PipelineDocumentCreateApi.post() - Prepare data for document creation (POST, supports JSON and multipart/form-data)
2. PipelineDocumentUpdateApi.patch() - Prepare data for document update (PATCH, supports JSON and multipart/form-data)
3. Helper functions:
   • _get_first_datasource_node_id() - Automatically get first datasource node ID
   • _get_datasource_type_from_node() - Get datasource type from start_node_id
   • _get_default_credential_id() - Get default credential from datasource node
   • _get_online_document_list() - Get document list for online documents (Notion, Google Drive)
   • _build_datasource_info_list() - Build datasource_info_list with automatic credential injection

7.3 Register Endpoints

Import new module in api/controllers/service_api/dataset/rag_pipeline/__init__.py

7.4 Testing

• Test data preparation for all datasource types (local_file, online_document, website_crawl, online_drive)
• Test online_document and online_drive document list retrieval functionality
• Test user workflow of selecting documents from list and executing workflow
• Test file upload (small files, large files with streaming)
• Test text content input
• Test direct datasource_info_list input
• Test automatic credential_id injection
• Test document update data preparation
• Test error scenarios (non-pipeline knowledge base, no datasource node in workflow, invalid start_node_id, file too large, etc.)
• Test workflow execution using prepared data (via /pipeline/run)

[dosubot[bot]]

The proposed Service API endpoints for creating and updating documents (by text or file, with streaming support) in Pipeline Knowledge Bases are not yet implemented in the Dify codebase—they remain in the planning stage and are tracked in issue #31524.

Currently, the available pipeline-related endpoints are /v1/datasets/{dataset_id}/pipeline/run for triggering pipeline workflows and /v1/datasets/pipeline/file-upload for uploading files. However, these do not provide the dedicated document creation/update functionality described in your proposal, and may not reliably trigger the full pipeline (especially for custom extractors or advanced workflows) via API alone. The Web UI remains the most reliable way to ensure the pipeline is executed as intended for document ingestion source.

Authentication for these endpoints typically requires a user-scoped API token with appropriate permissions; using a dataset token will result in errors source.

The technical rationale for the proposed endpoints is sound: separating pipeline document APIs from vendor KB APIs, using RESTful conventions (POST for create, PATCH for update), and ensuring all document operations run through the pipeline workflow (PipelineGenerateService.generate()). This approach will provide clarity, maintainability, and compatibility with existing APIs.

For now, keep an eye on issue #31524 for updates, and use the existing endpoints with the above limitations in mind. If you need guaranteed pipeline execution (especially with custom extractors), use the Web UI until these new endpoints are available.

To reply, just mention @dosu.

---

Docs are dead. Just use Dosu.

[antoine-lasuite]

Any idea as to when the Pipeline will be fully accessible via API endpoint ?