init

.gitignore

+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[codz]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py.cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# UV
+#   Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#uv.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+#poetry.toml
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#   pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
+#   https://pdm-project.org/en/latest/usage/project/#working-with-version-control
+#pdm.lock
+#pdm.toml
+.pdm-python
+.pdm-build/
+
+# pixi
+#   Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
+#pixi.lock
+#   Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
+#   in the .venv directory. It is recommended not to include this directory in version control.
+.pixi
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.envrc
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+.idea/
+
+# Abstra
+# Abstra is an AI-powered process automation framework.
+# Ignore directories containing user credentials, local state, and settings.
+# Learn more at https://abstra.io/docs
+.abstra/
+
+# Visual Studio Code
+#  Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore 
+#  that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
+#  and can be added to the global gitignore or merged into this file. However, if you prefer, 
+#  you could uncomment the following to ignore the entire vscode folder
+# .vscode/
+
+# Ruff stuff:
+.ruff_cache/
+
+# PyPI configuration file
+.pypirc
+
+# Cursor
+#  Cursor is an AI-powered code editor. `.cursorignore` specifies files/directories to
+#  exclude from AI features like autocomplete and code analysis. Recommended for sensitive data
+#  refer to https://docs.cursor.com/context/ignore-files
+.cursorignore
+.cursorindexingignore
+
+# Marimo
+marimo/_static/
+marimo/_lsp/
+__marimo__/

Dockerfile

+FROM python:3.10-alpine
+
+# 设置工作目录
+WORKDIR /app
+
+# 设置环境变量
+ENV PYTHONDONTWRITEBYTECODE 1
+ENV PYTHONUNBUFFERED 1
+
+# 复制项目文件
+COPY . .
+
+RUN pip install --no-cache-dir -r requirements.txt
+
+CMD ["python", "app.py"]

LICENSE

+MIT License
+
+Copyright (c) 2025 Ladlee
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

+# chatwoot-bot
+Chatwoot's agent bots
 # chatwoot-agent
 

api/__init__.py

