Agent Queue 客户端使用指南

Agent Queue 是一个基于 Redis + Celery 的私有化任务队列系统，通过 gRPC 接口提供服务。本文档重点介绍客户端封装类的使用方法，提供简洁易用的 Python API。

目录

• 快速开始
• 安装
• 客户端 API 使用 ⭐
  • PrivateAgentTasksQueue - 基础客户端
  • AgentQueue - 针对特定 Agent 的队列
  • AgentQueueManager - 任务分配管理器
  • AgentQueueReportClient - 实时报表客户端
• 查看接口定义 ⭐
• API 参考
• 完整示例

快速开始

方式一：使用 PrivateAgentTasksQueue（推荐）⭐

from agent_queues import PrivateAgentTasksQueue, DATA_PROCESSING, COMPLETED
import json

# 1. 创建客户端（使用默认配置）
client = PrivateAgentTasksQueue()

# 2. 创建队列
agent_id = "my_agent_001"
client.create_queue(agent_id)

# 3. 提交任务
response = client.submit_task(
    agent_id=agent_id,
    task_type=DATA_PROCESSING,
    payload=json.dumps({"data": "test"}),
    priority=9  # 高优先级
)
task_id = response.task_id

# 4. 获取任务
task_response = client.get_task(agent_id=agent_id, timeout=5)
if task_response.success:
    task = task_response.task
    print(f"获取到任务: {task.task_id}")
    
    # 5. 处理任务...
    
    # 6. 更新任务状态
    client.update_task_status(
        task_id=task.task_id,
        agent_id=agent_id,
        status=COMPLETED,
        result=json.dumps({"result": "success"})
    )

方式二：使用 AgentQueue（更简洁）⭐

from agent_queues import PrivateAgentTasksQueue, DATA_PROCESSING, COMPLETED
import json

# 1. 创建客户端并获取队列对象
client = PrivateAgentTasksQueue()
queue = client.get_queue("my_agent_001", create_if_not_exists=True)

# 2. 提交任务（不需要再指定 agent_id）
response = queue.submit_task(
    task_type=DATA_PROCESSING,
    payload=json.dumps({"data": "test"}),
    priority=9
)

# 3. 获取任务
task_response = queue.get_task(timeout=5)
if task_response.success:
    task = task_response.task
    
    # 4. 更新任务状态
    queue.update_task_status(
        task_id=task.task_id,
        status=COMPLETED,
        result=json.dumps({"result": "success"})
    )

安装

从 PyPI 安装（推荐）

# 使用 pip 安装（会自动安装依赖）
pip install queues-manager

# 或使用 Poetry 安装（会自动安装依赖）
poetry add queues-manager

说明：

• ✅ 两种方式都会自动安装所有必需的依赖
• ✅ Poetry 完全支持，依赖会自动添加到 pyproject.toml 和 poetry.lock
• ✅ 依赖包括：grpcio, grpcio-tools, protobuf, pyyaml

依赖要求

• Python >= 3.12
• grpcio >= 1.60.0
• protobuf >= 4.25.0
• pyyaml >= 6.0.0

客户端 API 使用

1. PrivateAgentTasksQueue - 基础客户端

PrivateAgentTasksQueue 是核心客户端类，封装了所有 gRPC 接口，提供简洁的 Python API。

创建客户端

from agent_queues import PrivateAgentTasksQueue, create_client

# 方式1：使用默认配置（推荐）
client = PrivateAgentTasksQueue()

# 方式2：自定义配置
client = PrivateAgentTasksQueue(
    grpc_host="192.168.1.100",
    grpc_port=50052,
    grpc_timeout=60
)

# 方式3：使用配置文件
client = PrivateAgentTasksQueue(config_path="/path/to/config.yaml")

# 方式4：使用便捷函数
client = create_client(grpc_host="192.168.1.100", grpc_port=50052)

队列管理接口

# 创建队列
response = client.create_queue(agent_id="agent_001")

