Dify
openai 代理
import time
import requests
from fastapi import FastAPI, Request
DIFY_API_KEY = "xxx"
DIFY_API_URL = "http://xxx/v1/chat-messages"
app = FastAPI()
@app.post("/v1/chat/completions")
async def chat(req: Request):
data = await req.json()
messages = data.get("messages", [])
user_input = messages[-1]["content"] if messages else ""
r = requests.post(
DIFY_API_URL,
headers={
"Authorization": f"Bearer {DIFY_API_KEY}",
"Content-Type": "application/json",
},
json={
"inputs": {},
"query": user_input,
"response_mode": "blocking",
"user": "openai-proxy",
},
)
result = r.json()
answer = result.get("answer", "")
return {
"id": "chatcmpl-" + str(int(time.time())),
"object": "chat.completion",
"created": int(time.time()),
"model": data.get("model", "dify"),
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": answer,
},
"finish_reason": "stop",
}
],
}
.env
# Dify 应用的 API 地址 (通常是 http://<你的Dify域名>/v1/chat-messages)
DIFY_API_URL=http://xxx/v1/chat-messages
# Dify 应用的 API Key (在 Dify 控制台 -> 访问 API -> 密钥管理中获取)
DIFY_API_KEY=
# 默认用户 ID (用于标识对话发起者)
DEFAULT_USER_ID=user-123
uvicorn main:app --port 8000
http://localhost:8000/v1
安装
Warning
需要科学上网
git clone https://github.com/langgenius/dify.git
cd dify
cd docker
cp .env.example .env
docker compose up -d
use
分段模式
- 分段标识符：系统将在文本出现分段标识符时自动执行分段。默认值为 \n\n，即按照文章段落进行分块
对接xinference
- 安装xinference插件
对接firecrawl
Q&A
注册账户没权限
root@ai:/home/wjn/dify/volumes# docker exec -it dify-api-1 id uid=1001(dify) gid=1001(dify) groups=1001(dify) root@ai:/home/wjn/dify/volumes# sudo chown -R 1001:1001 ./volumes/app/storage
embedding
Xorbits Inference
jina-embeddings-v2-base-zh
mcp
Dify Agent 策略 MCP Agent 策略 Agent 策略（支持 MCP 工具） MCP SSE
小红书 dag
mcp
- chromedriver
- xhs-mcp-server
$env:phone="18636192514" $env:json_path="C://wjn/tmp/" uvx --from xhs_mcp_server@latest login
- https://modelscope.cn/headlines/article/1139
https://pptr.nodejs.cn/
api
import requests
# 基础 URL 和 API 密钥
base_url = "http://localhost/v1"
api_key = "app-lS80BymnOSfajBGSzf8B4ex9"
# 请求头
headers = {
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
}
# URL 定义
url_run = f"{base_url}/workflows/run"
def run_workflow(q:str):
"""
运行工作流
:param q:
:return:
"""
# 请求体
data = {
'inputs': {
"q": q
},
'response_mode': 'blocking',
'user': 'abc-123'
}
# 发送 POST 请求
response = requests.post(url_run, headers=headers, json=data)
# 打印响应结果
if response.status_code == 200:
r = response.json()
print(r)
c = r['data']['outputs']['text']
c = c.replace("```html", "").replace("```", "")
return c
else:
print(f"请求失败: {response.status_code}, {response.text}")