+"""
+@Project ：chatwoot-agent 
+@File    ：__init__.py.py
+@Author  ：Lijun
+@Date    ：2025/8/19 11:11 
+"""
+import sys
+from importlib.util import spec_from_file_location, module_from_spec
+from pathlib import Path
+from typing import List
+
+from flask import Blueprint, Flask, request
+from flask_cors import CORS
+
+from config.logger import FlaskLogger, setup_request_logging
+from config.settings import config_dict
+from cache import redis_cache
+
+# 配置常量
+API_VERSION = "v1"
+PAGE_SUFFIX = "_app.py"
+
+
+def register_page(page_path: Path, app: Flask) -> str:
+    """
+    动态注册单个蓝图页面
+    """
+    # 提取页面名称和模块名称
+    page_name = page_path.stem.removesuffix("_app")
+    module_name = ".".join(
+        page_path.parts[page_path.parts.index("api"): -1] + (page_name,)
+    )
+
+    try:
+        # 动态导入模块
+        spec = spec_from_file_location(module_name, page_path)
+        if spec is None or spec.loader is None:
+            raise ImportError(f"无法为 {page_path} 创建模块规范")
+
+        page_module = module_from_spec(spec)
+
+        # 设置模块级变量
+        setattr(page_module, "app", app)
+
+        # 创建或获取蓝图
+        blueprint = getattr(page_module, "manager", None)
+        if blueprint is None:
+            blueprint_name = getattr(page_module, "page_name", page_name)
+            blueprint = Blueprint(blueprint_name, module_name)
+            setattr(page_module, "manager", blueprint)
+
+        sys.modules[module_name] = page_module
+        spec.loader.exec_module(page_module)
+
+        # 确定URL前缀
+        url_prefix = determine_url_prefix(page_name)
+
+        # 注册蓝图
+        app.register_blueprint(blueprint, url_prefix=url_prefix)
+
+        print(f"成功注册蓝图: {page_name} 在 {url_prefix}")
+        return url_prefix
+
+    except Exception as e:
+        print(f"注册蓝图 {page_path} 失败: {str(e)}")
+        raise
+
+
+def determine_url_prefix(page_name: str) -> str:
+    """
+    根据文件路径确定URL前缀
+    :param page_name: 页面名称
+    :return: URL前缀
+    """
+    return f"/{API_VERSION}/{page_name}"
+
+
+def search_pages_path(directory: Path) -> List[Path]:
+    """
+    搜索目录下的所有蓝图文件
+    :param directory: 要搜索的目录
+    :return: 找到的蓝图文件路径列表
+    """
+    if not directory.is_dir():
+        return []
+
+    return [
+        path for path in directory.glob(f"*{PAGE_SUFFIX}")
+        if path.is_file() and not path.name.startswith(".")
+    ]
+
+
+def get_pages_directories() -> List[Path]:
+    """
+    获取要搜索的目录列表
+    :return: 目录路径列表
+    """
+    base_dir = Path(__file__).parent
+    return [
+        base_dir
+    ]
+
+
+def register_all_pages(app) -> List[str]:
+    """
+    注册所有蓝图页面
+    :return: 所有注册的URL前缀列表
+    """
+    pages_dirs = get_pages_directories()
+    return [
+        register_page(page_path, app)
+        for directory in pages_dirs
+        for page_path in search_pages_path(directory)
+    ]
+
+
+def create_app(config_name="development"):
+    """应用工厂函数"""
+    app = Flask(__name__)
+    # 全局跨域配置(接口层面: @cross_origin(origin='*'))
+    # CORS(app, resources={r"/api/*": {"origins": "*"}})
+    CORS(app, resources=r"/*")
+    # 加载配置
+    app.config.from_object(config_dict[config_name])
+    # 初始化日志
+    logger = FlaskLogger()
+    logger.init_app(app)
+    # 设置请求日志
+    setup_request_logging(app)
+    # 初始化缓存
+    initialize_cache(app)
+    # 注册蓝图
+    register_blueprints(app)
+    # 注册错误处理
+    register_error_handlers(app)
+    return app
+
+
+def initialize_cache(app):
+    redis_cache.init_app(app)
+
+
+def register_blueprints(app):
+    register_all_pages(app)
+
+
+def register_error_handlers(app):
+    """注册错误处理器"""
+    from flask import jsonify
+    import logging
+
+    logger = logging.getLogger(__name__)
+
+    @app.errorhandler(404)
+    def not_found(error):
+        logger.warning(f"404 Not Found: {request.path}")
+        return jsonify({'error': 'Not found'}), 404
+
+    @app.errorhandler(500)
+    def internal_error(error):
+        logger.error(f"500 Internal Error: {str(error)}")
+        return jsonify({'error': 'Internal server error'}), 500
+
+    @app.errorhandler(Exception)
+    def handle_exception(error):
+        logger.exception("Unhandled exception occurred")
+        return jsonify({'error': 'Internal server error'}), 500

api/chatwoot_app.py

+"""
+@Project ：chatwoot-agent 
+@File    ：chatwoot_app.py
+@Author  ：Lijun
+@Date    ：2025/8/19 11:22 
+"""
+from functools import wraps
+from typing import Optional
+
+from flask import (
+    request as flask_request, jsonify
+)
+
+from config.logger import get_logger
+from model.chatwoot_model import MessageEvent
+from service.chatwoot_service import ChatwootConversationService
+
+logger = get_logger(__name__)
+
+
+def validate_request():
+    def wrapper(func):
+        @wraps(func)
+        def decorated_function():
+            json_data = flask_request.json
+            logger.info(f"请求参数:{json_data}")
+            message_event = MessageEvent(**json_data)
+            return func(message_event)
+
+        return decorated_function
+
+    return wrapper
+
+
+@manager.route('', methods=['POST'])
+@validate_request()
+def chatwoot_agent(message_event: Optional[MessageEvent]):
+    ChatwootConversationService.handle_conversation(message_event)
+    return jsonify({"status": "success"}), 200

app.py

+from api import create_app
+from config.settings import Config
+
+app = create_app(config_name=Config.APP_ENV)
+
+if __name__ == '__main__':
+    app.run(host='0.0.0.0', port=Config.FLASK_PORT, debug=True)