# 检查队列是否存在
response = client.queue_exists(agent_id="agent_001")
if response.exists:
    print("队列存在")

# 获取队列信息
info = client.get_queue_info(agent_id="agent_001")
print(f"待处理: {info.pending_count}")
print(f"处理中: {info.processing_count}")
print(f"已完成: {info.completed_count}")
print(f"失败: {info.failed_count}")
print(f"总计: {info.total_count}")

# 清空队列（按状态）
from agent_queues import FAILED
response = client.clear_queue(agent_id="agent_001", status=0, confirm=True)  # 清空所有
response = client.clear_queue(agent_id="agent_001", status=FAILED, confirm=True)  # 只清空失败的

# 删除队列
response = client.delete_queue(agent_id="agent_001", confirm=True)

任务操作接口

from agent_queues import DATA_PROCESSING, COMPLETED, FAILED
import json

# 提交任务
response = client.submit_task(
    agent_id="agent_001",
    task_type=DATA_PROCESSING,
    payload=json.dumps({"data": "test"}),
    custom_task_type="",  # 当 task_type 为 CUSTOM 时使用
    priority=9,  # 优先级 1-10，数字越大优先级越高
    client_request_id="unique_id_123"  # 幂等性保证
)
task_id = response.task_id

# 批量提交任务
from agent_queues import SubmitTaskItem
tasks = [
    SubmitTaskItem(
        type=DATA_PROCESSING,
        payload=json.dumps({"id": i}),
        priority=5
    )
    for i in range(10)
]
response = client.batch_submit_tasks(agent_id="agent_001", tasks=tasks)

# 获取任务
task_response = client.get_task(agent_id="agent_001", timeout=5)  # timeout=0 表示不等待
if task_response.success:
    task = task_response.task
    print(f"任务ID: {task.task_id}")
    print(f"优先级: {task.priority}")
    print(f"状态: {task.status}")

# 更新任务状态
client.update_task_status(
    task_id=task_id,
    agent_id="agent_001",
    status=COMPLETED,
    result=json.dumps({"result": "success"}),
    error_message=""  # 失败时填写错误信息
)

# 查询任务
response = client.query_task(task_id=task_id, agent_id="agent_001")
if response.success:
    task = response.task

# 删除任务
client.delete_task(task_id=task_id, agent_id="agent_001")

# 批量删除任务
task_ids = ["task_1", "task_2", "task_3"]
response = client.batch_delete_tasks(
    agent_id="agent_001",
    task_ids=task_ids,
    status=0  # 0 表示不限制状态
)

# 取消任务
client.cancel_task(task_id=task_id, agent_id="agent_001", reason="用户取消")

# 重试任务
response = client.retry_task(
    task_id=task_id,
    agent_id="agent_001",
    client_request_id="retry_unique_id"  # 幂等性保证
)

任务查询接口

from agent_queues import PENDING

# 列出任务
response = client.list_tasks(
    agent_id="agent_001",
    status=0,  # 0 表示所有状态
    limit=100,
    offset=0
)
if response.success:
    for task in response.tasks:
        print(f"任务ID: {task.task_id}, 状态: {task.status}")

# 获取任务统计信息
stats = client.get_task_stats(agent_id="agent_001", status=0)
print(f"总任务数: {stats.total_count}")
print(f"待处理: {stats.pending_count}")
print(f"处理中: {stats.processing_count}")
print(f"已完成: {stats.completed_count}")
print(f"失败: {stats.failed_count}")

# 获取某个状态的任务
response = client.get_tasks_by_status(
    agent_id="agent_001",
    status=PENDING,  # 必须指定状态，不能为 0
    limit=100,
    offset=0
)

获取队列对象

# 获取队列对象（如果不存在则自动创建）
queue = client.get_queue("agent_001", create_if_not_exists=True)

# 使用队列对象进行操作（见 AgentQueue 部分）

上下文管理器

