用 ChatterBot 和 Ollama 在本地搭一个属于你的 Python 聊天机器人

用户输入 → 逻辑适配器(Logic Adapter) → 语料存储(Statement Storage) → 返回最佳匹配回复

import pandas as pd
import re

def clean_message(text: str) -> str:
    """去掉表情符号和多余空白，保留核心文本"""
    text = re.sub(r'[^\w\s，。！？、；：""（）]', '', text)
    text = re.sub(r'\s+', ' ', text).strip()
    return text

def build_pairs(df: pd.DataFrame) -> list[tuple[str, str]]:
    """从聊天记录中提取 (输入, 回复) 对"""
    df = df.sort_values('timestamp')
    messages = df['message'].tolist()
    users = df['user'].tolist()
    pairs = []
    for i in range(len(messages) - 1):
        if users[i] != users[i + 1]:  # 不同用户之间的相邻消息
            inp = clean_message(messages[i])
            resp = clean_message(messages[i + 1])
            if len(inp) >= 3 and len(resp) >= 3:  # 过滤过短消息
                pairs.append((inp, resp))
    # 去重：同一输入保留最长回复
    deduped = {}
    for inp, resp in pairs:
        if inp not in deduped or len(resp) > len(deduped[inp]):
            deduped[inp] = resp
    return list(deduped.items())

# 使用示例
df = pd.read_csv('chat_logs.csv')
pairs = build_pairs(df)
print(f"清洗后得到 {len(pairs)} 个对话对")

from chatterbot import ChatBot
from chatterbot.trainers import ListTrainer

# 创建机器人，指定中文语料存储
chatbot = ChatBot(
    'MyBot',
    storage_adapter='chatterbot.storage.SQLStorageAdapter',
    logic_adapters=[
        {
            'import_path': 'chatterbot.logic.BestMatch',
            'default_response': '我暂时不太明白你的意思，换个说法试试？',
            'maximum_similarity_threshold': 0.90  # 低于此阈值返回默认回复
        }
    ],
    database_uri='sqlite:///chatbot_db.sqlite3'
)

trainer = ListTrainer(chatbot)

# pairs 是上一节清洗出来的 (输入, 回复) 列表
for statement, response in pairs:
    trainer.train([statement, response])

# 快速测试
print(chatbot.get_response('你好'))

# 安装 Ollama（macOS / Linux）
curl -fsSL https://ollama.com/install.sh | sh

# 拉一个轻量中文模型
ollama pull qwen2:1.5b

# 验证服务可用
curl http://localhost:11434/api/tags

import requests
from chatterbot.logic import LogicAdapter
from chatterbot.conversation import Statement

class OllamaFallbackAdapter(LogicAdapter):
    """ChatterBot 匹配失败时，调用 Ollama 生成回复"""

    def __init__(self, chatbot, **kwargs):
        super().__init__(chatbot, **kwargs)
        self.ollama_url = kwargs.get('ollama_url', 'http://localhost:11434/api/generate')
        self.model = kwargs.get('ollama_model', 'qwen2:1.5b')
        self.confidence_threshold = kwargs.get('confidence_threshold', 0.5)

    def process(self, input_statement: Statement, additional_response_selection_parameters=None):
        # 先用 BestMatch 的逻辑找最佳匹配
        best_match = self.chatbot.storage.search(
            input_statement.text,
            maximum_similarity_threshold=self.confidence_threshold
        )

        if best_match and best_match.confidence >= self.confidence_threshold:
            return best_match, best_match.confidence

        # 匹配失败，调用 Ollama
        prompt = input_statement.text
        payload = {
            'model': self.model,
            'prompt': prompt,
            'stream': False
        }
        try:
            resp = requests.post(self.ollama_url, json=payload, timeout=30)
            resp.raise_for_status()
            generated = resp.json().get('response', '').strip()
        except Exception as e:
            generated = f'本地模型暂时无法响应：{e}'

        response_statement = Statement(text=generated)
        response_statement.confidence = 0.4  # 生成式回复置信度标记低一些
        return response_statement, response_statement.confidence

chatbot = ChatBot(
    'HybridBot',
    storage_adapter='chatterbot.storage.SQLStorageAdapter',
    logic_adapters=[
        {
            'import_path': 'OllamaFallbackAdapter',
            'ollama_url': 'http://localhost:11434/api/generate',
            'ollama_model': 'qwen2:1.5b',
            'confidence_threshold': 0.5
        }
    ],
    database_uri='sqlite:///chatbot_db.sqlite3'
)

# 测试：语料里有的问题走检索，没有的走 Ollama
print(chatbot.get_response('你好'))          # 走检索
print(chatbot.get_response('解释一下量子计算'))  # 走 Ollama