QQbot/src/ai/docs/README.md

AI模型接入完整指南

概述

本系统提供了完整的AI模型接入能力，包括：

• ✅ 多模型支持：OpenAI、Anthropic Claude、DeepSeek、Qwen等
• ✅ 人格系统：可自定义AI性格特征和说话风格
• ✅ 记忆系统：短期记忆（滑动窗口）+ 长期记忆（RAG检索）
• ✅ 长任务处理：支持多步骤复杂任务的执行和管理
• ✅ Skills插件：可扩展的技能插件系统
• ✅ MCP支持：Model Context Protocol标准支持

快速开始

1. 安装依赖

pip install openai anthropic numpy

2. 配置环境变量

在 .env 文件中添加：

# AI模型配置
AI_PROVIDER=openai  # openai, anthropic, deepseek, qwen
AI_MODEL=gpt-4
AI_API_KEY=your_api_key_here
AI_API_BASE=https://api.openai.com/v1  # 可选，自定义API地址

3. 基础使用

from pathlib import Path
from src.ai import AIClient
from src.ai.base import ModelConfig, ModelProvider

# 创建模型配置
config = ModelConfig(
    provider=ModelProvider.OPENAI,
    model_name="gpt-4",
    api_key="your_api_key",
    temperature=0.7
)

# 初始化AI客户端
client = AIClient(config, data_dir=Path("data/ai"))

# 对话
response = await client.chat(
    user_id="user123",
    user_message="你好，介绍一下你自己",
    use_memory=True,
    use_tools=True
)

print(response)

核心功能详解

1. 人格系统

人格系统允许你为AI定义独特的性格特征和说话风格。

使用预设人格

# 列出所有人格
personalities = client.list_personalities()
print(personalities)  # ['default', 'tech_expert', 'creative']

# 切换人格
client.set_personality('tech_expert')

自定义人格

from src.ai.personality import PersonalityProfile, PersonalityTrait

# 创建自定义人格
custom_personality = PersonalityProfile(
    name="游戏助手",
    description="专业的游戏顾问和陪玩伙伴",
    traits=[
        PersonalityTrait.FRIENDLY,
        PersonalityTrait.HUMOROUS,
        PersonalityTrait.CREATIVE
    ],
    speaking_style="活泼、热情、善用游戏术语和梗",
    example_responses=[
        "哇！这个操作太秀了！",
        "让我们一起冲冲冲！",
        "别慌，稳住我们能赢！"
    ],
    custom_instructions="始终保持积极乐观的态度，鼓励玩家"
)

# 添加到系统
client.personality.add_personality('gamer', custom_personality)
client.set_personality('gamer')

2. 记忆系统

记忆系统分为短期记忆和长期记忆两部分。

短期记忆

• 自动保存最近的对话（默认20条）
• 使用滑动窗口，自动清理过期记忆（默认30分钟）
• 无需手动管理

长期记忆

• 自动评估重要性，重要对话会被保存
• 支持向量检索（RAG）
• 可手动查询历史记忆

# 获取用户的记忆上下文
short_term, long_term = await client.memory.get_context(
    user_id="user123",
    query="我们之前讨论过什么？",
    max_short_term=10,
    max_long_term=5
)

# 格式化为文本
context = client.memory.format_context(short_term, long_term)
print(context)

# 清除短期记忆
client.clear_memory("user123")

记忆重要性评估

系统会自动评估对话的重要性：

• 内容长度（>100字符 +0.1分）
• 关键词（"重要"、"记住"、"别忘了" +0.2分）
• 元数据（问题 +0.1分，代码 +0.15分）
• 阈值：≥0.6分会被保存到长期记忆

3. 长任务处理

支持将复杂任务分解为多个步骤执行。

