agent-redis-framework 1.1.1

A clean, elegant Python library for Redis-based multi-task scheduling with SortedSets and Streams.

Agent Redis Framework

一个优雅、高效的 Python Redis 框架，专为多任务调度与消息流处理而设计。提供基于 Redis Sorted Sets 的轻量任务队列和基于 Redis Streams 的消费组封装，适合构建可扩展的分布式任务系统。

✨ 核心特性

🚀 SortedSetQueue - 智能任务队列

• 优先级调度: 基于 Redis Sorted Set 实现的轻量任务队列
• 原子操作: 使用 ZPOPMIN/ZPOPMAX 确保多消费者场景下的任务安全分发
• 灵活排序: 支持按分数升序/降序弹出任务
• 失败处理: 内置任务处理失败的回调机制
• 批量处理: 支持一次性弹出并处理多个任务

📡 RedisStreamsClient - 流式消息处理

• 消费组管理: 完整的 Redis Streams 消费组封装
• 自动 ACK: 消息处理成功后自动确认
• 流量控制: 支持流长度限制和自动修剪
• 阻塞消费: 可配置的阻塞时间和批量消费
• 错误恢复: 处理待确认消息和消费者故障恢复

🔧 企业级特性

• 连接池管理: 自动连接池复用，优化高并发性能
• 环境配置: 灵活的 .env 文件和环境变量支持
• 类型安全: 完整的 Python 类型注解
• 线程安全: 所有核心组件支持多线程并发
• 零依赖: 仅依赖 redis 和 typing-extensions

🚀 快速开始

Redis 连接配置

环境变量配置

创建 .env 文件：

REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
REDIS_PASSWORD=your_password
REDIS_MAX_CONNECTIONS=20

from agent_redis_framework import get_redis

# 自动从环境变量加载配置
redis_client = get_redis()
print(redis_client.ping())  # True

SortedSetQueue - 任务队列示例

from agent_redis_framework import SortedSetQueue, Task, get_redis
import time

# 初始化
redis_client = get_redis()
queue = SortedSetQueue(redis_client=redis_client)
queue_key = "my_task_queue"

# 清空队列（可选）
queue.clear(queue_key)

# 推送任务（分数越小优先级越高）
import json
queue.push(queue_key, Task(id="urgent_task", payload=json.dumps({"priority": "high"})), score=1)
queue.push(queue_key, Task(id="normal_task", payload=json.dumps({"priority": "normal"})), score=5)
queue.push(queue_key, Task(id="delayed_task", payload=json.dumps({"priority": "low"})), score=10)

# 定义任务处理函数
def process_task(task: Task) -> bool:
    print(f"Processing task {task.id}: {task.payload}")
    # 模拟任务处理
    time.sleep(0.1)
    return True  # 返回 True 表示处理成功

# 定义失败处理函数
def handle_failure(task: Task) -> None:
    print(f"Task {task.id} failed, logging for retry...")

# 原子弹出并处理任务
processed = queue.pop_and_handle(
    queue_key,
    callback=process_task,
    on_failure=handle_failure,
    count=2  # 一次处理 2 个任务
)

print(f"Processed {len(processed)} tasks")
print(f"Remaining tasks: {queue.size(queue_key)}")

RedisStreamsClient - 流式消息示例

from agent_redis_framework import RedisStreamsClient, StreamMsg, get_redis
import threading
import time
import json

# 初始化
redis_client = get_redis()
stream_client = RedisStreamsClient(redis_client=redis_client)

stream_name = "user_events"
group_name = "analytics_group"
consumer_name = "consumer_1"

# 创建消费组（幂等操作）
stream_client.ensure_group(stream_name, group_name)

# 推送消息
msg = StreamMsg(stream=stream_name, payload=json.dumps({
    "event_type": "user_login",
    "user_id": "12345",
    "timestamp": time.time(),
    "ip_address": "192.168.1.100"
}))
msg_key = stream_client.push(msg)
print(f"Message pushed with ID: {msg_key}")

# 定义消息处理函数
def handle_message(msg: StreamMsg) -> None:
    print(f"Processing message {msg.msg_key} from {msg.stream}")
    print(f"Payload: {json.loads(msg.payload)}")
    # 处理业务逻辑
    time.sleep(0.1)

# 在单独线程中启动消费者
def start_consumer():
    stream_client.consume(
        streams=[stream_name],
        group=group_name,
        consumer=consumer_name,
        callback=handle_message,
        block_ms=5000,  # 5秒阻塞超时
        count=10        # 每次最多读取10条消息
    )

# 启动消费者线程
consumer_thread = threading.Thread(target=start_consumer, daemon=True)
consumer_thread.start()

# 继续推送更多消息
for i in range(5):
    msg = StreamMsg(stream=stream_name, payload=json.dumps({
        "event_type": "page_view",
        "user_id": f"user_{i}",
        "page": f"/page/{i}",
        "timestamp": time.time()
    }))
    stream_client.push(msg)
    time.sleep(1)

🏗️ 高级用法

延时任务调度

import time
import json
from agent_redis_framework import SortedSetQueue, Task, get_redis

queue = SortedSetQueue(redis_client=get_redis())
delay_queue = "delayed_tasks"

# 推送延时任务（5分钟后执行）
future_time = time.time() + 300  # 5分钟后
queue.push(delay_queue, Task(
    id="send_email",
    payload=json.dumps({"to": "[e-mail]", "subject": "Welcome!"})
), score=future_time)

# 处理到期任务
def process_if_ready(task: Task) -> bool:
    current_time = time.time()
    # 检查任务是否到期（这里需要从Redis重新获取分数）
    return True  # 简化示例