cache/__init__.py

+"""
+@Project ：chatwoot-agent 
+@File    ：__init__.py.py
+@Author  ：Lijun
+@Date    ：2025/8/19 9:46 
+"""
+from redis import Redis, RedisError
+
+
+class RedisCache:
+    def __init__(self, app=None):
+        self.app = app
+        self.redis_client = None
+        if app is not None:
+            self.init_app(app)
+
+    def init_app(self, app):
+        """初始化连接redis"""
+        try:
+            self.redis_client = Redis(
+                host=app.config["REDIS_HOST"],
+                port=int(app.config["REDIS_PORT"]),
+                db=app.config['REDIS_DB'],
+                password=app.config['REDIS_PASSWORD'],
+                decode_responses=True,
+                socket_timeout=5,
+                socket_connect_timeout=5
+            )
+            # 测试连接
+            self.redis_client.ping()
+            app.logger.info('Redis cache connected successfully')
+        except RedisError as e:
+            app.logger.error(f"Failed to connect to Redis: {str(e)}")
+            raise
+
+    def get_client(self):
+        if not self.redis_client:
+            raise RuntimeError("Redis cache not initialized")
+        return self.redis_client
+
+
+# 初始化扩展实例
+redis_cache = RedisCache()

cache/services.py

+"""
+@Project ：chatwoot-agent 
+@File    ：services.py
+@Author  ：Lijun
+@Date    ：2025/8/19 14:17 
+"""
+from typing import List
+
+from redis import Redis
+
+from cache import redis_cache
+from config.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class CacheService:
+
+    @staticmethod
+    def set(cache_key: str, data: str, ttl: int = None):
+        try:
+            redis_client = redis_cache.get_client()
+            if ttl is None:
+                redis_client.set(cache_key, data)
+            else:
+                redis_client.setex(cache_key, ttl, data)
+        except Exception as e:
+            logger.error(f"Cache set error: {str(e)}")
+
+    @staticmethod
+    def get(cache_key: str):
+        try:
+            redis_client: Redis = redis_cache.get_client()
+            return redis_client.get(cache_key)
+        except Exception:
+            logger.error(f"未获取到key:{cache_key}的值.")
+            return None
+
+    @staticmethod
+    def lrange(cache_key: str, start: int, end: int) -> List:
+        try:
+            redis_client = redis_cache.get_client()
+            return redis_client.lrange(cache_key, start, end)
+        except Exception:
+            logger.error(f"未获取到key:{cache_key}的值.")
+            return []
+
+    @staticmethod
+    def invalidate(cache_key: str):
+        redis_client: Redis = redis_cache.get_client()
+        try:
+            logger.info(f"删除历史对话的session_id: {cache_key}")
+            redis_client.delete(cache_key)
+        except Exception as e:
+            logger.error(f'Cache invalidation error: {str(e)}')

config/logger.py

