ylbot 1.0.1

基于ylbot机器人标准的Python聊天机器人框架，支持历史记录管理、状态管理和多平台接口

ylbot

基于ylbot机器人标准的Python聊天机器人框架，支持历史记录管理、状态管理和多平台接口。

特性

• ✅ 历史记录管理: 支持公平算法和普通算法的历史记录大小限制
• ✅ 状态管理: 用户状态管理，防止重复处理
• ✅ 回复构建: 自动格式化回复，支持@前缀添加
• ✅ 多模态支持: 支持文本、图片、语音、视频等多种消息类型
• ✅ 配置灵活: 通过环境变量或代码配置所有参数
• ✅ 异步支持: 纯异步操作，支持高并发
• ✅ LLM集成: 默认集成OpenAI API，支持自定义LLM回调

安装

pip install ylbot

或者从源码安装：

git clone https://github.com/bifu123/ylbot.git
cd ylbot
pip install -e .

快速开始

基本使用

import asyncio
from ylbot import YLBot, Config

async def main():
    # 创建配置（可以从环境变量加载）
    config = Config.from_env()

    # 创建机器人实例
    bot = YLBot(config)

    # 构建查询消息
    query_message = {
        "message_id": "msid2025111345682",
        "source_id": "413135222",
        "name_space": "default",
        "method": "private",
        "bot": {
            "bot_id": "346016822",
            "bot_nick_name": "元龙令"
        },
        "from_user": {
            "user_id": "862354",
            "user_nick_name": "王五"
        },
        "message": [
            {
                "type": "text",
                "data": {"text": "你好，今天天气怎么样？"}
            }
        ],
        "raw_message": "你好，今天天气怎么样？",
        "method_allow": ["private", "group_at"],
        "must_answer": "yes"
    }

    # 处理查询
    response = await bot.process_query(query_message)
    print(f"AI回复: {response['answer']}")

# 运行异步函数
asyncio.run(main())

环境变量配置

# 历史记录配置
export MAX_SINGLE_CHAR=1000
export MAX_USER_RECORD_COUNT=50
export MAX_HISTORY_CONTENT_CHAR=4000
export CLEAR_CYCLE=7

# LLM配置
export LLM_API_KEY=sk-xxx
export LLM_BASE_URL=https://api.openai.com/v1
export LLM_MODEL=gpt-3.5-turbo

# 自动回复配置
export AUTO_ANSWER_TIMEOUT=10

高级用法

自定义LLM回调

import asyncio
from ylbot import YLBot, Config

# 同步LLM回调
def custom_llm_callback_sync(prompt: str, **kwargs) -> str:
    # 调用自定义LLM服务
    # 例如使用本地模型、其他API等
    return "这是自定义LLM的回复"

# 异步LLM回调
async def custom_llm_callback_async(prompt: str, **kwargs) -> str:
    # 异步调用自定义LLM服务
    return "这是异步自定义LLM的回复"

config = Config()
# 使用同步回调
bot_sync = YLBot(config, llm_callback=custom_llm_callback_sync)
# 使用异步回调  
bot_async = YLBot(config, llm_callback=custom_llm_callback_async)

使用历史记录管理器

import asyncio
from ylbot import HistoryManager, Config

async def main():
    config = Config()
    history_manager = HistoryManager(config)

    # 添加历史记录
    await history_manager.add_record(
        source_id="wx_001",
        user_id="user_123",
        user="张三",
        content="你好"
    )

    # 获取历史记录
    history = await history_manager.get_history(
        source_id="wx_001",
        user_id="user_123",
        bot_nick_name="助手",
        question="今天天气怎么样？"
    )
    print(f"历史记录: {history}")

asyncio.run(main())

演示示例

1. 最简演示 (demo_lite.py)

这是一个最简示例，展示了如何使用最少的代码实现用户问题"2+2=?"的回复。

功能特点：

• 配置管理（环境变量和代码配置）
• 异步LLM回调（连接Ollama）
• Redis状态和历史记录存储
• 高级定制参数应用

运行方式：

cd ylbot
python demo_lite.py

代码摘要：

import asyncio
from ylbot import YLBot, Config

async def main():
    # 创建配置
    config = Config.from_env()
    
    # 创建机器人实例
    bot = YLBot(config)
    
    # 构建查询消息
    query_message = {
        "message_id": "lite_001",
        "source_id": "lite_source",
        "name_space": "default",
        "method": "private",
        "bot": {"bot_id": "lite_bot", "bot_nick_name": "精简演示机器人"},
        "from_user": {"user_id": "lite_user", "user_nick_name": "演示用户"},
        "message": [{"type": "text", "data": {"text": "2+2=?"}}],
        "raw_message": "2+2=?",
        "method_allow": ["private", "group_at"],
        "must_answer": "yes"
    }
    
    # 处理查询
    response = await bot.process_query(query_message)
    print(f"回复内容: {response['answer']}")

asyncio.run(main())

2. 流式输出演示 (demo_streaming.py)