# 使用上下文管理器自动关闭连接
with PrivateAgentTasksQueue() as client:
    response = client.create_queue("agent_001")
    # 自动关闭连接

2. AgentQueue - 针对特定 Agent 的队列

AgentQueue 封装了针对特定 agent 的队列操作，无需重复指定 agent_id。

获取队列对象

from agent_queues import PrivateAgentTasksQueue

client = PrivateAgentTasksQueue()

# 方式1：获取队列（如果不存在则自动创建）
queue = client.get_queue("agent_001", create_if_not_exists=True)

# 方式2：先检查是否存在
if client.queue_exists("agent_001").exists:
    queue = client.get_queue("agent_001")
else:
    client.create_queue("agent_001")
    queue = client.get_queue("agent_001")

队列管理方法

# 获取队列信息
info = queue.get_info()
print(f"待处理: {info.pending_count}, 处理中: {info.processing_count}")

# 检查队列是否存在
exists = queue.exists()

# 清空队列
queue.clear(status=0, confirm=True)  # 清空所有
queue.clear(status=FAILED, confirm=True)  # 只清空失败的

# 删除队列
queue.delete(confirm=True)

任务操作方法

from agent_queues import DATA_PROCESSING, COMPLETED

# 提交任务（不需要 agent_id）
response = queue.submit_task(
    task_type=DATA_PROCESSING,
    payload=json.dumps({"data": "test"}),
    custom_task_type="",
    priority=9,
    client_request_id="unique_id"
)

# 批量提交任务
from agent_queues import SubmitTaskItem
tasks = [
    SubmitTaskItem(type=DATA_PROCESSING, payload=json.dumps({"id": i}))
    for i in range(10)
]
response = queue.batch_submit_tasks(tasks)

# 获取任务（不需要 agent_id）
task_response = queue.get_task(timeout=5)

# 查询任务（不需要 agent_id）
task = queue.query_task(task_id)

# 更新任务状态（不需要 agent_id）
queue.update_task_status(
    task_id=task_id,
    status=COMPLETED,
    result=json.dumps({"result": "success"}),
    error_message=""
)

# 删除任务（不需要 agent_id）
queue.delete_task(task_id)

# 批量删除任务
queue.batch_delete_tasks(task_ids=["task_1", "task_2"], status=0)

# 取消任务
queue.cancel_task(task_id, reason="用户取消")

# 重试任务
queue.retry_task(task_id, client_request_id="retry_id")

任务查询方法

# 列出任务
tasks = queue.list_tasks(status=0, limit=100, offset=0)

# 获取任务统计
stats = queue.get_task_stats(status=0)

# 获取某个状态的任务
pending_tasks = queue.get_tasks_by_status(status=PENDING, limit=100, offset=0)

3. AgentQueueManager - 任务分配管理器

AgentQueueManager 提供高级任务分配功能，支持负载均衡和任务分发。如果目标 agent 的队列不存在，会自动创建队列后再分配任务。

创建管理器

from agent_queues import AgentQueueManager, create_manager

# 方式1：使用默认配置
manager = AgentQueueManager()

# 方式2：使用便捷函数
manager = create_manager(grpc_host="192.168.1.100", grpc_port=50052)

# 方式3：使用已有的客户端
from agent_queues import PrivateAgentTasksQueue
client = PrivateAgentTasksQueue()
manager = AgentQueueManager(queue_client=client)

分配任务到指定 Agent

from agent_queues import AgentQueueManager, DATA_PROCESSING
import json

manager = AgentQueueManager()

# 分配单个任务（如果队列不存在会自动创建）
result = manager.assign_task(
    target_agent_id="agent_001",
    task_type=DATA_PROCESSING,
    payload={"data": "test"},  # 可以是字典或 JSON 字符串
    custom_task_type="",
    priority=9,
    client_request_id="unique_id",
    ensure_queue_exists=True  # 默认 True，自动创建队列
)

if result["success"]:
    print(f"任务已分配，任务ID: {result['task_id']}")
