# dify-workflow.yaml — a production RAG + agent pipeline
app:
  name: "customer-support-agent"
  mode: "workflow"
  version: "1.14.2"

nodes:
  - id: "start"
    type: "start"
    variables:
      - name: "user_query"
        type: "string"
        required: true

  - id: "retriever"
    type: "knowledge-retrieval"
    dataset_ids: ["faq-dataset-v3"]
    top_k: 5
    score_threshold: 0.7
    depends_on: ["start"]

  - id: "llm-agent"
    type: "llm"
    model: "gpt-4o"
    prompt_template: |
      Context: {{ retriever.documents }}
      Question: {{ start.user_query }}
      Answer concisely using only the context above.
    depends_on: ["retriever"]

  - id: "output"
    type: "end"
    output: "{{ llm-agent.text }}"
    depends_on: ["llm-agent"]

tracing:
  enabled: true
  backend: "langfuse"  # or opik, arize-phoenix
  sample_rate: 1.0

# dify_model_config.py — Configure multi-model routing via Dify API
import requests

DIFY_API_KEY = "your-api-key"
DIFY_BASE = "https://your-dify-instance.com/v1"

def configure_model_fallback():
    """Set up a 3-tier model fallback chain for production resilience."""

    # Primary: GPT-4o for high-quality reasoning
    # Fallback 1: Claude 3.5 Sonnet (different provider, same tier)
    # Fallback 2: GPT-4o-mini (cheap, fast, good enough for simple steps)
    config = {
        "model": "gpt-4o",
        "provider": "openai",
        "fallback_chain": [
            {
                "model": "claude-3-5-sonnet-20241022",
                "provider": "anthropic",
                "trigger": "rate_limit_error"  # switch on 429
            },
            {
                "model": "gpt-4o-mini",
                "provider": "openai",
                "trigger": "any_error",  # last resort
                "max_retries": 2
            }
        ],
        "timeout_seconds": 30,
        "retry_policy": {
            "max_retries": 3,
            "backoff_multiplier": 2.0
        }
    }

    resp = requests.post(
        f"{DIFY_BASE}/models/configure",
        headers={"Authorization": f"Bearer {DIFY_API_KEY}"},
        json=config,
        timeout=15
    )
    return resp.json()

# Usage: call this during deployment to ensure resilience
result = configure_model_fallback()
print(f"Model config applied: {result.get('status')}")

# dify_rag_config.py — Configure advanced RAG with hybrid search
import requests

DIFY_API_KEY = "your-api-key"
DIFY_BASE = "https://your-dify-instance.com/v1"

def create_optimized_dataset(name: str, chunking_strategy: str = "markdown_header"):
    """Create a knowledge base with production-grade retrieval settings."""

    # Step 1: Create the dataset with custom chunking
    dataset_config = {
        "name": name,
        "description": "Production knowledge base with hybrid search",
        "indexing_technique": "high_quality",  # uses embedding model
        "chunk_setting": {
            "chunk_size": 512,
            "chunk_overlap": 64,
            "separator": "\n\n",  # split on double newlines
            "chunking_strategy": chunking_strategy  # or "recursive", "token"
        },
        "retrieval_model": {
            "search_method": "hybrid",  # vector + keyword BM25
            "reranking_enable": True,
            "reranking_model": {
                "reranking_provider_name": "cohere",
                "reranking_model_name": "rerank-english-v3.0"
            },
            "top_k": 5,
            "score_threshold": 0.6,  # filter low-relevance chunks
            "score_threshold_enabled": True
        }
    }

    resp = requests.post(
        f"{DIFY_BASE}/datasets",
        headers={"Authorization": f"Bearer {DIFY_API_KEY}"},
        json=dataset_config,
        timeout=30
    )
    dataset_id = resp.json().get("id")
    print(f"Dataset created: {dataset_id}")
    return dataset_id

# Step 2: Upload documents with custom processing
def upload_and_index(dataset_id: str, file_path: str):
    with open(file_path, "rb") as f:
        resp = requests.post(
            f"{DIFY_BASE}/datasets/{dataset_id}/documents/upload",
            headers={"Authorization": f"Bearer {DIFY_API_KEY}"},
            files={"file": (file_path, f, "application/pdf")},
            timeout=60
        )
    return resp.json()

# Usage
ds_id = create_optimized_dataset("engineering-docs", chunking_strategy="markdown_header")
upload_and_index(ds_id, "./api-reference.pdf")