+"""
+@Project ：chatwoot-agent 
+@File    ：logger.py
+@Author  ：Lijun
+@Date    ：2025/8/20 11:11 
+"""
+import logging
+import time
+from functools import wraps
+from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
+import os
+
+
+class FlaskLogger:
+
+    def __init__(self, app=None):
+        self.app = app
+        if app is not None:
+            self.init_app(app)
+
+    def init_app(self, app):
+        """初始化应用日志"""
+        config = app.config
+
+        # 确保日志目录存在
+        if not os.path.exists(config["LOG_DIR"]):
+            os.makedirs(config["LOG_DIR"])
+
+        # 配置根日志器
+        root_logger = logging.getLogger()
+        root_logger.setLevel(config["LOG_LEVEL"])
+
+        # 清除现有的处理器
+        for handler in root_logger.handlers[:]:
+            root_logger.removeHandler(handler)
+
+        # 创建格式化器
+        formatter = logging.Formatter(config["LOG_FORMAT"])
+
+        # 控制台处理器
+        console_handler = logging.StreamHandler()
+        console_handler.setLevel(config["LOG_LEVEL"])
+        console_handler.setFormatter(formatter)
+        root_logger.addHandler(console_handler)
+
+        # 应用日志文件处理器（按大小分割）
+        file_handler = RotatingFileHandler(
+            config["APP_LOG_FILE"],
+            maxBytes=10 * 1024 * 1024,  # 10MB
+            backupCount=5,
+            encoding='utf-8'
+        )
+        file_handler.setLevel(config["LOG_LEVEL"])
+        file_handler.setFormatter(formatter)
+        root_logger.addHandler(file_handler)
+
+        # 错误日志文件处理器
+        error_handler = RotatingFileHandler(
+            config["ERROR_LOG_FILE"],
+            maxBytes=10 * 1024 * 1024,
+            backupCount=5,
+            encoding='utf-8'
+        )
+        error_handler.setLevel(logging.ERROR)
+        error_handler.setFormatter(formatter)
+        root_logger.addHandler(error_handler)
+
+        # 访问日志文件处理器（按天分割）
+        access_handler = TimedRotatingFileHandler(
+            config["ACCESS_LOG_FILE"],
+            when='midnight',
+            interval=1,
+            backupCount=30,
+            encoding='utf-8'
+        )
+        access_handler.setLevel(logging.INFO)
+        access_handler.setFormatter(formatter)
+
+        # 创建访问日志器
+        access_logger = logging.getLogger('access')
+        access_logger.setLevel(logging.INFO)
+        access_logger.addHandler(access_handler)
+        access_logger.propagate = False
+
+        # 设置Flask应用日志器
+        app.logger = root_logger
+
+        # 禁用Werkzeug的默认日志
+        logging.getLogger('werkzeug').setLevel(logging.WARNING)
+
+
+def get_logger(name='app'):
+    """获取日志器"""
+    return logging.getLogger(name)
+
+
+def log_execution_time(func):
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        logger = get_logger("performance")
+        start_time = time.time()
+
+        try:
+            result = func(*args, **kwargs)
+            execution_time = time.time() - start_time
+            if execution_time > 1.0:
+                logger.warning(
+                    f"Slow execution: {func.__name__} took {execution_time:.3f}s"
+                )
+            elif execution_time > 0.5:
+                logger.info(
+                    f"Execution time: {func.__name__} took {execution_time:.3f}s"
+                )
+
+            return result
+        except Exception as e:
+            execution_time = time.time() - start_time
+            logger.error(
+                f"Error in {func.__name__} after {execution_time:.3f}s: {str(e)}"
+            )
+            raise
+
+    return wrapper
+
+
+def setup_request_logging(app):
+    """设置请求日志中间件"""
+    access_logger = get_logger('access')
+
+    @app.before_request
+    def before_request():
+        """记录请求开始"""
+        from flask import request
+        request.start_time = time.time()
+
+    @app.after_request
+    def after_request(response):
+        """记录请求完成"""
+        from flask import request
+
+        # 计算处理时间
+        process_time = time.time() - getattr(request, 'start_time', time.time())
+
+        # 记录访问日志
+        access_logger.info(
+            f"{request.method} {request.path} - {response.status_code} "
+            f"- {process_time:.3f}s - {request.remote_addr}"
+        )
+
+        return response

config/settings.py

