跳转至

常见问题

快速问题

Q: 如何保证任务不丢失？

A: Redis Streams 提供持久化机制。Worker 使用 XACK 确认消息处理完成，未确认的消息会被重新投递。

Q: 如何实现 Worker 负载均衡？

A: 多个 Worker 连接同一个 Redis Stream，Redis 会自动在消费者组内进行负载分配。

Q: 如何横向扩展 Worker？

A: 启动多个不同 worker_id 的 Worker 进程，共享同一个 Redis 实例和相同的 target_agent_type stream。

Q: 消息处理失败后会自动重试吗？

A: 是的，未 ACK 的消息会在重新读取时被重新投递。但建议在 processCommand 中实现重试逻辑。

高级问题

Q: 如何实现定向消息发送？

A: 传入 target_worker_id 参数，消息会写入 worker:{worker_id} stream：

response = await client.send_message(
    target_agent_type="my_agent",
    session_id="sess_123",
    content="hello",
    target_worker_id="specific-worker-01",
)

SendResponse response = client.sendMessage(
    "my_agent", "sess_123", "hello",
    null, null, null,
    "specific-worker-01",  // targetWorkerId
    null, null, null, null
);

const response = await client.sendMessage({
    targetAgentType: "my_agent",
    sessionId: "sess_123",
    content: "hello",
    targetWorkerId: "specific-worker-01",
});

Q: 如何处理人机交互场景？

A: 使用 context.askUser() 挂起任务，等待用户回复：

result = await context.ask_user(
    AskUserEvent(prompt="请确认")
)
# 任务会挂起，用户回复后以 ResumeCommand 形式恢复

Object result = context.askUser("请确认");
// 任务会挂起，用户回复后以 ResumeCommand 形式恢复

const result = await context.askUser({ prompt: "请确认" });
// 任务会挂起，用户回复后以 ResumeCommand 形式恢复

Q: 如何监控 Worker 状态？

workers = await registry.get_online_workers(agent_type="my_agent")
for worker in workers:
    print(f"Worker: {worker['worker_id']}, Last seen: {worker['last_heartbeat']}")

List<Map<String, Object>> workers = registry.getOnlineWorkers("my_agent");
for (Map<String, Object> worker : workers) {
    System.out.println("Worker: " + worker.get("worker_id"));
}

const workers = await registry.getOnlineWorkers("my_agent");
for (const worker of workers) {
    console.log(`Worker: ${worker.worker_id}`);
}

Q: 如何配置插件超时？

run_worker(
    worker_class=MyWorker,
    plugin_hook_timeout_seconds=30,
)

runWorker(MyWorker, {
    pluginHookTimeoutSeconds: 30,
});

Q: 如何使用服务发现？

from by_framework.core.discovery import ServiceRegistry

registry = ServiceRegistry(redis_client=redis)
await registry.register("my-service", "192.168.1.100", 8080)

ServiceRegistry registry = new ServiceRegistry(redisClient);
registry.register("my-service", "192.168.1.100", 8080);

const registry = new ServiceRegistry(redis);
await registry.register({
    serviceName: "my-service",
    host: "192.168.1.100",
    port: 8080,
});

部署 / Redis Cluster

Q: by-framework 能不能接 Redis Cluster，而不是单机 Redis？

A: 可以，三个 SDK 都已经完整支持 Redis Cluster 部署，业务代码不需要改动。只需要设置一个环境变量 REDIS_CLUSTER_HOST（逗号分隔的 host:port 节点列表），就会自动切换部署形态并适配 key 格式，详见 部署指南 · 集群部署。

Q: 为什么切到 Redis Cluster 后报 CROSSSLOT Keys in request don't hash to the same slot？

A: 大概率是 key schema 版本没有切到 v2。Redis Cluster 下多 key 操作必须落在同一个 slot，而默认的 v1 key 格式没有做 这个保证。检查环境变量：如果只设置了 REDIS_CLUSTER_NODES（旧变量名）而不是 REDIS_CLUSTER_HOST，或者显式设置了 REDIS_MODE=cluster + REDIS_KEY_SCHEMA_VERSION=v1，都不会自动推断出 v2，需要手动加上 REDIS_KEY_SCHEMA_VERSION=v2（或者干脆改用 REDIS_CLUSTER_HOST，它会自动带出 v2）。

如果确认环境变量都配对了，但报错的是自己业务代码里手写的自定义 Redis key（不是 SDK 内置的 key），说明这些自定义 key 之间存在跨 slot 的多 key 操作（比如把它们放进同一个 MULTI/EXEC 或 Lua 脚本），需要给这组 key 加上统一的 hash tag，比如 custom:{tenant_id}:a / custom:{tenant_id}:b 这种写法，两个 key 只要花括号里的内容相同就必定落在同一个 slot。

Q: 切换到 Redis Cluster / v2 key schema 后，之前的数据去哪了？

A: v2 key 的物理名字和 v1 完全不同（比如 byai_gateway:registry:worker:online:{worker_id} 变成 byai_gateway:v2:registry:worker:{worker_id}:online），旧 key 下的数据在新代码里读不到，但也没有被删除——原样留在 Redis 里，跟着自己的 TTL 自然过期。绝大多数 key 本身就是运行时状态（会话数据、心跳、任务组进度等），带 7 天以内的 TTL，直接部署新版本、等旧 key 过期即可，不需要迁移脚本。详见 Redis Streams 深度剖析 · Key Schema 版本。

Q: 一定要同时设置 REDIS_MODE=cluster 和 REDIS_KEY_SCHEMA_VERSION=v2 吗？

A: 不需要。只设置 REDIS_CLUSTER_HOST 就够了，SDK 会自动推断出这两项——这是较新版本才有的简化，如果你看到的示例代码 或旧文档里还写着"两个变量都要设"，那是升级前的写法，现在仍然兼容（显式设置的值优先于自动推断），但不再是必须的。