Dify对接钉钉开发者后台添加的机器人

前面给你的版本偏“概念代码”，这里改成 工程化 services 分层结构 。 

 目标： 

 

 

 钉钉 开发者后台机器人 

 

 

 Stream 模式 

 

 

 Python dingtalk-stream 

 

 

 FastAPI 项目结构 

 

 

 Dify Chat API Streaming 

 

 

 Service 层隔离 

 

 

 后续可扩展 Redis、知识库、权限 

 

 

 dingtalk-stream SDK 本身就是针对钉钉 Stream 模式的机器人收消息、事件回调设计的。当前 Python 包版本可直接通过 pip 安装。( PyPI ) 

 

 一、项目结构 

 dify-dingtalk-bot/

├── app/

│

│ ├── main.py

│ │

│ ├── config/

│ │ └── settings.py

│ │

│ ├── services/

│ │ │

│ │ ├── dify_service.py

│ │ │

│ │ ├── dingtalk_service.py

│ │ │

│ │ └── chat_service.py

│ │

│ ├── handlers/

│ │ └── robot_handler.py

│ │

│ └── utils/

│ └── logger.py

│

├── requirements.txt

└── .env

 

 

 二、依赖 

 requirements.txt 

 fastapi

uvicorn

python-dotenv

requests

dingtalk-stream

 

 安装： 

 pip install -r requirements.txt

 

 

 三、配置 

 .env 

 # 钉钉Stream

DING_CLIENT_ID=dingxxxx

DING_CLIENT_SECRET=xxxx

# Dify

DIFY_API_KEY=app-xxxx

DIFY_API_URL=https://api.dify.ai/v1/chat-messages

# 服务

APP_NAME=dify-dingtalk-bot

 

 

 四、配置服务 

 app/config/settings.py 

 import os

from dotenv import load_dotenv

load_dotenv()

class Settings:

 DING_CLIENT_ID = os.getenv(

 "DING_CLIENT_ID"

 )

 DING_CLIENT_SECRET = os.getenv(

 "DING_CLIENT_SECRET"

 )

 DIFY_API_KEY=os.getenv(

 "DIFY_API_KEY"

 )

 DIFY_API_URL=os.getenv(

 "DIFY_API_URL"

 )

settings=Settings()

 

 

 五、Dify Service 

 负责： 

 

 

 调用 Dify 

 

 

 消费 SSE 

 

 

 返回完整答案 

 

 

 app/services/dify_service.py 

 import requests

import json

from app.config.settings import settings

class DifyService:

 def __init__(self):

 self.url = (

 settings.DIFY_API_URL

 )

 self.key = (

 settings.DIFY_API_KEY

 )

 def chat(

 self,

 question:str,

 user:str

 ):

 headers={

 "Authorization":

 f"Bearer {self.key}",

 "Content-Type":

 "application/json"

 }

 payload={

 "inputs":{},

 "query":

 question,

 "response_mode":

 "streaming",

 "user":

 user

 }

 response=requests.post(

 self.url,

 headers=headers,

 json=payload,

 stream=True,

 timeout=120

 )

 answer=""

 for line in response.iter_lines():

 if not line:

 continue

 line=line.decode(

 "utf-8"

 )

 if not line.startswith(

 "data:"

 ):

 continue

 data=json.loads(

 line[5:]

 )

 event=data.get(

 "event"

 )

 if event=="message":

 answer += data.get(

 "answer",

 ""

 )

 if event=="message_end":

 break

 return answer

 

 

 六、聊天业务 Service 

 以后加： 

 

 

 用户权限 

 

 

 Redis 

 

 

 会话ID 

 

 

 日志 

 

 

 都放这里。 

 app/services/chat_service.py 

 from app.services.dify_service import DifyService