+"""
+@Project ：chatwoot-agent 
+@File    ：settings.py
+@Author  ：Lijun
+@Date    ：2025/8/19 10:36 
+"""
+import logging
+import os
+
+from dotenv import load_dotenv
+
+
+class Config:
+    # 加载配置
+    load_dotenv()
+    # redis配置
+    REDIS_HOST = os.getenv("REDIS_HOST", "10.10.31.26")
+    print(f"redis的连接地址为:{REDIS_HOST}")
+    REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
+    REDIS_DB = int(os.getenv("REDIS_DB", 2))
+    REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", None)
+    REDIS_CACHE_TTL = os.getenv("REDIS_CACHE_TTL", 300)  # 默认5分钟
+
+    # 环境配置
+    APP_ENV = os.getenv("APP_ENV", "development")
+    FLASK_PORT = int(os.getenv("FLASK_PORT", 4000))
+
+    # chatwoot配置
+    CHATWOOT_BOT_TOKEN = os.getenv("CHATWOOT_BOT_TOKEN")
+    CHATWOOT_URL = os.getenv("CHATWOOT_URL")
+
+    # ragflow配置
+    RAGFLOW_AGENT_AUTHORIZATION = os.getenv("RAGFLOW_AGENT_AUTHORIZATION")
+    RAGFLOW_AGENT_URL = os.getenv("RAGFLOW_AGENT_URL")
+    RAGFLOW_AGENT_ID = os.getenv("RAGFLOW_AGENT_ID")
+
+    # 日志配置
+    LOG_LEVEL = logging.INFO
+    LOG_DIR = 'logs'
+
+    if not os.path.exists(LOG_DIR):
+        os.makedirs(LOG_DIR)
+
+    # 日志文件路径
+    APP_LOG_FILE = os.path.join(LOG_DIR, 'app.log')
+    ERROR_LOG_FILE = os.path.join(LOG_DIR, 'error.log')
+    ACCESS_LOG_FILE = os.path.join(LOG_DIR, 'access.log')
+
+    # 日志格式
+    LOG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s'
+
+
+class ProductionConfig(Config):
+    # 生产环境1小时
+    REDIS_CACHE_TTL = 3600
+    LOG_LEVEL = logging.INFO
+
+
+class DevelopmentConfig(Config):
+    DEBUG = True
+    LOG_LEVEL = logging.DEBUG
+
+
+config_dict = {
+    "production": ProductionConfig,
+    "development": DevelopmentConfig
+}

docker-compose.yml

+version: "3"
+
+services:
+  chatwoot:
+    build: .
+    ports:
+      - "${FLASK_PORT}:${FLASK_PORT}"
+    env_file:
+      - .env
+    volumes:
+      - ./logs:/app/logs 

handler/__init__.py