这个示例展示了ylbot的流式输出功能和Agent工具调用。

功能特点：

• Web流式输出（实时逐字输出）
• Onebot流式输出（显示思考过程）
• Agent工具调用（计算器、获取时间等）
• 历史记录管理和Redis状态存储

运行方式：

cd ylbot
python demo_streaming.py

演示内容：

1. Web流式输出 - 实时逐字输出，适合Web界面
2. Onebot流式输出 - 显示思考过程，适合聊天机器人
3. Agent工具调用 - 支持自定义工具，扩展性强
4. 历史记录和Redis - 数据持久化，支持会话管理

代码摘要（Agent工具调用）：

import asyncio
from ylbot import YLBot, Config

async def main():
    # 自定义工具
    async def get_weather(city: str) -> str:
        return f"{city}的天气：晴，25°C，湿度60%"
    
    async def search_web(query: str) -> str:
        return f"搜索'{query}'的结果：相关文章3篇，最新更新2025年1月"
    
    # 创建配置和机器人
    config = Config.from_env()
    bot = YLBot(config)
    
    # 构建查询消息
    query_message = {
        "message_id": "stream_001",
        "source_id": "demo",
        "method": "private",
        "bot": {"bot_nick_name": "演示机器人"},
        "from_user": {"user_id": "user1", "user_nick_name": "用户"},
        "message": [{"type": "text", "data": {"text": "请帮我查一下北京的天气"}}],
        "raw_message": "请帮我查一下北京的天气",
        "method_allow": ["private", "group_at"],
        "must_answer": "yes"
    }
    
    # 使用Agent流式输出（带自定义工具）
    async for token in bot.process_query_stream(
        query_message,
        source="onebot",
        use_agent=True,
        agent_tools=[get_weather, search_web]
    ):
        print(token, end="", flush=True)

asyncio.run(main())

输出示例：

🔧 我需要使用工具: get_weather({'city': '北京'})
😊 📤 执行工具结果: 北京的天气：晴，25°C，湿度60%
😊 根据天气工具查询，北京今日天气晴朗，气温25°C，空气湿度60%，适宜外出活动。

3. 完整演示 (demo.py)

除了上述两个专门演示，项目还包含一个完整的演示文件 demo.py，展示了ylbot的所有核心功能。

运行方式：

cd ylbot
python demo.py

使用状态管理器

import asyncio
from ylbot import StatusManager, Config

async def main():
    config = Config()
    status_manager = StatusManager(config)

    # 设置用户状态
    await status_manager.set_llm_status("wx_001", "user_123", "busy")

    # 检查用户是否忙碌
    if await status_manager.is_user_busy("wx_001", "user_123"):
        print("用户正在处理中...")

    # 清理资源
    await status_manager.close()

asyncio.run(main())

配置选项

历史记录配置

• MAX_SINGLE_CHAR: 单条消息最大字符数（默认: 1000）
• MAX_USER_RECORD_COUNT: 每个用户最大记录数（默认: 50）
• MAX_HISTORY_CONTENT_CHAR: 历史记录总字符数限制（默认: 4000）
• CLEAR_CYCLE: 清理旧记录的周期（天）（默认: 7）
• HISTORY_ALGORITHM: 历史记录算法，取值为 "fair"（公平算法）或 "normal"（普通算法，默认）

状态管理配置

• STATUS_CLEANUP_INTERVAL: 状态清理间隔（秒）（默认: 3600）

自动回复配置

• AUTO_ANSWER_TIMEOUT: 自动回复超时时间（秒）（默认: 10）

LLM配置

• LLM_API_KEY: OpenAI API密钥
• LLM_BASE_URL: API基础URL
• LLM_MODEL: 模型名称（默认: deepseek-chat）

接口规范

查询消息格式

query_message = {
    "message_id": "消息ID",
    "source_id": "来源ID",
    "name_space": "命名空间",
    "method": "聊天类型",  # private/group_at/group
    "bot": {
        "bot_id": "机器人ID",
        "bot_nick_name": "机器人昵称"
    },
    "from_user": {
        "user_id": "用户ID",
        "user_nick_name": "用户昵称"
    },
    "message": [
        {
            "type": "消息类型",  # text/image/voice/video
            "data": {"text": "消息内容"}  # 或其他媒体数据
        }
    ],
    "raw_message": "原始消息",
    "method_allow": ["允许的聊天类型"],  # 默认["private", "group_at"]
    "must_answer": "回复模式"  # yes/no/auto
}

回复消息格式

response_message = {
    "query_message": query_message,
    "to_users": [用户列表],
    "messages": [
        {
            "type": "消息类型",
            "data": {"text": "回复内容"}
        }
    ],
    "answer": "原始回复内容"
}

开发

安装开发依赖

pip install -e .[dev]

运行测试

pytest

构建包

python -m build

许可证

MIT License

贡献

欢迎提交Issue和Pull Request！

支持

如有问题，请查看文档或提交Issue。