else:
    print(f"分配失败: {result['message']}")

批量分配任务

# 批量分配任务到指定 agent
tasks = [
    {
        "type": DATA_PROCESSING,
        "payload": {"data": f"task_{i}"},  # 可以是字典或 JSON 字符串
        "priority": 5,
        "client_request_id": f"request_{i}"
    }
    for i in range(10)
]

result = manager.batch_assign_tasks(
    target_agent_id="agent_001",
    tasks=tasks,
    ensure_queue_exists=True  # 默认 True
)

if result["success"]:
    print(f"成功分配 {result['summary']['success_count']} 个任务")
    print(f"失败 {result['summary']['failure_count']} 个任务")
    for task_result in result["results"]:
        print(f"任务ID: {task_result['task_id']}")

负载均衡分配

# 根据负载情况自动选择 agent
agent_ids = ["agent_001", "agent_002", "agent_003"]

result = manager.assign_task_with_load_balance(
    candidate_agents=agent_ids,
    task_type=DATA_PROCESSING,
    payload={"data": "test"},
    custom_task_type="",
    priority=9,
    client_request_id="unique_id",
    strategy="least_pending",  # 选择待处理任务最少的 agent
    # strategy="round_robin",  # 轮询分配
    # strategy="random",  # 随机选择
    # strategy="custom",  # 使用自定义函数
    # load_balance_func=custom_func,  # 自定义负载均衡函数
    ensure_queue_exists=True
)

if result["success"]:
    print(f"任务已分配到: {result['selected_agent_id']}")
    print(f"任务ID: {result['task_id']}")
    print(f"使用的策略: {result['strategy']}")

负载均衡策略说明：

• least_pending: 选择待处理任务最少的 agent（默认）
• round_robin: 轮询分配
• random: 随机选择
• custom: 使用自定义函数（需要提供 load_balance_func 参数）

任务分发策略

# 分发任务到多个 agent
agent_ids = ["agent_001", "agent_002", "agent_003"]
tasks = [
    {
        "type": DATA_PROCESSING,
        "payload": {"id": i}
    }
    for i in range(9)
]

result = manager.distribute_tasks(
    tasks=tasks,
    target_agents=agent_ids,  # 可以是单个 agent_id 字符串或列表
    distribution_strategy="round_robin",  # 轮询分发
    # distribution_strategy="random",  # 随机分发
    # distribution_strategy="single",  # 所有任务发给单个 agent
    ensure_queue_exists=True
)

if result["success"]:
    print(f"总共分发 {result['summary']['total']} 个任务")
    print(f"成功: {result['summary']['success_count']}")
    print(f"失败: {result['summary']['failure_count']}")
    for agent_id, agent_result in result["results"].items():
        print(f"Agent {agent_id}: {agent_result['summary']['success_count']} 个任务")

队列状态查询

# 获取单个 agent 的队列状态
status = manager.get_agent_queue_status("agent_001")
print(f"Agent: {status['agent_id']}")
print(f"待处理: {status['pending_count']}")
print(f"处理中: {status['processing_count']}")
print(f"已完成: {status['completed_count']}")
print(f"失败: {status['failed_count']}")

# 获取多个 agent 的队列状态
statuses = manager.get_multiple_agents_status(["agent_001", "agent_002", "agent_003"])
for agent_id, status in statuses.items():
    print(f"{agent_id}: {status['pending_count']} 待处理")

# 获取任务分配统计
stats = manager.get_assignment_stats()
print(f"分配统计: {stats}")

# 重置分配统计
manager.reset_assignment_stats()

上下文管理器

# 使用上下文管理器自动关闭连接
with AgentQueueManager() as manager:
    result = manager.assign_task(
        target_agent_id="agent_001",
        task_type=DATA_PROCESSING,
        payload={"data": "test"}
    )
    # 自动关闭连接

4. AgentQueueReportClient - 实时报表客户端

AgentQueueReportClient 提供实时队列监控和报表功能，支持多个 agent 的队列状态监控。