+"""
+@Project ：chatwoot-agent 
+@File    ：__init__.py.py
+@Author  ：Lijun
+@Date    ：2025/8/19 10:09 
+"""
+
+from abc import ABC, abstractmethod
+
+from cache.services import CacheService
+from config.logger import get_logger
+from handler.base_api import create_new_message, toggle_status, ragflow_agent
+from model.chatwoot_model import MessageEvent
+
+logger = get_logger(__name__)
+
+
+# 事件数据
+class Event:
+    def __init__(self, event_type: str, data: MessageEvent):
+        self.event_type = event_type
+        self.data = data
+
+
+# 消息数据
+class Message:
+    def __init__(self, message_type, message_status, data: MessageEvent):
+        self.message_type = message_type
+        self.message_status = message_status
+        self.data = data
+
+
+# 事件的处理器接口
+class EventHandler(ABC):
+    @abstractmethod
+    def handle_event(self, event: Event):
+        pass
+
+    @abstractmethod
+    def support(self, event: Event) -> bool:
+        return False
+
+
+# 消息的处理器接口
+class MessageHandler(ABC):
+    @abstractmethod
+    def handle_message(self, message: Message):
+        pass
+
+    @abstractmethod
+    def support(self, message: Message) -> bool:
+        return False
+
+
+# 消息已创建事件
+class MessageCreatedEventHandler(EventHandler):
+
+    def handle_event(self, event: Event):
+        processor = MessageProcessor()
+        message = Message(event.data.message_type, event.data.conversation.status, event.data)
+        return processor.process(message=message)
+
+    def support(self, event: Event) -> bool:
+        return event.event_type == 'message_created'
+
+
+# 对话已解决事件
+class ConversationResolvedHandler(EventHandler):
+
+    def handle_event(self, event: Event):
+        logger.info(f"ConversationResolvedHandler处理事件消息.")
+        if event.data is None:
+            return
+        contact_inbox = event.data.contact_inbox
+        if contact_inbox is None or contact_inbox.id is None:
+            return
+        contact_id = contact_inbox.id
+        CacheService.invalidate(f"c:cr:{contact_id}")
+
+    def support(self, event: Event) -> bool:
+        return event.event_type == 'conversation_resolved'
+
+
+class NothingEventHandler(EventHandler):
+
+    def handle_event(self, event: Event):
+        logger.info(f"Event type:{event.event_type}, Nothing to do")
+
+    def support(self, event: Event) -> bool:
+        return True
+
+
+class NormalMessageHandler(MessageHandler):
+
+    def handle_message(self, message: Message):
+        logger.info(f"NormalMessageHandler处理消息.")
+        message_data = message.data
+        if message_data is None:
+            return
+        message_sender = message_data.sender
+        if message_sender is None:
+            return
+        # 发送给bot
+        sender_id = message_sender.id
+        agent_response = ragflow_agent(sender_id, message_data.content)
+        # 发送给chatwoot
+        conversation_id = message_data.conversation.id
+        account_id = message_data.account.id
+        # 发送给chatwoot
+        create_new_message(account_id, conversation_id, agent_response)
+
+    def support(self, message: Message) -> bool:
+        if message.message_type == "incoming" and message.message_status == "pending":
+            return True
+        return False
+
+
+class ToggleStatusHandler(MessageHandler):
+
+    def handle_message(self, message: Message):
+        logger.info(f"ToggleStatusHandler处理消息.")
+        conversation_id = message.data.conversation.id
+        account_id = message.data.account.id
+        return toggle_status(account_id, conversation_id, "open")
+
+    def support(self, message: Message) -> bool:
+        if message.message_type == "incoming" and message.message_status == "pending":
+            # 从redis获取需要转换状态的关键词
+            content = message.data.content
+            kw_words = CacheService.lrange("c:kw:toggle", 0, -1)
+            logger.info(f"Redis中转换消息状态的关键词:{kw_words}")
+            if any(key in content for key in kw_words):
+                return True
+        return False
+
+
+class OtherMessageHandler(MessageHandler):
+
+    def handle_message(self, message: Message):
+        logger.info(f"Message type:{message.message_type}, status:{message.message_status},  Nothing to do")
+
+    def support(self, message: Message) -> bool:
+        return True
+
+
+class MessageProcessor:
+
+    def __init__(self):
+        self._message_handlers = [
+            ToggleStatusHandler(),
+            NormalMessageHandler(),
+            OtherMessageHandler()
+        ]
+
+    def process(self, message: Message):
+        for handler in self._message_handlers:
+            if handler.support(message):
+                return handler.handle_message(message)
+        raise ValueError(f"没有找到支持消息类型 {message.message_type} 和状态 {message.message_status} 的处理器")

handler/base_api.py

+"""
+@Project ：chatwoot-agent 
+@File    ：base_api.py
+@Author  ：Lijun
+@Date    ：2025/8/29 16:19 
+"""
+import requests
+
+from cache.services import CacheService
+from config.logger import get_logger
+from config.settings import Config
+
+logger = get_logger(__name__)
+
+
+def remove_null_bytes(data):
+    if isinstance(data, dict):
+        return {k: remove_null_bytes(v) for k, v in data.items()}
+    elif isinstance(data, list):
+        return [remove_null_bytes(item) for item in data]
+    elif isinstance(data, str):
+        return data.replace('\x00', '')
+    else:
+        return data
+
+
+def create_new_message(account_id, conversation_id, message):
+    data = {
+        'content': remove_null_bytes(message)
+    }
+    url = f"{Config.CHATWOOT_URL}/api/v1/accounts/{account_id}/conversations/{conversation_id}/messages"
+    headers = {"Content-Type": "application/json",
+               "Accept": "application/json",
+               "api_access_token": f"{Config.CHATWOOT_BOT_TOKEN}"}
+
+    r = requests.post(url, json=data, headers=headers)
+    return r.json()
+
+
+def toggle_status(account_id, conversation_id, status):
+    data = {
+        'status': status
+    }
+    url = f"{Config.CHATWOOT_URL}/api/v1/accounts/{account_id}/conversations/{conversation_id}/toggle_status"
+    headers = {"Content-Type": "application/json",
+               "Accept": "application/json",
+               "api_access_token": f"{Config.CHATWOOT_BOT_TOKEN}"}
+    r = requests.post(url, json=data, headers=headers)
+    return r.json()
+
+
+def ragflow_agent(sender_id: int, msg: str):
+    try:
+        agent_url = f"{Config.RAGFLOW_AGENT_URL}/api/v1/agentbots/{Config.RAGFLOW_AGENT_ID}/completions"
+        # 从redis获取session_id
+        session_id = CacheService.get(f"c:cr:{sender_id}")
+
+        agent_data = {
+            "question": msg,
+            "stream": 0,
+            "session_id": session_id
+        }
+
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+            "Authorization": f"Bearer {Config.RAGFLOW_AGENT_AUTHORIZATION}"
+        }
+        agent_res = requests.post(agent_url, json=agent_data, headers=headers)
+        json_res = agent_res.json()
+        code = json_res.get("code")
+        if code != 0:
+            raise ValueError
+        answer = json_res["data"]["answer"]
+        session_id = json_res["data"]["session_id"]
+        if session_id is not None:
+            CacheService.set(f"c:cr:{sender_id}", session_id)
+        return answer
+    except Exception as e:
+        logger.info(f"请求ragflow出现故障:{e}")
+        return "智能系统出现故障,转入人工客服."

