ylbot 1.0.4

基于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())

配置文件配置（推荐）

ylbot支持使用 .env 文件进行配置，适用于所有操作系统（Windows/Linux/macOS）。

1. 创建配置文件

在项目根目录下创建 .env 文件，内容如下：

# ==================== 历史记录配置 ====================
# 单条消息最大字符数
MAX_SINGLE_CHAR=1000

# 每个用户最大记录数
MAX_USER_RECORD_COUNT=50

# 历史记录总字符数限制
MAX_HISTORY_CONTENT_CHAR=4000

# 清理旧记录的周期（天）
CLEAR_CYCLE=7

# 历史记录算法：取值为 "fair"（公平算法）或 "normal"（普通算法，默认）
HISTORY_ALGORITHM=normal

# ==================== 状态管理配置 ====================
# 状态清理间隔（秒）
STATUS_CLEANUP_INTERVAL=3600

# ==================== 自动回复配置 ====================
# 自动回复超时时间（秒）
AUTO_ANSWER_TIMEOUT=10

# ==================== LLM配置 ====================
# LLM API密钥（使用Ollama时设置为"ollama"）
LLM_API_KEY=ollama

# LLM API基础URL（Ollama默认地址）
LLM_BASE_URL=http://localhost:11434/v1

# LLM模型名称
LLM_MODEL=qwen3:latest

# ==================== Redis配置 ====================
# Redis连接地址
REDIS_URL=redis://localhost:6379/0

2. 自动创建配置文件

如果你没有创建 .env 文件，ylbot会在首次运行时自动创建一个基本的配置文件，并提示你根据实际情况修改。

3. 环境变量覆盖

你也可以使用系统环境变量来覆盖 .env 文件中的配置，优先级为：代码参数 > 环境变量 > .env文件 > 默认值。

环境变量配置（备用）

如果你仍然希望使用环境变量，可以设置以下变量：

# Windows (PowerShell)
$env:MAX_SINGLE_CHAR=1000
$env:MAX_USER_RECORD_COUNT=50
$env:MAX_HISTORY_CONTENT_CHAR=4000
$env:CLEAR_CYCLE=7
$env:LLM_API_KEY="ollama"
$env:LLM_BASE_URL="http://localhost:11434/v1"
$env:LLM_MODEL="qwen3:latest"
$env:AUTO_ANSWER_TIMEOUT=10

# Linux/macOS
export MAX_SINGLE_CHAR=1000
export MAX_USER_RECORD_COUNT=50
export MAX_HISTORY_CONTENT_CHAR=4000
export CLEAR_CYCLE=7
export LLM_API_KEY=ollama
export LLM_BASE_URL=http://localhost:11434/v1
export LLM_MODEL=qwen3:latest
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。