class ChatService:

 def __init__(self):

 self.dify=DifyService()

 def ask(

 self,

 message,

 user

 ):

 return self.dify.chat(

 question=message,

 user=user

 )

 

 

 七、钉钉发送 Service 

 app/services/dingtalk_service.py 

 import dingtalk_stream

class DingTalkService:

 def reply(

 self,

 handler,

 text,

 message

 ):

 handler.reply_text(

 text,

 message

 )

 

 

 八、机器人 Handler 

 这是核心。 

 app/handlers/robot_handler.py 

 import logging

import dingtalk_stream

from dingtalk_stream import AckMessage

from app.services.chat_service import ChatService

from app.services.dingtalk_service import DingTalkService

class RobotHandler(

 dingtalk_stream.ChatbotHandler

):

 def __init__(self):

 super().__init__()

 self.chat_service=(

 ChatService()

 )

 self.ding_service=(

 DingTalkService()

 )

 async def process(

 self,

 callback:

 dingtalk_stream.CallbackMessage

 ):

 incoming = (

 dingtalk_stream

 .ChatbotMessage

 .from_dict(

 callback.data

 )

 )

 question=(

 incoming

 .text

 .content

 .strip()

 )

 user=(

 incoming

 .sender_staff_id

 )

 logging.info(

 f"user={user}, msg={question}"

 )

 try:

 answer=(

 self.chat_service.ask(

 question,

 user

 )

 )

 self.ding_service.reply(

 self,

 answer,

 incoming

 )

 except Exception as e:

 self.reply_text(

 "服务异常："+str(e),

 incoming

 )

 return (

 AckMessage.STATUS_OK,

 "OK"

 )

 

 

 九、启动 Stream Client 

 app/main.py 

 import logging

import dingtalk_stream

from app.config.settings import settings

from app.handlers.robot_handler import RobotHandler

def start():

 logging.basicConfig(

 level=logging.INFO

 )

 credential=(

 dingtalk_stream

 .Credential(

 settings.DING_CLIENT_ID,

 settings.DING_CLIENT_SECRET

 )

 )

 client=(

 dingtalk_stream

 .DingTalkStreamClient(

 credential

 )

 )

 client.register_callback_handler(

 dingtalk_stream

 .chatbot

 .ChatbotMessage

 .TOPIC,

 RobotHandler()

 )

 client.start_forever()

if __name__=="__main__":

 start()

 

 

 十、启动 

 python -m app.main

 

 正常日志： 

 INFO endpoint is wss://xxxx

INFO websocket connected

 

 钉钉开发者后台机器人上线后： 

 群里： 

 @机器人 公司的年假政策是什么？

 

 流程： 

 ChatbotMessage

 |

 ↓

RobotHandler

 |

 ↓

ChatService

 |

 ↓

DifyService

 |

 ↓

Dify SSE

 |

 ↓

reply_text()

 |

 ↓

钉钉群

 

 

 十一、下一步建议改造（生产版） 

 你这个架构建议继续升级： 

 1. 增加 Redis 会话 

 现在： 

 用户

 |

Dify

 |

新会话

 

 升级： 

 staff_id

 |

Redis

 |

conversation_id

 |

Dify上下文

 

 

 2. 增加真正 Stream 卡片回复 

 当前： 

 Dify:

chunk1

chunk2

chunk3

钉钉:

一次回复

 

 升级： 

 Dify chunk

 |

 |

钉钉AI卡片stream

 |

 |

逐字刷新

 

 可以做到类似 ChatGPT 的效果。 

 

 3. 推荐生产目录 

 大型项目： 

 services

├── dingtalk

│ ├── stream_client.py

│ ├── message.py

│ └── card.py

│

├── dify

│ ├── client.py

│ ├── stream.py

│ └── workflow.py

│

├── memory

│ └── redis.py

│