# 创建长任务
task_id = await client.create_long_task(
    user_id="user123",
    title="数据分析任务",
    description="分析用户行为数据并生成报告",
    steps=[
        {
            "description": "加载数据",
            "action": "load_data",
            "params": {"file_path": "data.csv"}
        },
        {
            "description": "数据清洗",
            "action": "clean_data",
            "params": {}
        },
        {
            "description": "统计分析",
            "action": "analyze",
            "params": {"metrics": ["avg", "max", "min"]}
        },
        {
            "description": "生成报告",
            "action": "generate_report",
            "params": {"format": "pdf"}
        }
    ]
)

# 定义进度回调
async def on_progress(task_id, progress, message):
    print(f"任务 {task_id}: {progress*100:.1f}% - {message}")

# 启动任务
await client.start_task(task_id, progress_callback=on_progress)

# 查询任务状态
status = client.get_task_status(task_id)
print(status)

注册任务动作处理器

# 注册动作处理器
async def load_data_handler(file_path: str):
    # 实现数据加载逻辑
    return {"rows": 1000, "columns": 10}

client.task_manager.register_action("load_data", load_data_handler)

4. Skills插件系统

Skills系统允许你扩展AI的能力。

加载技能

from pathlib import Path
from src.ai.skills import SkillsManager

# 创建技能管理器
skills_manager = SkillsManager(skills_dir=Path("skills"))

# 加载单个技能
await skills_manager.load_skill("weather")

# 加载所有技能
await skills_manager.load_all_skills()

# 获取所有工具
tools = skills_manager.get_all_tools()

创建自定义技能

from src.ai.skills import create_skill_template

# 创建技能模板
create_skill_template("my_skill", Path("skills"))

技能目录结构：

skills/
└── my_skill/
    ├── skill.json      # 元数据
    ├── main.py         # 主要实现
    ├── __init__.py
    └── README.md

main.py 示例：

from src.ai.skills.base import Skill

class MySkill(Skill):
    async def initialize(self):
        # 注册工具
        self.register_tool("my_tool", self.my_tool)
    
    async def my_tool(self, param: str) -> str:
        return f"处理: {param}"
    
    async def cleanup(self):
        pass

5. MCP支持

MCP (Model Context Protocol) 提供标准化的上下文访问接口。

使用内置MCP服务器

from pathlib import Path
from src.ai.mcp import MCPManager
from src.ai.mcp.servers import FileSystemMCPServer

# 创建MCP管理器
mcp_manager = MCPManager(config_path=Path("config/mcp.json"))

# 注册文件系统服务器
fs_server = FileSystemMCPServer(root_path=Path("data"))
await mcp_manager.register_server(fs_server)

# 获取所有工具
tools = await mcp_manager.get_all_tools_for_ai()

# 执行工具
result = await mcp_manager.execute_tool(
    "filesystem.read_file",
    {"path": "test.txt"}
)

创建自定义MCP服务器

from src.ai.mcp.base import MCPServer

class MyMCPServer(MCPServer):
    def __init__(self):
        super().__init__(name="my_server", version="1.0.0")
    
    async def initialize(self):
        self.register_tool(
            name="my_tool",
            description="我的工具",
            input_schema={
                "type": "object",
                "properties": {
                    "param": {"type": "string"}
                },
                "required": ["param"]
            },
            handler=self.my_tool_handler
        )
    
    async def my_tool_handler(self, param: str):
        return f"处理: {param}"

集成到QQ机器人

修改消息处理器

# src/handlers/message_handler.py
from src.ai import AIClient
from src.ai.base import ModelConfig, ModelProvider
from pathlib import Path

class MessageHandler:
    def __init__(self, bot):
        self.bot = bot
        
        # 初始化AI客户端
        config = ModelConfig(
            provider=ModelProvider.OPENAI,
            model_name="gpt-4",
            api_key=Config.AI_API_KEY,
            api_base=Config.AI_API_BASE
        )
        self.ai_client = AIClient(config, data_dir=Path("data/ai"))
        
        # 加载技能
        from src.ai.skills import SkillsManager
        self.skills = SkillsManager(Path("skills"))
        asyncio.create_task(self.skills.load_all_skills())
    
    async def handle_at_message(self, message: Message):
        try:
            # 提取用户消息
            user_message = message.content.strip()
            user_id = message.author.id
            
            # 调用AI
            response = await self.ai_client.chat(
                user_id=user_id,
                user_message=user_message,
                use_memory=True,
                use_tools=True
            )
            
            # 发送回复
            await message.reply(content=response)
            
        except Exception as e:
            logger.error(f"处理消息失败: {e}")
            await message.reply(content="抱歉，处理消息时出错了")