# dify_custom_tool.py — Register a custom tool for Dify agent
import requests
import json

DIFY_API_KEY = "your-api-key"
DIFY_BASE = "https://your-dify-instance.com/v1"

def register_custom_tool():
    """Register an internal API as a Dify agent tool."""

    tool_def = {
        "name": "query_inventory",
        "description": "Query product inventory levels by SKU code. Returns stock count, warehouse location, and restock date.",
        "method": "get",
        "url": "https://api.internal.company.com/v1/inventory",
        "headers": {
            "Authorization": "Bearer ${INVENTORY_API_TOKEN}",
            "Content-Type": "application/json"
        },
        "parameters": {
            "type": "object",
            "properties": {
                "sku": {
                    "type": "string",
                    "description": "Product SKU code (e.g., 'WID-001-2026')"
                },
                "warehouse": {
                    "type": "string",
                    "description": "Optional warehouse ID. If omitted, checks all warehouses.",
                    "required": False
                }
            },
            "required": ["sku"]
        }
    }

    resp = requests.post(
        f"{DIFY_BASE}/tools",
        headers={"Authorization": f"Bearer {DIFY_API_KEY}"},
        json=tool_def,
        timeout=15
    )
    return resp.json()

# Connect an MCP server (e.g., a database MCP)
def connect_mcp_server():
    """Connect an MCP server to extend agent capabilities."""
    mcp_config = {
        "name": "postgres-mcp",
        "type": "mcp_server",
        "transport": "stdio",
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-postgres"],
        "env": {
            "DATABASE_URL": "${DATABASE_URL}"
        }
    }

    resp = requests.post(
        f"{DIFY_BASE}/mcp/servers",
        headers={"Authorization": f"Bearer {DIFY_API_KEY}"},
        json=mcp_config,
        timeout=15
    )
    return resp.json()

# Usage
tool = register_custom_tool()
mcp = connect_mcp_server()
print(f"Tool registered: {tool.get('name')}, MCP server: {mcp.get('name')}")

# dify_baas.py — Use Dify as a backend AI service
import requests
import json

DIFY_API_KEY = "your-api-key"
DIFY_BASE = "https://your-dify-instance.com/v1"

class DifyClient:
    """Production client for Dify Backend-as-a-Service."""

    def __init__(self, api_key: str, base_url: str):
        self.api_key = api_key
        self.base_url = base_url
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def run_workflow(self, workflow_id: str, inputs: dict) -> dict:
        """Execute a workflow synchronously and return the output."""
        resp = requests.post(
            f"{self.base_url}/workflows/{workflow_id}/run",
            headers=self.headers,
            json={"inputs": inputs, "response_mode": "blocking"},
            timeout=120
        )
        return resp.json()

    def chat(self, app_id: str, query: str, user_id: str,
             conversation_id: str = None) -> dict:
        """Send a message to a chatbot/agent app."""
        payload = {
            "inputs": {},
            "query": query,
            "user": user_id,
            "response_mode": "blocking"
        }
        if conversation_id:
            payload["conversation_id"] = conversation_id

        resp = requests.post(
            f"{self.base_url}/chat-messages",
            headers=self.headers,
            json=payload,
            timeout=60
        )
        return resp.json()

    def stream_chat(self, app_id: str, query: str, user_id: str):
        """Stream a chat response for real-time UI updates."""
        payload = {
            "inputs": {},
            "query": query,
            "user": user_id,
            "response_mode": "streaming"
        }
        resp = requests.post(
            f"{self.base_url}/chat-messages",
            headers=self.headers,
            json=payload,
            stream=True,
            timeout=120
        )
        for line in resp.iter_lines():
            if line and line.startswith(b"data:"):
                yield json.loads(line[5:])

# Usage: embed Dify in your existing app
client = DifyClient(DIFY_API_KEY, DIFY_BASE)

# Run a workflow from your backend
result = client.run_workflow(
    workflow_id="wf-abc123",
    inputs={"user_query": "How do I reset my password?", "user_tier": "enterprise"}
)
print(f"Workflow output: {result.get('data', {}).get('outputs', {})}")

# Chat with an agent from your frontend
response = client.chat(
    app_id="agent-xyz789",
    query="What's the status of order #12345?",
    user_id="user-42"
)
print(f"Agent reply: {response.get('answer')}")