创建报表客户端

from agent_queues import AgentQueueReportClient, create_report_client

# 方式1：使用默认配置
report_client = AgentQueueReportClient()

# 方式2：使用便捷函数
report_client = create_report_client(
    agent_ids=["agent_001", "agent_002"],  # 可选，如果提供会自动开始监控
    update_interval=5,
    grpc_host="192.168.1.100",
    grpc_port=50052
)

# 方式3：使用已有的客户端
from agent_queues import PrivateAgentTasksQueue
client = PrivateAgentTasksQueue()
report_client = AgentQueueReportClient(queue_client=client)

获取单个 Agent 报表

# 获取单个 agent 的报表
report = report_client.get_single_agent_report(
    agent_id="agent_001",
    include_task_details=False  # 是否包含任务详情
)

if report:
    print(f"Agent: {report.agent_id}")
    print(f"时间戳: {report.timestamp}")
    print(f"队列存在: {report.queue_exists}")
    print(f"待处理: {report.pending_count}")
    print(f"处理中: {report.processing_count}")
    print(f"已完成: {report.completed_count}")
    print(f"失败: {report.failed_count}")
    print(f"总计: {report.total_count}")
    print(f"成功率: {report.success_rate}%")
    print(f"完成率: {report.completion_rate}%")
    print(f"健康状态: {report.health}")  # HEALTHY, WARNING, CRITICAL
    print(f"健康信息: {report.health_message}")
    
    # 如果包含任务详情
    if report.recent_tasks:
        for task in report.recent_tasks:
            print(f"  任务: {task['task_id']}, 状态: {task['status']}")
else:
    print("队列不存在")

获取多个 Agent 报表

# 获取多个 agent 的报表
report = report_client.get_multi_agent_report(
    agent_ids=["agent_001", "agent_002", "agent_003"],
    include_task_details=False
)

print(f"报表时间: {report.timestamp}")
print(f"总 Agent 数: {report.total_agents}")
print(f"活跃 Agent 数: {report.active_agents}")
print(f"总待处理: {report.total_pending}")
print(f"总处理中: {report.total_processing}")
print(f"总已完成: {report.total_completed}")
print(f"总失败: {report.total_failed}")
print(f"总任务数: {report.total_tasks}")
print(f"全局成功率: {report.global_success_rate}%")
print(f"全局完成率: {report.global_completion_rate}%")

# 遍历每个 agent 的报表
for agent_report in report.agent_reports:
    print(f"\nAgent {agent_report.agent_id}:")
    print(f"  待处理: {agent_report.pending_count}")
    print(f"  健康状态: {agent_report.health}")

# 获取指定 agent 的报表
agent_report = report.get_agent_report("agent_001")
if agent_report:
    print(f"Agent 001 的待处理任务: {agent_report.pending_count}")

实时监控（回调方式）

import time

def on_report(report):
    """报表更新回调函数"""
    print(f"\n报表时间: {report.timestamp}")
    print(f"总 Agent 数: {report.total_agents}, 活跃: {report.active_agents}")
    print(f"总待处理: {report.total_pending}, 总处理中: {report.total_processing}")
    
    for agent_report in report.agent_reports:
        print(f"\nAgent {agent_report.agent_id}:")
        print(f"  待处理: {agent_report.pending_count}")
        print(f"  处理中: {agent_report.processing_count}")
        print(f"  已完成: {agent_report.completed_count}")
        print(f"  失败: {agent_report.failed_count}")
        print(f"  健康状态: {agent_report.health}")

# 开始监控
report_client.start_monitoring(
    agent_ids=["agent_001", "agent_002"],
    update_interval=5,  # 每 5 秒更新一次（最小1秒，最大60秒）
    callback=on_report,
    include_task_details=False  # 是否包含任务详情
)

# 运行一段时间后停止
time.sleep(60)
report_client.stop_monitoring()

实时监控（流式方式）