handler/event_processor.py

+"""
+@Project ：chatwoot-agent 
+@File    ：event_processor.py
+@Author  ：Lijun
+@Date    ：2025/8/20 9:26 
+"""
+from handler import ConversationResolvedHandler, MessageCreatedEventHandler, NothingEventHandler, Event
+
+
+class EventProcessor:
+
+    def __init__(self):
+        self.processors = [
+            MessageCreatedEventHandler(),
+            ConversationResolvedHandler(),
+            NothingEventHandler()
+        ]
+
+    def process(self, event: Event):
+        for processor in self.processors:
+            if processor.support(event):
+                return processor.handle_event(event)
+        raise ValueError(f"没有找到支持事件类型 {event.event_type}的处理器")

model/__init__.py

+"""
+@Project ：chatwoot-agent 
+@File    ：__init__.py.py
+@Author  ：Lijun
+@Date    ：2025/8/19 16:10 
+"""

model/chatwoot_model.py

+import json
+from datetime import datetime
+from typing import Optional, Dict, List, Any
+from pydantic import BaseModel, Field
+
+
+# 基础模型
+class Account(BaseModel):
+    id: int
+    name: str
+
+
+class Inbox(BaseModel):
+    id: int
+    name: str
+
+
+class Sender(BaseModel):
+    id: int
+    name: Optional[str] = Field(default=None)
+    type: Optional[str] = Field(default=None)
+    avatar_url: Optional[str] = Field(default=None)
+
+
+class ContactInbox(BaseModel):
+    id: int
+    contact_id: int
+    inbox_id: int
+    source_id: str
+    created_at: datetime
+    updated_at: datetime
+    hmac_verified: bool
+    pubsub_token: str
+
+
+class BrowserInfo(BaseModel):
+    device_name: str
+    browser_name: str
+    platform_name: str
+    browser_version: str
+    platform_version: str
+
+
+class InitiatedAt(BaseModel):
+    timestamp: str
+
+
+class AdditionalAttributes(BaseModel):
+    browser: Optional[BrowserInfo] = None
+    referer: Optional[Any] = None
+    initiated_at: Optional[InitiatedAt] = None
+    browser_language: Optional[str] = None
+
+
+class MessageConversation(BaseModel):
+    assignee_id: Optional[int] = None
+    unread_count: int
+    last_activity_at: int
+    contact_inbox: Dict[str, str]
+
+
+class MessageSender(BaseModel):
+    id: int
+    name: Optional[str] = Field(default=None)
+    avatar_url: Optional[str] = Field(default=None)
+    type: Optional[str] = Field(default=None)
+
+
+class Message(BaseModel):
+    id: int
+    content: Optional[str] = Field(default=False)
+    account_id: int
+    inbox_id: int
+    conversation_id: int
+    message_type: int = None
+    created_at: int
+    updated_at: datetime
+    private: Optional[bool] = Field(default=False)
+    status: Optional[str] = Field(default=False)
+    source_id: Optional[str] = Field(default=False)
+    content_type: Optional[str] = Field(default=False)
+    content_attributes: Dict[str, Any] = Field(default=False)
+    sender_type: Optional[str] = Field(default=False)
+    sender_id: Optional[int] = Field(default=None)
+    external_source_ids: Dict[str, Any]
+    additional_attributes: Dict[str, Any]
+    processed_message_content: Optional[str] = Field(default=None)
+    sentiment: Dict[str, Any]
+    conversation: Optional[MessageConversation] = Field(default=None)
+    sender: Optional[MessageSender] = Field(default=None)
+
+
+class MetaSender(BaseModel):
+    additional_attributes: Dict[str, Any]
+    custom_attributes: Dict[str, Any]
+    email: Optional[str] = Field(default=None)
+    id: int
+    identifier: Optional[str] = Field(default=None)
+    name: Optional[str] = Field(default=None)
+    phone_number: Optional[str] = Field(default=None)
+    thumbnail: Optional[str] = Field(default=None)
+    blocked: Optional[bool] = Field(default=False)
+    type: Optional[str] = Field(default=None)
+
+
+class Meta(BaseModel):
+    sender: Optional[MetaSender] = Field(default=None)
+    assignee: Optional[Any] = None
+    team: Optional[Any] = None
+    hmac_verified: bool
+
+
+class Conversation(BaseModel):
+    additional_attributes: AdditionalAttributes = Field(default=None)
+    can_reply: bool
+    channel: str
+    contact_inbox: Optional[ContactInbox] = Field(default=None)
+    id: int
+    inbox_id: int
+    messages: Optional[List[Message]] = Field(default=[])
+    labels: List[Any]
+    meta: Optional[Meta] = Field(default=None)
+    status: str
+    custom_attributes: Dict[str, Any]
+    snoozed_until: Optional[Any] = None
+    unread_count: int
+    first_reply_created_at: Optional[Any] = None
+    priority: Optional[Any] = None
+    waiting_since: int
+    agent_last_seen_at: int
+    contact_last_seen_at: int
+    last_activity_at: int
+    timestamp: int
+    created_at: int
+    updated_at: float
+
+
+# 主模型
+class MessageEvent(BaseModel):
+    account: Optional[Account] = Field(default=None)
+    additional_attributes: Dict[str, Any] = Field(default=None)
+    content_attributes: Dict[str, Any] = Field(default=None)
+    content_type: Optional[str] = Field(default=None)
+    content: Optional[str] = Field(default=None)
+    conversation: Optional[Conversation] = Field(default=None)
+    created_at: Optional[datetime] = Field(default=None)
+    id: int
+    inbox: Optional[Inbox] = Field(default=None)
+    message_type: Optional[str] = None
+    private: Optional[bool] = Field(default=False)
+    sender: Optional[Sender] = Field(default=None)
+    source_id: Optional[str] = Field(default=None)
+    event: str
+    contact_inbox: Optional[ContactInbox] = Field(default=None)

requirements.txt

+Flask==3.1.1
+redis==6.4.0
+flask-cors==6.0.1
+python-dotenv==1.1.0
+requests==2.32.4
+pydantic==2.11.5
\ No newline at end of file

service/__init__.py

+"""
+@Project ：chatwoot-agent 
+@File    ：__init__.py.py
+@Author  ：Lijun
+@Date    ：2025/8/19 9:48 
+"""

service/chatwoot_service.py

+"""
+@Project ：chatwoot-agent 
+@File    ：chatwoot_service.py
+@Author  ：Lijun
+@Date    ：2025/8/19 9:48 
+"""
+from handler import Event
+from handler.event_processor import EventProcessor
+from model.chatwoot_model import MessageEvent
+
+
+class ChatwootConversationService:
+
+    @classmethod
+    def handle_conversation(cls, message_event: MessageEvent):
+        processor = EventProcessor()
+        processor.process(event=Event(message_event.event, message_event))

utils/__init__.py

+"""
+@Project ：chatwoot-agent 
+@File    ：__init__.py.py
+@Author  ：Lijun
+@Date    ：2025/8/19 15:40 
+"""