Metadata-Version: 2.4
Name: queues-manager
Version: 1.0.10
Summary: A private task queue system for agents using Redis + Celery with gRPC interface (Client Package)
Author-email: NightSailingShip <149816081+NightSailingShip@users.noreply.github.com>
License: MIT
Project-URL: Homepage, https://github.com/NightSailingShip/queues
Project-URL: Documentation, https://github.com/NightSailingShip/queues#readme
Project-URL: Repository, https://github.com/NightSailingShip/queues
Project-URL: Issues, https://github.com/NightSailingShip/queues/issues
Keywords: task-queue,celery,redis,grpc,agent,distributed,client
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: grpcio>=1.60.0
Requires-Dist: grpcio-tools>=1.60.0
Requires-Dist: protobuf>=4.25.0
Requires-Dist: pyyaml>=6.0.0
Provides-Extra: dev
Requires-Dist: pytest>=7.4.0; extra == "dev"
Requires-Dist: build>=1.0.0; extra == "dev"
Requires-Dist: twine>=4.0.0; extra == "dev"
Dynamic: license-file

# Agent Queue 客户端使用指南

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

## 目录

- [快速开始](#快速开始)
- [安装](#安装)
- [客户端 API 使用](#客户端-api-使用) ⭐
  - [PrivateAgentTasksQueue - 基础客户端](#1-privateagenttasksqueue---基础客户端)
  - [AgentQueue - 针对特定 Agent 的队列](#2-agentqueue---针对特定-agent-的队列)
  - [AgentQueueManager - 任务分配管理器](#3-agentqueuemanager---任务分配管理器)
  - [AgentQueueReportClient - 实时报表客户端](#4-agentqueuereportclient---实时报表客户端)
- [查看接口定义](#查看接口定义) ⭐
- [API 参考](#api-参考)
- [完整示例](#完整示例)

## 快速开始

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

```python
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（更简洁）⭐

```python
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 安装（推荐）

```bash
# 使用 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。

#### 创建客户端

```python
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)
```

#### 队列管理接口

```python
# 创建队列
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)
```

#### 任务操作接口

```python
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"  # 幂等性保证
)
```

#### 任务查询接口

```python
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
)
```

#### 获取队列对象

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

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

#### 上下文管理器

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

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

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

#### 获取队列对象

```python
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")
```

#### 队列管理方法

```python
# 获取队列信息
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)
```

#### 任务操作方法

```python
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")
```

#### 任务查询方法

```python
# 列出任务
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 的队列不存在，会自动创建队列后再分配任务。**

#### 创建管理器

```python
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

```python
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']}")
```

#### 批量分配任务

```python
# 批量分配任务到指定 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']}")
```

#### 负载均衡分配

```python
# 根据负载情况自动选择 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` 参数）

#### 任务分发策略

```python
# 分发任务到多个 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']} 个任务")
```

#### 队列状态查询

```python
# 获取单个 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()
```

#### 上下文管理器

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

### 4. AgentQueueReportClient - 实时报表客户端

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

#### 创建报表客户端

```python
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 报表

```python
# 获取单个 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 报表

```python
# 获取多个 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}")
```

#### 实时监控（回调方式）

```python
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()
```

#### 实时监控（流式方式）

```python
# 使用生成器方式获取实时报表（基于 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
```

#### 上下文管理器

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

## 查看接口定义

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

### 查看源代码定义

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

```python
# 推荐：直接从子模块导入，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 代码查找文件路径

```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：使用工具函数（推荐）⭐

```python
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：直接访问文件

```python
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 跳转说明](docs/IDE_JUMP_TO_DEFINITION.md)
- [IDE 问题排查指南](docs/IDE_TROUBLESHOOTING.md)
- [Proto 接口文件说明](docs/PROTO_FILES.md)

## API 参考

### 任务状态枚举

```python
from agent_queues import PENDING, PROCESSING, COMPLETED, FAILED

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

### 任务类型枚举

```python
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` 参数保证任务提交的幂等性：

```python
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
```

## 完整示例

### 基本工作流程

```python
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 的简化流程

```python
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)
        )
```

### 任务分配示例

```python
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"
)
```

### 实时监控示例

```python
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 调用失败时自动重试（可配置）
- **自动重连**：连接断开时自动重连
- **指数退避**：重试间隔按指数增长

配置重试参数：

```python
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      # 退避倍数
)
```

### 任务优先级

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

```python
# 高优先级任务（优先级 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 跳转说明](docs/IDE_JUMP_TO_DEFINITION.md)
  - [IDE 问题排查指南](docs/IDE_TROUBLESHOOTING.md)
  - [Proto 接口文件说明](docs/PROTO_FILES.md)
  - [API 参考](docs/API_REFERENCE.md)

- **示例代码**：
  - `examples/get_queue_example.py` - AgentQueue 使用示例
  - `examples/report_example.py` - 实时报表示例
  - `examples/view_proto_example.py` - 查看 Proto 文件示例
