Chatwoot + Dify knowledge based Agent bot

[Hanhan0831]

Hello everyone! I'm excited to share with you a project I recently completed -- an automation system that seamlessly integrates the Chatwoot customer service platform with the Dify AI assistant.
At the heart of this project is an asynchronous Python script that implements the following features:

• Monitor open conversations on the Chatwoot platform in real-time.
• Automatically detect unanswered customer messages.
• Generate personalized responses using Dify AI.
• Automatically send AI-generated replies back to Chatwoot conversations.
• Track all conversations and messages in a local database.

The benefits of this system are manifold:

• Improved responsiveness: the AI assistant is able to work 24/7, significantly reducing the time customers have to wait for a response.
• Improve customer service quality: AI-generated responses are both professional and personalized, and can satisfy most customer inquiries.
• Reduce the pressure of manual customer service: automated handling of common problems allows manual customer service to focus on more complex tasks.
• Data accumulation and analysis: All conversations are recorded in a database, facilitating follow-up analysis and service improvement.
• Highly scalable: The system design allows for easy addition of new features or integration with other services.

On the technical level, we used asyncio and aiohttp libraries to implement asynchronous operations, which greatly improved the efficiency and concurrent processing capability of the system. Well-developed error handling and logging also ensured the stability and maintainability of the system.
This project not only improves our customer service efficiency, but also opens up new possibilities for AI-assisted customer service in the future. We believe that as AI technology continues to advance, systems like this will play an increasingly important role in enhancing customer experience.

Thank you for your attention! If you have any questions or suggestions about this project, feel free to leave a comment in the comments section to discuss.

import asyncio
import aiohttp
import aiosqlite
import time
import json
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 配置
CHATWOOT_BASE_URL = "https://*/api/v1"
CHATWOOT_API_TOKEN = "*"
ACCOUNT_ID = "1"
DIFY_API_URL = "https://*/v1/chat-messages"
DIFY_API_KEY = "Bearer *"
DB_NAME = "conversations.db"

# 初始化数据库
async def init_db():
    async with aiosqlite.connect(DB_NAME) as db:
        await db.execute('''CREATE TABLE IF NOT EXISTS conversations
                            (id INTEGER PRIMARY KEY, 
                             chatwoot_id INTEGER, 
                             dify_id TEXT, 
                             last_message_id INTEGER)''')
        await db.execute('''CREATE TABLE IF NOT EXISTS messages
                            (id INTEGER PRIMARY KEY,
                             conversation_id INTEGER,
                             content TEXT,
                             is_from_user BOOLEAN,
                             timestamp INTEGER,
                             FOREIGN KEY (conversation_id) REFERENCES conversations (id))''')
        await db.commit()
    logging.info("数据库初始化完成")

# 获取开放的对话
async def get_open_conversations(session):
    url = f"{CHATWOOT_BASE_URL}/accounts/{ACCOUNT_ID}/conversations"
    headers = {"api_access_token": CHATWOOT_API_TOKEN, "Authorization": f"Bearer {CHATWOOT_API_TOKEN}"}
    params = {"status": "open"}
    try:
        async with session.get(url, headers=headers, params=params) as response:
            response.raise_for_status()
            data = await response.json()
            conversations = data.get('data', {}).get('payload', [])
            logging.info(f"成功获取到 {len(conversations)} 个开放对话")
            return conversations
    except aiohttp.ClientError as e:
        logging.error(f"获取开放对话时发生错误: {str(e)}")
        return []

# 发送消息到Chatwoot
async def send_message_to_chatwoot(session, conversation_id, content):
    url = f"{CHATWOOT_BASE_URL}/accounts/{ACCOUNT_ID}/conversations/{conversation_id}/messages"
    headers = {
        "api_access_token": CHATWOOT_API_TOKEN,
        "Content-Type": "application/json"
    }
    data = {
        "content": content,
        "message_type": "outgoing"
    }
    try:
        async with session.post(url, headers=headers, json=data) as response:
            response.raise_for_status()
            result = await response.json()
            logging.info(f"成功发送消息到对话 {conversation_id}")
            return result
    except aiohttp.ClientError as e:
        logging.error(f"发送消息到Chatwoot时发生错误: {str(e)}")
        return None

# 获取或创建Dify对话
async def get_or_create_dify_conversation(session, user_message, user_email, dify_conversation_id=None):
    headers = {
        "Authorization": DIFY_API_KEY,
        "Content-Type": "application/json"
    }
    data = {
        "inputs": {},
        "query": user_message,
        "response_mode": "blocking",
        "conversation_id": dify_conversation_id or "",
        "user": user_email,
    }
    try:
        async with session.post(DIFY_API_URL, headers=headers, json=data) as response:
            response.raise_for_status()
            result = await response.json()
            logging.info(f"成功从Dify获取响应，对话ID: {result['conversation_id']}, 用户: {user_email}")
            return result['conversation_id'], result['answer']
    except aiohttp.ClientError as e:
        logging.error(f"调用Dify API时发生错误: {str(e)}")
        return None, None