# 定期检查并处理到期任务
while True:
    processed = queue.pop_and_handle(delay_queue, process_if_ready, count=5)
    if not processed:
        time.sleep(10)  # 没有任务时等待10秒

流消息批量处理

from agent_redis_framework import RedisStreamsClient, StreamMsg, get_redis
import json

stream_client = RedisStreamsClient(redis_client=get_redis())

# 批量推送消息
messages = [
    {"order_id": f"order_{i}", "amount": i * 100, "status": "pending"}
    for i in range(100)
]

for msg in messages:
    msg = StreamMsg(stream="order_stream", payload=json.dumps(msg))
    stream_client.push(msg, maxlen=1000)  # 限制流长度

# 批量消费处理
def batch_process_orders(msg: StreamMsg) -> None:
    order_data = json.loads(msg.payload) if msg.payload else {}
    print(f"Processing order: {order_data.get('order_id', 'unknown')}")
    # 批量处理逻辑

stream_client.consume(
    streams=["order_stream"],
    group="order_processors",
    consumer="processor_1",
    callback=batch_process_orders,
    count=50,  # 每次批量处理50条消息
    block_ms=1000
)

🔧 配置选项

RedisConfig 完整配置

from agent_redis_framework import RedisConfig

config = RedisConfig(
    host="localhost",                    # Redis 主机地址
    port=6379,                          # Redis 端口
    db=0,                               # 数据库编号
    username="myuser",                  # 用户名（Redis 6.0+）
    password="mypassword",              # 密码
    ssl=False,                          # 是否使用 SSL
    socket_timeout=30.0,                # Socket 超时时间
    socket_connect_timeout=10.0,        # 连接超时时间
    health_check_interval=30,           # 健康检查间隔（秒）
    max_connections=20,                 # 连接池最大连接数
    retry_on_timeout=True               # 超时时是否重试
)

环境变量配置

支持的环境变量：

# 基础连接配置
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
REDIS_USERNAME=myuser
REDIS_PASSWORD=mypassword
REDIS_SSL=false

# 连接池配置
REDIS_MAX_CONNECTIONS=20
REDIS_SOCKET_TIMEOUT=30.0
REDIS_SOCKET_CONNECT_TIMEOUT=10.0
REDIS_HEALTH_CHECK_INTERVAL=30
REDIS_RETRY_ON_TIMEOUT=true

🧪 测试和开发

项目提供了完整的测试 API，方便开发和调试：

# 安装开发依赖
uv sync --extra dev

# 启动测试 API 服务
UV_INDEX_URL=https://pypi.org/simple/ uv run uvicorn tests.test_api:app --host 0.0.0.0 --port 8081

访问 http://localhost:8081/docs 查看交互式 API 文档。

详细的测试说明请参考 tests/README.md。

🏭 生产环境最佳实践

1. 连接池优化

# 根据并发需求调整连接池大小
config = RedisConfig(
    max_connections=50,  # 高并发场景
    health_check_interval=30,  # 定期健康检查
    socket_timeout=10.0,  # 避免长时间阻塞
    retry_on_timeout=True  # 自动重试
)

2. 错误处理和监控

import logging
from agent_redis_framework import SortedSetQueue, Task

logger = logging.getLogger(__name__)

def robust_task_handler(task: Task) -> bool:
    try:
        # 任务处理逻辑
        process_business_logic(task.payload)
        return True
    except Exception as e:
        logger.error(f"Task {task.id} failed: {e}")
        return False

def failure_handler(task: Task) -> None:
    # 记录失败任务，可能需要重试
    logger.warning(f"Task {task.id} marked for retry")
    # 可以将失败任务重新入队或发送到死信队列

3. 优雅关闭

import signal
import threading
from agent_redis_framework import RedisStreamsClient

class GracefulConsumer:
    def __init__(self):
        self.running = True
        self.stream_client = RedisStreamsClient()
        
    def signal_handler(self, signum, frame):
        print("Received shutdown signal, stopping consumer...")
        self.running = False
        
    def consume_loop(self):
        while self.running:
            try:
                # 使用较短的阻塞时间以便及时响应关闭信号
                self.stream_client.consume(
                    streams=["my_stream"],
                    group="my_group",
                    consumer="my_consumer",
                    callback=self.handle_message,
                    block_ms=1000  # 1秒超时
                )
            except KeyboardInterrupt:
                break
                
    def handle_message(self, msg):
        if not self.running:
            return
        # 处理消息逻辑
        
# 使用示例
consumer = GracefulConsumer()
signal.signal(signal.SIGINT, consumer.signal_handler)
signal.signal(signal.SIGTERM, consumer.signal_handler)
consumer.consume_loop()

📚 API 参考

SortedSetQueue

• push(key, task, score=None) - 推送任务到队列
• pop_and_handle(key, callback, on_failure=None, ascending=True, count=1) - 原子弹出并处理任务
• size(key) - 获取队列大小
• clear(key) - 清空队列

RedisStreamsClient

• push(msg, maxlen=None) - 推送消息到流（msg 为 StreamMsg，payload 建议为 JSON 字符串）
• ensure_group(stream, group, id='$') - 确保消费组存在
• consume(streams, group, consumer, callback, block_ms=5000, count=1) - 消费消息

数据类

• Task(id, payload, meta={}) - 任务数据类，payload 为字符串格式（JSON 或原生字符串）
• StreamMsg(stream, msg_key, payload, meta={}) - 流消息数据类，payload 为字符串格式
• RedisConfig(...) - Redis 连接配置

🔗 相关链接

• Redis 官方文档
• Redis Streams 指南
• Redis Sorted Sets 指南