└── security

 └── permission.py

 

 这个结构可以直接扩展成企业级 钉钉 AI 助手 + Dify RAG 知识库平台 。 

 现在信息完整了，可以确定有 两个独立问题 ： 

 

 

 .env 没有生效（最关键） 

 

 

 websockets 版本和 dingtalk-stream 0.24.3 不兼容 

 

 

 我们逐个修。 

 

 一、你的 .env 当前全部被注释了 

 你贴出来： 

 # DING_CLIENT_ID=ding09cjpitrovdwlax0

# DING_CLIENT_SECRET=xxxxx

 

 注意前面的： 

 #

 

 意味着 Python 根本读不到。 

 所以： 

 settings.DING_CLIENT_ID

 

 实际应该是： 

 None

 

 SDK 发起： 

 POST https://api.dingtalk.com/v1.0/gateway/connections/open

 

 时，没有合法 credential，钉钉返回： 

 {

 "code":"systemError"

}

 

 

 修改 .env 

 去掉所有 # ： 

 vim /opt/dify-dingtalk-bot/.env

 

 改成： 

 # 钉钉Stream

DING_CLIENT_ID=xxxxxxxxxxxxx

DING_CLIENT_SECRET=你的secret

# Dify

DIFY_API_KEY=app-xxxxxxxxxxxxxxxxxxxxx

DIFY_API_URL=http://192.168.4.20/v1/chat-messages

APP_NAME=dify-dingtalk-bot

 

 注意： 

 你的 Dify 地址之前： 

 DIFY_API_URL=http://192.168.4.20/v1

 

 少了： 

 /chat-messages

 

 应该： 

 http://192.168.4.20/v1/chat-messages

 

 

 二、验证环境变量 

 不要启动机器人，先测试： 

 python

 

 执行： 

 from app.config.settings import settings

print("client:", settings.DING_CLIENT_ID)

print("secret:",

 bool(settings.DING_CLIENT_SECRET))

print("dify:",

 settings.DIFY_API_URL)

 

 正确应该： 

 client: xxxxxxxxxxxx

secret: True

dify: http://192.168.4.20/v1/chat-messages

 

 如果还是： 

 client: None

 

 说明 dotenv 路径问题。 

 

 三、修复 websockets 兼容问题 

 你的错误： 

 AttributeError:

module 'websockets' has no attribute 'exceptions'

 

 原因： 

 你现在： 

 dingtalk-stream 0.24.3

 

 依赖： 

 websockets

 

 但是装到了新版： 

 大概率： 

 websockets 15.x

 

 新版删除了一些旧接口。 

 

 查看： 

 pip show websockets

 

 应该类似： 

 Version: 15.x

 

 

 降级： 

 pip uninstall websockets -y

pip install websockets==10.4

 

 然后确认： 

 pip show websockets

 

 应该： 

 Version: 10.4

 

 

 四、修改 requirements.txt 

 否则以后重新部署还会坏。 

 你的： 

 app/requirements.txt 

 改： 

 fastapi

uvicorn

python-dotenv

requests

dingtalk-stream==0.24.3

websockets==10.4

 

 

 五、重新测试 

 执行： 

 cd /opt/dify-dingtalk-bot

source /opt/ding_dify/bin/activate

python -m app.main

 

 正常应该看到： 

 类似： 

 INFO:dingtalk_stream.client:

open connection

INFO:dingtalk_stream.client:

connection established

INFO:dingtalk_stream.client:

websocket connected

 

 而不是循环： 

 400 systemError

 

 

 六、如果仍然 400 

 那就只剩钉钉后台配置。 

 重点检查： 

 1. Client ID 

 必须来自： 

 钉钉开放平台： 

 应用开发

 ↓

企业内部应用

 ↓

凭证与基础信息

 ↓

Client ID

 

 不是： 

 

 

 AppKey 

 

 

 AgentId 

 

 

 RobotCode 

 

 

 

 2. Stream模式开启 

 后台： 

 机器人管理

 ↓

消息接收模式

 

 必须： 

 Stream模式

 

 

 3. 发布应用 

 必须： 

 版本管理与发布

 ↓