# 更新数据库
async def update_db(chatwoot_id, dify_id, last_message_id, message_content, is_from_user):
    try:
        async with aiosqlite.connect(DB_NAME) as db:
            await db.execute("INSERT OR REPLACE INTO conversations (chatwoot_id, dify_id, last_message_id) VALUES (?, ?, ?)",
                            (chatwoot_id, dify_id, last_message_id))
            cursor = await db.execute("SELECT last_insert_rowid()")
            conversation_id = await cursor.fetchone()
            await db.execute("INSERT INTO messages (conversation_id, content, is_from_user, timestamp) VALUES (?, ?, ?, ?)",
                            (conversation_id[0], message_content, is_from_user, int(time.time())))
            await db.commit()
        logging.info(f"成功更新数据库，对话ID: {chatwoot_id}")
    except aiosqlite.Error as e:
        logging.error(f"更新数据库时发生错误: {str(e)}")

# 获取存储的Dify对话ID
async def get_stored_dify_conversation_id(chatwoot_id):
    try:
        async with aiosqlite.connect(DB_NAME) as db:
            async with db.execute("SELECT dify_id FROM conversations WHERE chatwoot_id = ?", (chatwoot_id,)) as cursor:
                result = await cursor.fetchone()
        if result:
            logging.info(f"获取到存储的Dify对话ID: {result[0]} (Chatwoot ID: {chatwoot_id})")
        else:
            logging.info(f"未找到Chatwoot ID {chatwoot_id} 对应的Dify对话ID")
        return result[0] if result else None
    except aiosqlite.Error as e:
        logging.error(f"从数据库获取Dify对话ID时发生错误: {str(e)}")
        return None

# 检查未回复的消息
async def check_unreplied_messages(session):
    conversations = await get_open_conversations(session)
    unreplied_messages = []
    for conv in conversations:
        try:
            chatwoot_id = conv['id']
            messages = conv.get('messages', [])
            last_message = messages[-1] if messages else None
            if last_message:
                sender = last_message.get('sender', {})
                if sender.get('type') == 'contact':
                    async with aiosqlite.connect(DB_NAME) as db:
                        async with db.execute("SELECT last_message_id FROM conversations WHERE chatwoot_id = ?", (chatwoot_id,)) as cursor:
                            result = await cursor.fetchone()
                    if not result or result[0] < last_message['id']:
                        unreplied_messages.append({
                            'conversation_id': chatwoot_id,
                            'customer_email': conv.get('meta', {}).get('sender', {}).get('email', 'unknown'),
                            'message_content': last_message.get('content', ''),
                            'last_reply': await get_last_reply(messages),
                            'timestamp': last_message.get('created_at', 0)
                        })
        except KeyError as e:
            logging.error(f"处理对话 {conv.get('id', 'unknown')} 时出现KeyError: {str(e)}")
            logging.debug(f"对话数据: {json.dumps(conv, indent=2)}")
        except Exception as e:
            logging.error(f"处理对话时发生未知错误: {str(e)}")
    
    logging.info(f"检测到 {len(unreplied_messages)} 条未回复的消息")
    return unreplied_messages

# 获取最后一条回复
async def get_last_reply(messages):
    for message in reversed(messages):
        sender = message.get('sender', {})
        if sender.get('type') == 'user':
            return message.get('content', '')
    return "尚未回复"

# 主循环
async def main():
    logging.info("开始运行Chatwoot-Dify集成脚本")
    await init_db()
    async with aiohttp.ClientSession() as session:
        while True:
            try:
                logging.info("开始新的检查循环")
                unreplied = await check_unreplied_messages(session)
                if unreplied:
                    logging.info("发现未回复的消息:")
                    for msg in unreplied:
                        logging.info(f"对话ID: {msg['conversation_id']}")
                        logging.info(f"客户邮箱: {msg['customer_email']}")
                        logging.info(f"客户消息: {msg['message_content']}")
                        logging.info(f"最后回复: {msg['last_reply']}")
                        logging.info(f"消息时间: {datetime.fromtimestamp(msg['timestamp'])}")
                        logging.info("---")

                        # 处理未回复的消息
                        dify_conversation_id = await get_stored_dify_conversation_id(msg['conversation_id'])
                        dify_conversation_id, ai_response = await get_or_create_dify_conversation(
                            session,
                            msg['message_content'], 
                            msg['customer_email'],
                            dify_conversation_id
                        )
                        if ai_response:
                            sent_message = await send_message_to_chatwoot(session, msg['conversation_id'], ai_response)
                            if sent_message:
                                await update_db(msg['conversation_id'], dify_conversation_id, sent_message['id'], 
                                                msg['message_content'], True)
                                await update_db(msg['conversation_id'], dify_conversation_id, sent_message['id'], 
                                                ai_response, False)
                                logging.info(f"已回复对话 {msg['conversation_id']}")
                else:
                    logging.info("没有发现未回复的消息")
            except Exception as e:
                logging.error(f"在主循环中发生错误: {str(e)}", exc_info=True)
            
            await asyncio.sleep(5)

if __name__ == "__main__":
    asyncio.run(main())

[Hanhan0831]

The reason for developing this code is that I think Chatwoot's traditional integration is too unfriendly and can't be able to satisfy people who already have a lot of conversation data or knowledge base. I think anyone can configure the bot in 20 minutes at a very low cost and efficiently.

[panaras10]

The reason for developing this code is that I think Chatwoot's traditional integration is too unfriendly and can't be able to satisfy people who already have a lot of conversation data or knowledge base. I think anyone can configure the bot in 20 minutes at a very low cost and efficiently.

How about the handover to an agent when the bot doesn't know what to answer?

[suporteflowchatt]

Boa noite amigo tudo bem? Gostaria de saber como posso entrar em contato com você para possível implementação.