跳转至

by-framework-python

by-framework 是一个基于 Redis Streams 构建的分布式 Agent 调度框架，提供 Worker 编排、会话级运行时状态与插件化能力注册，适合构建 AI Agent 系统。

核心特性

  • 原生异步: 基于 Python asyncio 构建，完美契合 I/O 密集型 Agent 任务
  • 高度插件化: 通过插件和 AgentConfig 注册工具、提示词、技能与生命周期钩子
  • 状态管控: 完善的 AgentContext 支持，轻松实现流式输出、状态流转和附件处理
  • 解耦架构: 采用控制流与数据流分离设计，便于 Worker 横向扩展
  • Agent Type 路由: Worker 通过 get_agent_types() 声明支持的 agent type，并参与路由与在线校验

快速开始

# 安装依赖
cd by-framework-python
uv sync

# 启动 Redis
docker run -d -p 6379:6379 redis:7-alpine

# 创建 Worker
cat > my_agent.py << 'EOF'
import asyncio
from by_framework import GatewayWorker, AgentContext, run_worker

class MyAssistant(GatewayWorker):
    def get_agent_types(self):
        return ["chat_agent"]

    async def process_command(self, command, context: AgentContext):
        await context.emit_chunk("正在处理...\n")
        user_input = command.content if isinstance(command.content, str) else str(command.content)
        await context.emit_chunk(f"收到: {user_input}")
        return {"status": "success"}
EOF

# 启动 Worker
uv run python my_agent.py

架构概览

Client → Redis Input MQ → GatewayWorker → Redis Data Stream → WebSocket Backend

文档导航

章节 内容
快速入门 环境搭建与第一个 Worker
Worker 开发指南 深入理解 GatewayWorker
插件系统 扩展框架能力
客户端开发 发送任务与结果获取
架构设计 系统架构深度剖析
API 参考 完整的 API 文档