上线

 

 测试应用也需要发布。 

 

 4. 机器人权限 

 至少： 

 机器人消息接收

机器人发送消息

 

 

 你现在先做两个动作： 

 ① 取消 .env 注释 

 ② 降级 websocket 

 执行： 

 pip uninstall websockets -y

pip install websockets==10.4

 

 然后重新跑。 

 如果还有错误，把新的启动日志贴出来。你现在距离通已经很近了。 

   

 不对。你现在加日志的位置有问题，会导致程序启动时直接报错。 

 你写在文件顶部： 

 logger.info(

 "Dify request question=%s user=%s",

 question,

 user

)

logger.info(

 "Dify response=%s",

 answer

)

 

 但是此时： 

 question

user

answer

 

 三个变量都还不存在。 

 Python 加载 dify_service.py 时就会执行这里，所以会报： 

 NameError: name 'question' is not defined

 

 

 正确方式： 日志应该放到 chat() 方法内部。 

 修改后的 dify_service.py ： 

 import requests

import json

import logging

from app.config.settings import settings

logger = logging.getLogger(__name__)

class DifyService:

 def __init__(self):

 self.url = settings.DIFY_API_URL

 self.key = settings.DIFY_API_KEY

 def chat(

 self,

 question: str,

 user: str

 ):

 logger.info(

 "Dify request question=%s user=%s",

 question,

 user

 )

 headers = {

 "Authorization":

 f"Bearer {self.key}",

 "Content-Type":

 "application/json"

 }

 payload = {

 "inputs": {},

 "query": question,

 "response_mode":

 "streaming",

 "conversation_id":"",

 "user": user

 }

 logger.info(

 "Dify url=%s",

 self.url

 )

 response = requests.post(

 self.url,

 headers=headers,

 json=payload,

 stream=True,

 timeout=120

 )

 logger.info(

 "Dify http status=%s",

 response.status_code

 )

 answer = ""

 for line in response.iter_lines():

 if not line:

 continue

 line = line.decode(

 "utf-8"

 )

 logger.debug(

 "Dify stream=%s",

 line

 )

 if not line.startswith(

 "data:"

 ):

 continue

 data=json.loads(

 line[5:]

 )

 event=data.get(

 "event"

 )

 if event=="message":

 chunk=data.get(

 "answer",

 ""

 )

 answer += chunk

 logger.info(

 "Dify chunk=%s",

 chunk

 )

 elif event=="message_end":

 break

 elif event=="error":

 logger.error(

 "Dify error=%s",

 data

 )

 logger.info(

 "Dify response=%s",

 answer

 )

 return answer

 

 

 另外，你现在排查钉钉无回答，建议再加一个关键日志 

 在： 

 response=requests.post(...)

 

 后面增加： 

 logger.info(

 "Dify headers=%s",

 response.headers

)

 

 以及： 

 if response.status_code != 200:

 logger.error(

 "Dify error body=%s",

 response.text

 )

 

 否则如果 API Key 错误，你只看到： 

 answer=""

 

 不知道原因。 

 

 重启服务 

 如果你用 systemd： 

 systemctl restart dify-dingtalk

 

 查看： 

 journalctl -u dify-dingtalk -f

 

 然后钉钉发送问题。 

 你应该看到类似： 

 Dify request question=公司报销流程是什么 user=xxx

Dify url=http://192.168.4.20/v1/chat-messages

Dify http status=200

Dify chunk=根据公司制度...

Dify response=根据公司制度...

 

 

 如果日志显示： 

 Dify http status=200

Dify response=

 

 那就是 Dify 应用配置问题 。 

 如果显示： 

 401

 

 就是 API Key。 

 如果显示： 

 404

 

 就是 API URL。 

 如果显示： 

 200 有answer

 

 但是钉钉没回复，就是 dingtalk_service.py / handler 的问题。你下一步把日志贴出来即可。