Плагин источника данных

​

Предварительные требования

• Шаг 2. Оркестрация конвейера знаний
• Разработка плагинов Интеллос: руководство Hello World

​

Типы плагинов источников данных

• Веб-краулер: Jina Reader, FireCrawl
• Онлайн-документ: Notion, Confluence, GitHub
• Онлайн-диск: OneDrive, Google Drive, Box, AWS S3, Tencent COS

​

Разработка плагина источника данных

​

Создание плагина источника данных

dify plugin init

​

Структура плагина источника данных

• Файл manifest.yaml: описывает основную информацию о плагине.
• Каталог provider: содержит описание поставщика плагина и код реализации аутентификации.
• Каталог datasources: содержит описание и основную логику получения данных из источника данных.

├── _assets
│   └── icon.svg
├── datasources
│   ├── your_datasource.py
│   └── your_datasource.yaml
├── main.py
├── manifest.yaml
├── PRIVACY.md
├── provider
│   ├── your_datasource.py
│   └── your_datasource.yaml
├── README.md
└── requirements.txt

​

Установка правильной версии и тега

• В файле manifest.yaml задайте минимальную поддерживаемую версию Интеллос следующим образом:
  minimum_dify_version: 1.9.0
  
• В файле manifest.yaml добавьте следующий тег, чтобы плагин отображался в категории источников данных в Интеллос Маркетплейс:
  tags:
    - rag
  
• В файле requirements.txt задайте версию SDK плагина, используемую для разработки плагина источника данных, следующим образом:
  dify-plugin>=0.5.0,<0.6.0
  

​

Добавление поставщика источника данных

​

Создание YAML-файла поставщика

# Specify the provider type for the data source plugin: online_drive, online_document, or website_crawl
provider_type: online_drive # online_document, website_crawl

# Specify data sources
datasources:
  - datasources/PluginName.yaml

​

Создание файла кода поставщика

• При использовании режима аутентификации по API Key файл кода поставщика для плагинов источников данных идентичен файлу для плагинов-инструментов. Нужно лишь изменить родительский класс, наследуемый классом поставщика, на DatasourceProvider.
  class YourDatasourceProvider(DatasourceProvider):
  
      def _validate_credentials(self, credentials: Mapping[str, Any]) -> None:
          try:
              """
              IMPLEMENT YOUR VALIDATION HERE
              """
          except Exception as e:
              raise ToolProviderCredentialValidationError(str(e))
  
• При использовании режима аутентификации OAuth плагины источников данных немного отличаются от плагинов-инструментов. При получении прав доступа через OAuth плагины источников данных могут одновременно возвращать имя пользователя и аватар для отображения на фронтенде. Поэтому методы _oauth_get_credentials и _oauth_refresh_credentials должны возвращать тип DatasourceOAuthCredentials, который содержит name, avatar_url, expires_at и credentials. Класс DatasourceOAuthCredentials определён следующим образом и при возврате должен быть приведён к соответствующему типу:
  class DatasourceOAuthCredentials(BaseModel):
      name: str | None = Field(None, description="The name of the OAuth credential")
      avatar_url: str | None = Field(None, description="The avatar url of the OAuth")
      credentials: Mapping[str, Any] = Field(..., description="The credentials of the OAuth")
      expires_at: int | None = Field(
          default=-1,
          description="""The expiration timestamp (in seconds since Unix epoch, UTC) of the credentials.
          Set to -1 or None if the credentials do not expire.""",
      )
  

def _oauth_get_authorization_url(self, redirect_uri: str, system_credentials: Mapping[str, Any]) -> str:
"""
Generate the authorization URL for {{ .PluginName }} OAuth.
"""
try:
    """
    IMPLEMENT YOUR AUTHORIZATION URL GENERATION HERE
    """
except Exception as e:
    raise DatasourceOAuthError(str(e))
return ""

def _oauth_get_credentials(
self, redirect_uri: str, system_credentials: Mapping[str, Any], request: Request
) -> DatasourceOAuthCredentials:
"""
Exchange code for access_token.
"""
try:
    """
    IMPLEMENT YOUR CREDENTIALS EXCHANGE HERE
    """
except Exception as e:
    raise DatasourceOAuthError(str(e))
return DatasourceOAuthCredentials(
    name="",
    avatar_url="",
    expires_at=-1,
    credentials={},
)

def _oauth_refresh_credentials(
self, redirect_uri: str, system_credentials: Mapping[str, Any], credentials: Mapping[str, Any]
) -> DatasourceOAuthCredentials:
"""
Refresh the credentials
"""
return DatasourceOAuthCredentials(
    name="",
    avatar_url="",
    expires_at=-1,
    credentials={},
)

​

Добавление источника данных

​

Веб-краулер

output_schema:
    type: object
    properties:
      source_url:
        type: string
        description: the source url of the website
      content:
        type: string
        description: the content from the website
      title:
        type: string
        description: the title of the website
      "description":
        type: string
        description: the description of the website

