Dify 完整教程

# Python工具示例
from typing import Any

class WeatherTool:
    """天气查询工具"""
    
    @property
    def name(self) -> str:
        return "get_weather"
    
    @property
    def description(self) -> str:
        return "获取指定城市的天气信息"
    
    @property
    def parameters(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称，如：北京"
                }
            },
            "required": ["city"]
        }
    
    def _run(self, city: str) -> str:
        """执行工具"""
        # 调用天气API
        weather_data = fetch_weather(city)
        
        return f"{city}今天{weather_data['weather']}，温度{weather_data['temp']}℃"

# 注册工具到Dify
# 在Dify后台"工具"页面添加

# api/core/model_runtime/model_providers/my_llm/my_llm.py
from core.model_runtime.entities.llm_entities import LLMResult
from core.model_runtime.model_providers import ModelProvider

class MyLLMProvider(ModelProvider):
    def _invoke(self, model: str, credentials: dict, 
                prompt_messages: list, **kwargs) -> LLMResult:
        """调用自定义LLM API"""
        
        # 1. 构建请求
        api_url = credentials['api_endpoint']
        api_key = credentials['api_key']
        
        payload = {
            "model": model,
            "messages": prompt_messages,
            "temperature": kwargs.get('temperature', 0.7)
        }
        
        # 2. 发送请求
        response = requests.post(
            api_url,
            headers={"Authorization": f"Bearer {api_key}"},
            json=payload
        )
        
        # 3. 解析响应
        result = response.json()
        
        return LLMResult(
            message=result['choices'][0]['message'],
            usage=result['usage']
        )

# 配置文件：my_llm.yaml
provider: my_llm
label:
  zh_Hans: 我的LLM
  en_US: My LLM
supported_model_types:
  - llm
configurate_methods:
  - predefined-model
provider_credential_schema:
  credential_form_schemas:
    - variable: api_key
      type: secret-input
      required: true
      label:
        zh_Hans: API Key

# api/core/tools/provider/my_tools/tools/search.py
from core.tools.entities.tool_entities import ToolInvokeMessage
from core.tools.tool.builtin_tool import BuiltinTool

class SearchTool(BuiltinTool):
    def _invoke(self, user_id: str, tool_parameters: dict) -> list[ToolInvokeMessage]:
        """
        工具执行逻辑
        
        Args:
            user_id: 用户ID
            tool_parameters: 工具参数
        
        Returns:
            执行结果消息列表
        """
        query = tool_parameters.get('query', '')
        
        # 执行搜索
        results = self._search(query)
        
        # 格式化结果
        formatted = "\n".join([
            f"{i+1}. {r['title']}: {r['snippet']}"
            for i, r in enumerate(results[:5])
        ])
        
        return [
            self.create_text_message(formatted)
        ]
    
    def _search(self, query: str) -> list:
        """实际搜索逻辑"""
        # 调用搜索API
        pass

# 工具配置：search.yaml
identity:
  name: web_search
  author: Your Name
  label:
    zh_Hans: 网页搜索
    en_US: Web Search
description:
  zh_Hans: 在互联网上搜索信息
  en_US: Search information on the web
parameters:
  - name: query
    type: string
    required: true
    label:
      zh_Hans: 搜索关键词
    human_description:
      zh_Hans: 要搜索的内容