# 使用生成器方式获取实时报表（基于 gRPC 流式接口）
for report in report_client.stream_reports(
    agent_ids=["agent_001", "agent_002"],
    update_interval=5,  # 每 5 秒更新一次（最小1秒，最大60秒）
    include_task_details=False
):
    print(f"\n报表时间: {report.timestamp}")
    for agent_report in report.agent_reports:
        print(f"Agent {agent_report.agent_id}: {agent_report.pending_count} 待处理")
    
    # 可以随时中断
    if should_stop:
        break

上下文管理器

# 使用上下文管理器自动关闭连接
with AgentQueueReportClient() as report_client:
    report = report_client.get_single_agent_report("agent_001")
    # 自动关闭连接和停止监控

查看接口定义

客户端包中包含了完整的源代码和接口定义文件，方便查看和理解接口实现。

查看源代码定义

方法1：直接从子模块导入（最可靠）⭐

# 推荐：直接从子模块导入，IDE 可以正确跳转
from agent_queues.agent_queue import PrivateAgentTasksQueue
from agent_queues.agent_queue_manager import AgentQueueManager
from agent_queues.agent_queue_report import AgentQueueReportClient

# 点击类名可以跳转到源代码
queue = PrivateAgentTasksQueue()
manager = AgentQueueManager()
report_client = AgentQueueReportClient()

方法2：使用 IDE 的跳转功能

• VS Code:
  • 按住 Ctrl (Windows/Linux) 或 Cmd (Mac) 点击类名
  • 或按 F12 跳转到定义
  • 或按 Alt+F12 预览定义
• PyCharm:
  • 按住 Ctrl (Windows/Linux) 或 Cmd (Mac) 点击类名
  • 或按 Ctrl+B (Windows/Linux) 或 Cmd+B (Mac)

方法3：手动查找文件

• VS Code: 按 Ctrl+P，输入 agent_queue.py、agent_queue_manager.py 或 agent_queue_report.py
• PyCharm: 按 Ctrl+Shift+N，输入文件名

方法4：使用 Python 代码查找文件路径

from agent_queues import PrivateAgentTasksQueue, AgentQueueManager, AgentQueueReportClient
import inspect

# 获取源代码文件路径
print(f"PrivateAgentTasksQueue: {inspect.getfile(PrivateAgentTasksQueue)}")
print(f"AgentQueueManager: {inspect.getfile(AgentQueueManager)}")
print(f"AgentQueueReportClient: {inspect.getfile(AgentQueueReportClient)}")

查看 Proto 接口定义文件

方法1：使用工具函数（推荐）⭐

from agent_queues import get_proto_file_path, get_proto_content

# 获取文件路径
proto_path = get_proto_file_path()
print(f"Proto 文件位置: {proto_path}")

# 直接获取文件内容
proto_content = get_proto_content()
print(proto_content)

方法2：在 IDE 中查找

• VS Code: 按 Ctrl+P，输入 queue_service.proto
• PyCharm: 按 Ctrl+Shift+N，输入 queue_service.proto

方法3：直接访问文件

import agent_queues
import os

proto_path = os.path.join(os.path.dirname(client.__file__), 'queue_service.proto')
with open(proto_path, 'r', encoding='utf-8') as f:
    print(f.read())

更多信息请参考：

• IDE 跳转说明
• IDE 问题排查指南
• Proto 接口文件说明

API 参考

任务状态枚举

from agent_queues import PENDING, PROCESSING, COMPLETED, FAILED

PENDING = 0      # 待处理
PROCESSING = 1   # 处理中
COMPLETED = 2    # 已完成
FAILED = 3       # 失败

任务类型枚举

from agent_queues import (
    DATA_PROCESSING,   # 数据处理
    IMAGE_PROCESSING,  # 图像处理
    TEXT_ANALYSIS,     # 文本分析
    MODEL_INFERENCE,   # 模型推理
    DATA_EXTRACTION,   # 数据提取
    FILE_UPLOAD,       # 文件上传
    FILE_DOWNLOAD,     # 文件下载
    API_CALL,          # API 调用
    DATABASE_QUERY,    # 数据库查询
    CUSTOM             # 自定义类型
)