class YourDataSource(WebsiteCrawlDatasource):

    def _get_website_crawl(
        self, datasource_parameters: dict[str, Any]
    ) -> Generator[ToolInvokeMessage, None, None]:

        crawl_res = WebSiteInfo(web_info_list=[], status="", total=0, completed=0)
        crawl_res.status = "processing"
        yield self.create_crawl_message(crawl_res)
        
        ### your crawl logic
           ...
        crawl_res.status = "completed"
        crawl_res.web_info_list = [
            WebSiteInfoDetail(
                title="",
                source_url="",
                description="",
                content="",
            )
        ]
        crawl_res.total = 1
        crawl_res.completed = 1

        yield self.create_crawl_message(crawl_res)

​

Онлайн-документ

output_schema:
    type: object
    properties:
      workspace_id:
        type: string
        description: workspace id
      page_id:
        type: string
        description: page id
      content:
        type: string
        description: page content

def _get_pages(self, datasource_parameters: dict[str, Any]) -> DatasourceGetPagesResponse:
    # your get pages logic
    response = requests.get(url, headers=headers, params=params, timeout=30)
    pages = []
    for item in  response.json().get("results", []):
        page = OnlineDocumentPage(
            page_name=item.get("title", ""),
            page_id=item.get("id", ""),
            type="page",  
            last_edited_time=item.get("version", {}).get("createdAt", ""),
            parent_id=item.get("parentId", ""),
            page_icon=None, 
        )
        pages.append(page)
    online_document_info = OnlineDocumentInfo(
        workspace_name=workspace_name,
        workspace_icon=workspace_icon,
        workspace_id=workspace_id,
        pages=[page],
        total=pages.length(),
    )
    return DatasourceGetPagesResponse(result=[online_document_info])

def _get_content(self, page: GetOnlineDocumentPageContentRequest) -> Generator[DatasourceMessage, None, None]:
# your fetch content logic, example
response = requests.get(url, headers=headers, params=params, timeout=30)
...
yield self.create_variable_message("content", "")
yield self.create_variable_message("page_id", "")
yield self.create_variable_message("workspace_id", "")

​

Онлайн-диск

output_schema:
    type: object
    properties:
      file:
        $ref: "https://app.intellos.ru/schemas/v1/file.json"

def _browse_files(
self, request: OnlineDriveBrowseFilesRequest
) -> OnlineDriveBrowseFilesResponse:

credentials = self.runtime.credentials
bucket_name = request.bucket
prefix = request.prefix or ""  # Allow empty prefix for root folder; When you browse the folder, the prefix is the folder id
max_keys = request.max_keys or 10
next_page_parameters = request.next_page_parameters or {}

files = []
files.append(OnlineDriveFile(
    id="", 
    name="", 
    size=0, 
    type="folder" # or "file"
))

return OnlineDriveBrowseFilesResponse(result=[
    OnlineDriveFileBucket(
        bucket="", 
        files=files, 
        is_truncated=False, 
        next_page_parameters={}
    )
])

def _download_file(self, request: OnlineDriveDownloadFileRequest) -> Generator[DatasourceMessage, None, None]:
credentials = self.runtime.credentials
file_id = request.id

file_content = bytes()
file_name = ""

mime_type = self._get_mime_type_from_filename(file_name)

yield self.create_blob_message(file_content, meta={
    "file_name": file_name,
    "mime_type": mime_type
})

def _get_mime_type_from_filename(self, filename: str) -> str:
"""Determine MIME type from file extension."""
import mimetypes
mime_type, _ = mimetypes.guess_type(filename)
return mime_type or "application/octet-stream"

• prefix: представляет префикс пути к файлу. Например, prefix=container1/folder1/ извлекает файлы или список файлов из папки folder1 в бакете container1.
• bucket: представляет файловый бакет. Например, bucket=container1 извлекает файлы или список файлов в бакете container1. Это поле можно оставить пустым для дисков с нестандартным протоколом S3.
• id: поскольку метод _download_file не использует переменную prefix, полный путь к файлу должен быть включён в id. Например, id=container1/folder1/file1.txt означает извлечение файла file1.txt из папки folder1 в бакете container1.

​

Отладка плагина

• Если плагин использует аутентификацию OAuth, redirect_uri для удалённой отладки отличается от такового для локального плагина. Обновите соответствующую конфигурацию в OAuth-приложении вашего поставщика сервиса.
• Хотя плагины источников данных поддерживают пошаговую отладку, мы всё же рекомендуем тестировать их в полном конвейере знаний, чтобы убедиться в полной работоспособности.

​

Финальные проверки

• Задали минимальную поддерживаемую версию Интеллос как 1.9.0.
• Задали версию SDK как dify-plugin>=0.5.0,<0.6.0.
• Написали файлы README.md и PRIVACY.md.
• Включили в файлы кода только англоязычный контент.
• Заменили иконку по умолчанию на логотип поставщика источника данных.

​

Упаковка и публикация

dify plugin package . -o your_datasource.difypkg

• Импортировать и использовать плагин в вашей среде Интеллос.
• Опубликовать плагин в Интеллос Маркетплейс, отправив pull request.