Dify — 低代码 LLM 应用平台

简介

Dify 是开源的 LLM 应用开发平台，提供可视化工作流编排、RAG 知识库、Agent 构建等能力，支持私有化部署，适合快速构建金融 AI 应用原型。

Docker 部署

git clone https://github.com/langgenius/dify.git
cd dify/docker
cp .env.example .env
docker compose up -d
# 访问 http://localhost/

API 调用

import requests

DIFY_API_KEY = "app-xxx"
DIFY_BASE_URL = "http://localhost/v1"

# 对话接口
def chat_with_dify(query: str, conversation_id: str = None, user: str = "user"):
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "inputs": {},
        "query": query,
        "response_mode": "blocking",  # blocking | streaming
        "user": user
    }
    
    if conversation_id:
        payload["conversation_id"] = conversation_id
    
    response = requests.post(
        f"{DIFY_BASE_URL}/chat-messages",
        headers=headers,
        json=payload
    )
    
    data = response.json()
    return {
        "answer": data["answer"],
        "conversation_id": data["conversation_id"],
        "tokens": data.get("metadata", {}).get("usage", {})
    }

# 使用
result = chat_with_dify("分析当前信贷市场风险")
print(result["answer"])

# 继续对话
result2 = chat_with_dify(
    "具体说说小微企业贷款的风险",
    conversation_id=result["conversation_id"]
)

流式输出

import json

def stream_chat(query: str, user: str = "user"):
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "inputs": {},
        "query": query,
        "response_mode": "streaming",
        "user": user
    }
    
    with requests.post(
        f"{DIFY_BASE_URL}/chat-messages",
        headers=headers,
        json=payload,
        stream=True
    ) as response:
        for line in response.iter_lines():
            if line and line.startswith(b"data: "):
                data = json.loads(line[6:])
                if data.get("event") == "message":
                    print(data["answer"], end="", flush=True)
                elif data.get("event") == "message_end":
                    print()  # 换行
                    break

stream_chat("详细解释 RAG 技术在金融领域的应用")

知识库 API

# 上传文档到知识库
def upload_document(dataset_id: str, file_path: str):
    headers = {"Authorization": f"Bearer {DIFY_API_KEY}"}
    
    with open(file_path, "rb") as f:
        files = {"file": (file_path, f, "text/plain")}
        data = {
            "indexing_technique": "high_quality",
            "process_rule": json.dumps({
                "mode": "automatic"
            })
        }
        
        response = requests.post(
            f"{DIFY_BASE_URL}/datasets/{dataset_id}/document/create_by_file",
            headers=headers,
            files=files,
            data=data
        )
    
    return response.json()

# 查询知识库
def query_knowledge(dataset_id: str, query: str, top_k: int = 5):
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }
    
    response = requests.post(
        f"{DIFY_BASE_URL}/datasets/{dataset_id}/retrieve",
        headers=headers,
        json={
            "query": query,
            "retrieval_model": {
                "search_method": "hybrid_search",
                "top_k": top_k,
                "score_threshold": 0.5
            }
        }
    )
    
    return response.json()["records"]

Workflow API

# 调用 Dify Workflow（适合批处理场景）
def run_workflow(inputs: dict, user: str = "batch-processor"):
    headers = {
        "Authorization": f"Bearer {DIFY_API_KEY}",
        "Content-Type": "application/json"
    }
    
    response = requests.post(
        f"{DIFY_BASE_URL}/workflows/run",
        headers=headers,
        json={
            "inputs": inputs,
            "response_mode": "blocking",
            "user": user
        }
    )
    
    data = response.json()
    return data["data"]["outputs"]

# 批量处理贷款申请
applications = [
    {"company": "A公司", "revenue": "500万", "debt_ratio": "60%"},
    {"company": "B公司", "revenue": "200万", "debt_ratio": "80%"},
]

for app_data in applications:
    result = run_workflow(app_data)
    print(f"{app_data['company']}: {result}")

Dify 适用场景

• 快速原型：可视化拖拽构建 Agent，无需写代码
• 知识库问答：内置 RAG，上传 PDF/Word 即可问答
• 工作流自动化：多步骤审批、报告生成
• 私有化部署：金融数据不出内网