任务优先级

• 范围：1-10
• 数字越大，优先级越高
• 默认优先级：5
• 高优先级任务会优先被处理

幂等性保证

通过 client_request_id 参数保证任务提交的幂等性：

import uuid

# 第一次提交
client_request_id = f"request_{uuid.uuid4()}"
response1 = client.submit_task(
    agent_id="agent_001",
    task_type=DATA_PROCESSING,
    payload=json.dumps({"data": "test"}),
    client_request_id=client_request_id
)

# 使用相同的 client_request_id 再次提交，会返回相同的 task_id
response2 = client.submit_task(
    agent_id="agent_001",
    task_type=DATA_PROCESSING,
    payload=json.dumps({"data": "test"}),
    client_request_id=client_request_id
)

# response1.task_id == response2.task_id

完整示例

基本工作流程

from agent_queues import PrivateAgentTasksQueue, DATA_PROCESSING, COMPLETED, FAILED
import json

# 1. 创建客户端
client = PrivateAgentTasksQueue()
agent_id = "my_agent_001"

# 2. 创建队列
client.create_queue(agent_id)

# 3. 提交任务
response = client.submit_task(
    agent_id=agent_id,
    task_type=DATA_PROCESSING,
    payload=json.dumps({"data": "test"}),
    priority=9
)
task_id = response.task_id

# 4. 获取任务
task_response = client.get_task(agent_id=agent_id, timeout=5)
if task_response.success:
    task = task_response.task
    
    try:
        # 5. 处理任务
        result = process_task(task.payload)
        
        # 6. 更新任务状态为完成
        client.update_task_status(
            task_id=task.task_id,
            agent_id=agent_id,
            status=COMPLETED,
            result=json.dumps(result)
        )
    except Exception as e:
        # 7. 更新任务状态为失败
        client.update_task_status(
            task_id=task.task_id,
            agent_id=agent_id,
            status=FAILED,
            error_message=str(e)
        )

使用 AgentQueue 的简化流程

from agent_queues import PrivateAgentTasksQueue, DATA_PROCESSING, COMPLETED
import json

# 1. 创建客户端并获取队列
client = PrivateAgentTasksQueue()
queue = client.get_queue("my_agent_001", create_if_not_exists=True)

# 2. 提交任务
response = queue.submit_task(
    task_type=DATA_PROCESSING,
    payload=json.dumps({"data": "test"}),
    priority=9
)

# 3. 处理任务循环
while True:
    task_response = queue.get_task(timeout=5)
    if not task_response.success:
        break
    
    task = task_response.task
    try:
        result = process_task(task.payload)
        queue.update_task_status(
            task_id=task.task_id,
            status=COMPLETED,
            result=json.dumps(result)
        )
    except Exception as e:
        queue.update_task_status(
            task_id=task.task_id,
            status=FAILED,
            error_message=str(e)
        )

任务分配示例

from agent_queues import AgentQueueManager, DATA_PROCESSING
import json

manager = AgentQueueManager()

# 分配任务到指定 agent
result = manager.assign_task(
    target_agent_id="agent_001",
    task_type=DATA_PROCESSING,
    payload={"data": "task"},
    priority=9
)

# 负载均衡分配
agent_ids = ["agent_001", "agent_002", "agent_003"]
result = manager.assign_task_with_load_balance(
    candidate_agents=agent_ids,
    task_type=DATA_PROCESSING,
    payload={"data": "task"},
    strategy="least_pending",
    priority=9
)

# 批量分配
tasks = [
    {"type": DATA_PROCESSING, "payload": {"id": i}}
    for i in range(10)
]
result = manager.batch_assign_tasks(
    target_agent_id="agent_001",
    tasks=tasks
)