配置文件

config/personalities.json

{
  "default": {
    "name": "小助手",
    "description": "友好的AI助手",
    "traits": ["FRIENDLY", "PROFESSIONAL", "EMPATHETIC"],
    "speaking_style": "温和、清晰、有条理",
    "example_responses": [
      "我明白你的意思了，让我来帮你解决这个问题。",
      "这是个很好的问题！我来详细解释一下。"
    ],
    "custom_instructions": ""
  }
}

config/mcp.json

{
  "filesystem": {
    "enabled": true,
    "root_path": "data"
  }
}

高级用法

1. 流式响应

async for chunk in client.chat(
    user_id="user123",
    user_message="讲个故事",
    stream=True
):
    print(chunk, end='', flush=True)

2. 自定义工具

# 定义工具函数
async def search_web(query: str) -> str:
    # 实现搜索逻辑
    return f"搜索结果: {query}"

# 注册工具
client.register_tool(
    name="search_web",
    description="搜索网络信息",
    parameters={
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "搜索关键词"
            }
        },
        "required": ["query"]
    },
    function=search_web
)

3. 多模型切换

# 切换到Claude
claude_config = ModelConfig(
    provider=ModelProvider.ANTHROPIC,
    model_name="claude-3-5-sonnet-20241022",
    api_key="your_anthropic_key"
)
client.model = client._create_model(claude_config)

# 切换到DeepSeek
deepseek_config = ModelConfig(
    provider=ModelProvider.DEEPSEEK,
    model_name="deepseek-chat",
    api_key="your_deepseek_key",
    api_base="https://api.deepseek.com/v1"
)
client.model = client._create_model(deepseek_config)

性能优化

1. 记忆系统优化

# 调整短期记忆大小
client.memory.short_term.max_size = 30  # 默认20

# 调整记忆过期时间
from datetime import timedelta
client.memory.short_term.max_age = timedelta(hours=1)  # 默认30分钟

# 调整重要性阈值
client.memory.importance_threshold = 0.7  # 默认0.6

2. 并发任务处理

# 同时启动多个任务
tasks = []
for i in range(5):
    task_id = await client.create_long_task(...)
    tasks.append(client.start_task(task_id))

# 等待所有任务完成
await asyncio.gather(*tasks)

故障排查

常见问题

1. API调用失败

   • 检查API密钥是否正确
   • 检查网络连接
   • 检查API地址是否正确

2. 记忆系统不工作

   • 确保 use_memory=True
   • 检查数据目录权限
   • 查看日志文件

3. 技能加载失败

   • 检查 skill.json 格式
   • 确保 main.py 中有Skill子类
   • 查看错误日志

最佳实践

1. 合理使用记忆系统

   • 不要在每次对话都查询长期记忆
   • 定期清理不重要的记忆
   • 为重要对话添加元数据标记

2. 技能开发

   • 保持技能功能单一
   • 提供清晰的工具描述
   • 处理异常情况

3. 长任务设计

   • 将任务分解为合理的步骤
   • 提供进度反馈
   • 支持任务暂停和恢复

更新日志

v1.0.0 (2024-03-02)

• ✅ 初始版本发布
• ✅ 支持OpenAI、Anthropic、DeepSeek、Qwen
• ✅ 人格系统
• ✅ 记忆系统（短期+长期+RAG）
• ✅ 长任务处理
• ✅ Skills插件系统
• ✅ MCP支持

贡献

欢迎提交Issue和Pull Request！

许可证

MIT License