# 任务分发
result = manager.distribute_tasks(
    tasks=tasks,
    target_agents=agent_ids,
    distribution_strategy="round_robin"
)

实时监控示例

from agent_queues import AgentQueueReportClient
import time

report_client = AgentQueueReportClient()

def on_report(report):
    print(f"\n报表时间: {report.timestamp}")
    for agent_report in report.agent_reports:
        print(f"Agent {agent_report.agent_id}:")
        print(f"  待处理: {agent_report.pending_count}")
        print(f"  处理中: {agent_report.processing_count}")
        print(f"  已完成: {agent_report.completed_count}")
        print(f"  失败: {agent_report.failed_count}")
        print(f"  健康状态: {agent_report.health}")

# 开始监控
report_client.start_monitoring(
    agent_ids=["agent_001", "agent_002"],
    update_interval=5,
    callback=on_report
)

# 运行 60 秒
time.sleep(60)
report_client.stop_monitoring()

高级功能

自动重试和重连

客户端内置了自动重试和重连机制：

• 自动重试：gRPC 调用失败时自动重试（可配置）
• 自动重连：连接断开时自动重连
• 指数退避：重试间隔按指数增长

配置重试参数：

from agent_queues import PrivateAgentTasksQueue

client = PrivateAgentTasksQueue(
    grpc_max_retry_attempts=5,      # 最大重试次数
    grpc_initial_backoff=1.0,        # 初始退避时间（秒）
    grpc_max_backoff=10.0,           # 最大退避时间（秒）
    grpc_backoff_multiplier=2.0      # 退避倍数
)

任务优先级

任务按优先级排序，高优先级任务优先处理：

# 高优先级任务（优先级 9）
client.submit_task(
    agent_id="agent_001",
    task_type=DATA_PROCESSING,
    payload=json.dumps({"urgent": True}),
    priority=9  # 高优先级
)

# 普通优先级任务（优先级 5，默认）
client.submit_task(
    agent_id="agent_001",
    task_type=DATA_PROCESSING,
    payload=json.dumps({"normal": True}),
    priority=5  # 默认优先级
)

# 低优先级任务（优先级 1）
client.submit_task(
    agent_id="agent_001",
    task_type=DATA_PROCESSING,
    payload=json.dumps({"low": True}),
    priority=1  # 低优先级
)

注意事项

1. agent_id 路由：所有接口都通过 agent_id 路由到对应的私有化任务队列，确保每个 agent 使用唯一的 ID。

2. 任务负载格式：payload 和 result 字段必须是有效的 JSON 字符串。AgentQueueManager 的方法可以接受字典，会自动转换为 JSON 字符串。

3. 超时设置：get_task 的 timeout 参数设置为 0 表示不等待，立即返回；大于 0 表示等待指定秒数。

4. 任务优先级：优先级范围 1-10，数字越大优先级越高，高优先级任务会优先被处理。

5. 幂等性：使用 client_request_id 可以保证任务提交的幂等性，防止重复提交。

6. 连接管理：建议复用客户端实例，避免频繁创建连接。可以使用上下文管理器自动管理连接。

7. 队列自动创建：AgentQueueManager 的所有分配方法默认会自动创建目标 agent 的队列（ensure_queue_exists=True）。

更多信息

• 查看接口定义：

  • 源代码：
    • client/queues/agent_queue.py - PrivateAgentTasksQueue, AgentQueue
    • client/queues/agent_queue_manager.py - AgentQueueManager
    • client/queues/agent_queue_report.py - AgentQueueReportClient
  • Proto 定义：client/queue_service.proto
  • 使用 get_proto_file_path() 和 get_proto_content() 获取 Proto 文件

• 文档：

  • IDE 跳转说明
  • IDE 问题排查指南
  • Proto 接口文件说明
  • API 参考

• 示例代码：

  • examples/get_queue_example.py - AgentQueue 使用示例
  • examples/report_example.py - 实时报表示例
  • examples/view_proto_example.py - 查看 Proto 文件示例