从概念到实践：构建基于Harness层消息队列的Agent异步任务管控系统

---

第一部分：引言与基础 (Introduction & Foundation)

1. 引人注目的标题 (Compelling Title)

主标题： 从概念到实践：构建基于Harness层消息队列的Agent异步任务管控系统
副标题： 深入理解分布式Agent架构中的异步任务调度、执行与监控机制

2. 摘要/引言 (Abstract / Introduction)

问题陈述

在当今快速发展的人工智能和分布式系统领域，Agent系统正变得越来越复杂和重要。随着Agent数量的增加和任务复杂度的提升，如何高效地管理和协调多个Agent之间的异步任务执行，成为了一个亟待解决的关键问题。传统的同步任务执行方式往往无法满足大规模Agent系统的需求，会导致系统性能下降、资源利用率低、响应时间长等问题。

核心方案

本文提出了一种基于Harness层消息队列的Agent异步任务管控系统方案。通过在Agent架构中引入专门的Harness层，并利用消息队列作为异步通信和任务调度的核心基础设施，我们可以实现任务的高效分发、并行执行、状态监控和容错处理。这种架构不仅能够提高系统的整体性能和资源利用率，还能增强系统的可扩展性、可靠性和可维护性。

主要成果/价值

读完本文后，你将能够：

• 深入理解Agent异步任务管控的核心概念和理论基础
• 掌握基于消息队列的异步任务调度机制
• 学会设计和实现Harness层消息队列系统
• 了解如何在实际项目中应用这些技术
• 掌握性能优化和最佳实践
• 能够解决常见的技术问题

文章导览

本文将按照以下结构展开：首先介绍问题背景和动机，然后阐述核心概念和理论基础，接着详细说明环境准备和分步实现过程，之后对关键代码进行解析，展示系统运行结果，讨论性能优化和最佳实践，预判常见问题并提供解决方案，最后展望未来发展方向并进行总结。

3. 目标读者与前置知识 (Target Audience & Prerequisites)

目标读者

本文适合以下读者：

• 有一定后端开发经验的中高级开发者
• 对分布式系统有基础了解的技术人员
• 正在或计划开发Agent系统的工程师
• 对消息队列和异步任务处理感兴趣的开发者
• 系统架构师和技术负责人

前置知识

阅读本文前，建议你具备以下基础知识：

• 熟悉Python编程
• 了解基本的分布式系统概念
• 对消息队列（如RabbitMQ、Kafka）有基础了解
• 了解多线程和异步编程的基本概念
• 熟悉RESTful API设计（可选，但有帮助）

4. 文章目录 (Table of Contents)

1. 引言与基础
2. 问题背景与动机
3. 核心概念与理论基础
4. 环境准备
5. 分步实现
6. 关键代码解析与深度剖析
7. 结果展示与验证
8. 性能优化与最佳实践
9. 常见问题与解决方案
10. 未来展望与扩展方向
11. 总结
12. 参考资料
13. 附录

---

第二部分：核心内容 (Core Content)

5. 问题背景与动机 (Problem Background & Motivation)

5.1 问题背景

随着人工智能技术的快速发展，Agent系统作为一种模拟人类智能行为的技术，正在被广泛应用于各个领域，如智能客服、自动驾驶、工业自动化、金融分析等。一个典型的Agent系统通常由多个独立的Agent组成，每个Agent负责完成特定的任务，它们之间需要相互协作和通信，以实现共同的目标。

在早期的Agent系统中，任务执行通常采用同步方式，即一个任务完成后才能开始下一个任务。这种方式在Agent数量较少、任务复杂度较低的情况下是可行的，但随着系统规模的扩大和任务复杂度的提升，同步执行方式的弊端逐渐显现出来：

1. 性能瓶颈：同步执行方式无法充分利用系统资源，导致CPU、内存等资源利用率低，系统整体性能下降。
2. 响应时间长：由于任务需要依次执行，用户需要等待较长时间才能得到结果，用户体验差。
3. 可扩展性差：随着Agent数量的增加，同步执行方式难以满足系统的扩展需求，系统的复杂度会呈指数级增长。
4. 容错能力弱：如果某个Agent在执行任务时出现故障，可能会导致整个系统瘫痪，影响系统的可靠性。

为了解决这些问题，异步任务管控机制逐渐成为Agent系统的核心技术之一。通过将任务的提交和执行分离，利用消息队列作为中间件，我们可以实现任务的异步分发和并行执行，从而提高系统的性能、响应速度、可扩展性和容错能力。

5.2 现有解决方案的局限性

虽然目前已经有一些异步任务处理的解决方案，如Celery、RQ、Huey等，但这些方案在应用于Agent系统时，往往存在以下局限性：

1. 缺乏专门的Agent管控机制：现有方案主要关注任务的调度和执行，缺乏对Agent的专门管控，如Agent的注册、发现、监控、生命周期管理等。
2. 架构不够灵活：现有方案的架构往往比较固定，难以适应不同Agent系统的需求，如不同的Agent类型、任务类型、通信模式等。
3. 可扩展性有限：现有方案在处理大规模Agent系统时，往往会遇到性能瓶颈，难以支持数千甚至数万个Agent的并发执行。
4. 监控和调试能力不足：现有方案的监控和调试能力往往比较有限，难以实时了解系统的运行状态，快速定位和解决问题。

为了克服这些局限性，我们需要设计一种专门针对Agent系统的异步任务管控架构，这就是我们提出的Harness层消息队列系统。

5.3 技术选型理由

在设计Harness层消息队列系统时，我们需要选择合适的技术栈。经过综合考虑，我们选择了以下技术：

1. Python：Python是一种简单易学、功能强大的编程语言，拥有丰富的第三方库和框架，非常适合用于快速开发原型系统。
2. RabbitMQ：RabbitMQ是一个开源的消息代理软件，实现了高级消息队列协议（AMQP），具有可靠性高、灵活性强、易于使用等优点，非常适合作为Agent系统的消息中间件。
3. Asyncio：Asyncio是Python 3.4版本引入的标准库，用于编写单线程并发代码，通过协程（coroutine）和事件循环（event loop）实现高效的异步I/O操作，非常适合用于实现Agent的异步执行。
4. FastAPI：FastAPI是一个现代、快速（高性能）的Web框架，用于基于Python 3.7+构建API，具有自动生成API文档、类型提示、异步支持等优点，非常适合用于实现系统的管理接口。

选择这些技术的理由如下：

• Python：开发效率高，生态丰富，社区活跃。
• RabbitMQ：成熟稳定，功能强大，支持多种消息模式，易于集成。
• Asyncio：可以高效地处理大量并发连接，提高系统的性能和响应速度。
• FastAPI：可以快速构建高性能的RESTful API，提供友好的管理界面。

---

6. 核心概念与理论基础 (Core Concepts & Theoretical Foundation)

6.1 核心概念

在深入探讨Harness层消息队列系统之前，我们需要先了解一些核心概念：

6.1.1 Agent

Agent是指在特定环境下，能够自主感知环境、做出决策并采取行动，以实现特定目标的实体。在我们的系统中，Agent可以是一个软件程序、一个机器人、一个智能设备等。每个Agent都有自己的唯一标识符、能力描述、状态信息等。

Agent的核心属性包括：

• 身份（Identity）：Agent的唯一标识符，用于区分不同的Agent。
• 目标（Goal）：Agent需要完成的任务或实现的目标。
• 感知（Perception）：Agent感知环境的能力，通过传感器或API获取环境信息。
• 行动（Action）：Agent改变环境的能力，通过执行器或API对环境产生影响。
• 推理（Reasoning）：Agent根据感知到的信息和已有的知识，做出决策的能力。
• 通信（Communication）：Agent与其他Agent或系统进行信息交换的能力。

6.1.2 Harness层

Harness层是我们系统中的一个关键组件，位于Agent和底层基础设施之间，负责Agent的生命周期管理、任务调度、资源分配、状态监控等功能。Harness层的主要作用是为Agent提供一个统一的运行环境，屏蔽底层基础设施的复杂性，让Agent可以专注于完成自己的任务。

Harness层的核心功能包括：

• Agent注册与发现：管理Agent的注册信息，提供Agent发现服务。
• 任务分发与调度：接收任务请求，根据任务类型和Agent能力，将任务分发给合适的Agent执行。
• 资源管理：监控和管理系统资源，确保资源的合理分配和利用。
• 状态监控：实时监控Agent和任务的状态，提供告警和通知功能。
• 容错处理：处理Agent和任务的故障，确保系统的可靠性和可用性。

6.1.3 消息队列

消息队列是一种用于在应用程序之间传递消息的中间件，它可以实现异步通信和解耦。在我们的系统中，消息队列主要用于任务的分发和执行状态的反馈。生产者（任务提交者）将任务消息发送到消息队列，消费者（Agent）从消息队列中获取任务消息并执行，然后将执行结果发送回消息队列。

消息队列的核心概念包括：

• 生产者（Producer）：发送消息的应用程序。
• 消费者（Consumer）：接收和处理消息的应用程序。
• 队列（Queue）：存储消息的容器，消息按照先进先出（FIFO）的顺序被处理。
• 交换器（Exchange）：在RabbitMQ中，交换器负责接收生产者发送的消息，并将消息路由到一个或多个队列中。
• 绑定（Binding）：绑定是交换器和队列之间的关联关系，它告诉交换器应该将哪些消息路由到特定的队列中。
• 路由键（Routing Key）：路由键是消息的一个属性，交换器根据路由键和绑定规则来决定将消息路由到哪个队列中。

6.1.4 异步任务管控

异步任务管控是指对异步任务的整个生命周期进行管理，包括任务的提交、调度、执行、监控、容错等。在我们的系统中，异步任务管控主要由Harness层负责，它通过消息队列实现任务的异步分发和执行，确保任务能够高效、可靠地完成。

异步任务管控的核心功能包括：

• 任务提交：接收用户或其他系统提交的任务请求。
• 任务调度：根据任务的优先级、类型、资源需求等，决定任务的执行顺序和执行Agent。
• 任务执行：将任务分发给Agent执行，并监控执行过程。
• 状态跟踪：实时跟踪任务的执行状态，如待处理、执行中、已完成、失败等。
• 结果反馈：将任务的执行结果反馈给任务提交者。
• 容错处理：处理任务执行失败的情况，如重试、转移到其他Agent执行等。

6.2 概念结构与核心要素组成

我们的Harness层消息队列系统主要由以下几个核心要素组成：

1. 任务管理模块：负责任务的提交、存储、调度和状态跟踪。
2. Agent管理模块：负责Agent的注册、发现、监控和生命周期管理。
3. 消息队列模块：负责消息的传递和存储，是任务分发和执行状态反馈的核心基础设施。
4. API网关模块：提供统一的API接口，用于任务提交、状态查询、Agent管理等。
5. 监控与告警模块：实时监控系统的运行状态，提供告警和通知功能。
6. 数据存储模块：存储任务信息、Agent信息、系统日志等数据。

这些要素之间的关系可以用以下的架构图来表示：

6.3 概念之间的关系

为了更好地理解各个概念之间的关系，我们可以从以下几个维度进行对比：

6.3.1 概念核心属性维度对比

概念	核心属性	主要功能	交互对象	关键特性
Agent	身份、目标、感知、行动、推理、通信	执行具体任务	Harness层、其他Agent	自主性、反应性、主动性、社会性
Harness层	任务管理、Agent管理、资源管理、监控	管控Agent和任务	用户层、消息队列层、数据存储层	解耦、管控、协调、容错
消息队列	生产者、消费者、队列、交换器、绑定、路由键	传递和存储消息	Harness层、Agent层	异步、解耦、可靠、可扩展
异步任务管控	任务提交、调度、执行、跟踪、反馈、容错	管理任务生命周期	任务管理模块、Agent、消息队列	异步、并行、高效、可靠

6.3.2 概念联系的ER实体关系图

6.3.3 交互关系图

6.4 数学模型

在Harness层消息队列系统中，我们可以使用一些数学模型来描述和分析系统的性能。以下是一些常用的数学模型：

6.4.1 任务到达模型

我们可以使用泊松过程来描述任务的到达过程。泊松过程是一种描述随机事件发生次数的数学模型，它具有以下特性：

1. 在任意两个不重叠的时间区间内，事件发生的次数是相互独立的。
2. 在一个足够小的时间区间内，事件发生一次的概率与时间区间的长度成正比。
3. 在一个足够小的时间区间内，事件发生两次或两次以上的概率可以忽略不计。

泊松过程的概率质量函数为：
$$P(N(t)=k)=\frac{(\lambda t{)}^k{e}^{-\lambda t}}{k!}$$
其中，$N(t)$ 表示在时间区间 $[0,t]$ 内事件发生的次数，$\lambda$ 表示事件发生的速率（单位时间内事件发生的平均次数），$k$ 表示事件发生的次数。

6.4.2 任务执行时间模型

我们可以使用指数分布来描述任务的执行时间。指数分布是一种连续概率分布，常用于描述独立随机事件发生的时间间隔。它具有无记忆性（Memoryless Property），即：
$$P(T>t+s|T>s)=P(T>t)$$
其中，$T$ 表示任务执行时间，$t$ 和 $s$ 表示时间。

指数分布的概率密度函数为：
$$f(t)=\lambda e^{-\lambda t},t\ge 0$$
其中，$\lambda$ 表示任务执行的速率（单位时间内完成任务的平均次数）。

指数分布的累积分布函数为：
$$F(t)=1-e^{-\lambda t},t\ge 0$$

6.4.3 队列模型

我们可以使用M/M/1队列模型来描述系统中的任务队列。M/M/1队列模型是一种最简单的队列模型，它假设：

1. 任务到达过程是泊松过程，到达速率为 $\lambda$。
2. 任务执行时间服从指数分布，执行速率为 $\mu$。
3. 只有一个服务台（即只有一个Agent执行任务）。
4. 队列长度没有限制。
5. 任务按照先进先出（FIFO）的顺序被处理。

在M/M/1队列模型中，系统的利用率 $\rho$ 为：
$$\rho =\frac{\lambda }{\mu }$$

系统中的平均任务数 $L$ 为：
$$L=\frac{\rho }{1-\rho }=\frac{\lambda }{\mu -\lambda }$$

队列中的平均任务数 $L_q$ 为：
$$L_q=\frac{{\rho }^2}{1-\rho }=\frac{{\lambda }^2}{\mu (\mu -\lambda )}$$

任务在系统中的平均停留时间 $W$ 为：
$$W=\frac{1}{\mu -\lambda }$$

任务在队列中的平均等待时间 $W_q$ 为：
$$W_q=\frac{\lambda }{\mu (\mu -\lambda )}$$

需要注意的是，只有当 $\rho <1$（即 $\lambda <\mu$）时，系统才是稳定的，否则队列长度会无限增长。

6.4.4 多服务台队列模型

当系统中有多个Agent执行任务时，我们可以使用M/M/c队列模型来描述系统中的任务队列。M/M/c队列模型与M/M/1队列模型类似，只是有 $c$ 个服务台（即有 $c$ 个Agent执行任务）。

在M/M/c队列模型中，系统的利用率 $\rho$ 为：
$$\rho =\frac{\lambda }{c\mu }$$

系统空闲的概率 $P_0$ 为：
$$P_0={\left[\sum _{k=0}^{c-1}\frac{(\lambda /\mu {)}^k}{k!}+\frac{(\lambda /\mu {)}^c}{c!(1-\rho )}\right]}^{-1}$$

队列中的平均任务数 $L_q$ 为：
$$L_q=\frac{P_0(\lambda /\mu {)}^c\rho }{c!(1-\rho {)}^2}$$

系统中的平均任务数 $L$ 为：
$$L=L_q+\frac{\lambda }{\mu }$$

任务在队列中的平均等待时间 $W_q$ 为：
$$W_q=\frac{L_q}{\lambda }$$

任务在系统中的平均停留时间 $W$ 为：
$$W=W_q+\frac{1}{\mu }=\frac{L}{\lambda }$$

同样，只有当 $\rho <1$（即 $\lambda <c\mu$）时，系统才是稳定的。

---

7. 环境准备 (Environment Setup)

在开始实现Harness层消息队列系统之前，我们需要准备好开发环境。本节将详细介绍所需的软件、库、框架及其版本，并提供安装指南。

7.1 软件要求

我们的系统需要以下软件：

1. Python 3.8+：我们将使用Python作为主要的开发语言，建议使用Python 3.8或更高版本，以确保兼容性和性能。
2. RabbitMQ 3.8+：我们将使用RabbitMQ作为消息队列中间件，建议使用RabbitMQ 3.8或更高版本。
3. PostgreSQL 12+（可选）：我们将使用PostgreSQL作为数据库，存储任务信息、Agent信息等数据。如果只是为了快速原型开发，也可以使用SQLite。
4. Redis 6+（可选）：我们将使用Redis作为缓存和会话存储，提高系统的性能。
5. Docker 20.10+（可选）：我们将使用Docker来容器化我们的应用，方便部署和管理。

7.2 库和框架要求

我们的系统需要以下Python库和框架：

1. FastAPI 0.68+：用于构建RESTful API。
2. Uvicorn 0.15+：用于运行FastAPI应用的ASGI服务器。
3. Pika 1.2+：用于与RabbitMQ进行交互的Python客户端库。
4. SQLAlchemy 1.4+：用于与数据库进行交互的ORM框架。
5. Alembic 1.7+：用于数据库迁移的工具。
6. Pydantic 1.8+：用于数据验证和设置管理的库。
7. Python-dotenv 0.19+：用于从环境变量中加载配置的库。
8. Asyncio：Python标准库，用于异步编程。
9. Logging：Python标准库，用于日志记录。

7.3 安装指南

以下是在Ubuntu 20.04系统上安装所需软件的步骤：

7.3.1 安装Python 3.8+

Ubuntu 20.04通常已经预装了Python 3.8，你可以通过以下命令检查Python版本：

python3 --version

如果没有安装Python 3.8或更高版本，可以通过以下命令安装：

sudo apt update
sudo apt install python3.8 python3.8-venv python3.8-dev

7.3.2 安装RabbitMQ 3.8+

RabbitMQ依赖于Erlang，所以我们需要先安装Erlang：

sudo apt update
sudo apt install erlang

然后安装RabbitMQ：

sudo apt install rabbitmq-server

安装完成后，启动RabbitMQ服务：

sudo systemctl start rabbitmq-server

设置RabbitMQ服务开机自启：

sudo systemctl enable rabbitmq-server

检查RabbitMQ服务状态：

sudo systemctl status rabbitmq-server

（可选）启用RabbitMQ管理插件，方便通过Web界面管理RabbitMQ：

sudo rabbitmq-plugins enable rabbitmq_management

启用管理插件后，你可以通过浏览器访问 http://localhost:15672 来管理RabbitMQ，默认用户名和密码都是 guest。

7.3.3 安装PostgreSQL 12+（可选）

通过以下命令安装PostgreSQL：

sudo apt update
sudo apt install postgresql postgresql-contrib

安装完成后，启动PostgreSQL服务：

sudo systemctl start postgresql

设置PostgreSQL服务开机自启：

sudo systemctl enable postgresql

检查PostgreSQL服务状态：

sudo systemctl status postgresql

7.3.4 安装Redis 6+（可选）

通过以下命令安装Redis：

sudo apt update
sudo apt install redis-server

安装完成后，启动Redis服务：

sudo systemctl start redis-server

设置Redis服务开机自启：

sudo systemctl enable redis-server

检查Redis服务状态：

sudo systemctl status redis-server

7.3.5 安装Docker 20.10+（可选）

通过以下命令安装Docker：

sudo apt update
sudo apt install apt-transport-https ca-certificates curl software-properties-common
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /usr/share/keyrings/docker-archive-keyring.gpg
echo "deb [arch=amd64 signed-by=/usr/share/keyrings/docker-archive-keyring.gpg] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null
sudo apt update
sudo apt install docker-ce docker-ce-cli containerd.io

安装完成后，检查Docker版本：

docker --version

（可选）将当前用户添加到docker组，避免每次使用docker命令都需要sudo：

sudo usermod -aG docker $USER

注意：添加用户到docker组后，需要重新登录才能生效。

7.4 项目设置

现在我们来创建项目目录结构，并设置虚拟环境和依赖。

7.4.1 创建项目目录结构

首先，创建项目目录：

mkdir agent-harness-mq
cd agent-harness-mq

然后，创建以下目录结构：

agent-harness-mq/
├── app/
│   ├── api/
│   │   ├── __init__.py
│   │   ├── agents.py
│   │   └── tasks.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py
│   │   ├── database.py
│   │   └── mq.py
│   ├── models/
│   │   ├── __init__.py
│   │   ├── agent.py
│   │   └── task.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── agent.py
│   │   └── task.py
│   ├── services/
│   │   ├── __init__.py
│   │   ├── agent_service.py
│   │   └── task_service.py
│   ├── workers/
│   │   ├── __init__.py
│   │   ├── agent_worker.py
│   │   └── task_worker.py
│   └── __init__.py
├── alembic/
├── tests/
│   ├── __init__.py
│   ├── test_agents.py
│   └── test_tasks.py
├── .env.example
├── .gitignore
├── alembic.ini
├── docker-compose.yml
├── main.py
├── README.md
└── requirements.txt

你可以通过以下命令创建这些目录和文件：

mkdir -p app/api app/core app/models app/schemas app/services app/workers alembic tests
touch app/api/__init__.py app/api/agents.py app/api/tasks.py
touch app/core/__init__.py app/core/config.py app/core/database.py app/core/mq.py
touch app/models/__init__.py app/models/agent.py app/models/task.py
touch app/schemas/__init__.py app/schemas/agent.py app/schemas/task.py
touch app/services/__init__.py app/services/agent_service.py app/services/task_service.py
touch app/workers/__init__.py app/workers/agent_worker.py app/workers/task_worker.py
touch app/__init__.py
touch tests/__init__.py tests/test_agents.py tests/test_tasks.py
touch .env.example .gitignore alembic.ini docker-compose.yml main.py README.md requirements.txt

7.4.2 创建虚拟环境

在项目根目录下，创建虚拟环境：

python3.8 -m venv venv

然后，激活虚拟环境：

source venv/bin/activate

7.4.3 安装依赖

在 requirements.txt 文件中添加以下内容：

fastapi==0.68.0
uvicorn==0.15.0
pika==1.2.0
sqlalchemy==1.4.23
alembic==1.7.1
pydantic==1.8.2
python-dotenv==0.19.0
psycopg2-binary==2.9.1  # 如果使用PostgreSQL
# sqlite3  # Python标准库，不需要安装
redis==3.5.3  # 如果使用Redis

然后，通过以下命令安装依赖：

pip install -r requirements.txt

7.4.4 配置环境变量

在 .env.example 文件中添加以下内容：

# 应用配置
APP_NAME=Agent Harness MQ
APP_VERSION=0.1.0
DEBUG=True

# 数据库配置
DATABASE_URL=sqlite:///./agent_harness_mq.db
# 如果使用PostgreSQL
# DATABASE_URL=postgresql://user:password@localhost/agent_harness_mq

# Redis配置（可选）
# REDIS_URL=redis://localhost:6379/0

# RabbitMQ配置
RABBITMQ_URL=amqp://guest:guest@localhost:5672/%2F
RABBITMQ_TASK_QUEUE=agent_tasks
RABBITMQ_RESULT_QUEUE=task_results
RABBITMQ_EXCHANGE=agent_exchange
RABBITMQ_ROUTING_KEY=agent.task

# API配置
API_HOST=0.0.0.0
API_PORT=8000

然后，复制 .env.example 文件为 .env：

cp .env.example .env

根据你的实际情况，修改 .env 文件中的配置。

7.4.5 配置Docker Compose（可选）

在 docker-compose.yml 文件中添加以下内容：

version: '3.8'

services:
  rabbitmq:
    image: rabbitmq:3.8-management-alpine
    container_name: agent-harness-rabbitmq
    ports:
      - "5672:5672"
      - "15672:15672"
    environment:
      RABBITMQ_DEFAULT_USER: guest
      RABBITMQ_DEFAULT_PASS: guest
      RABBITMQ_DEFAULT_VHOST: /
    volumes:
      - rabbitmq_data:/var/lib/rabbitmq

  # 如果使用PostgreSQL
  # postgres:
  #   image: postgres:13-alpine
  #   container_name: agent-harness-postgres
  #   ports:
  #     - "5432:5432"
  #   environment:
  #     POSTGRES_USER: user
  #     POSTGRES_PASSWORD: password
  #     POSTGRES_DB: agent_harness_mq
  #   volumes:
  #     - postgres_data:/var/lib/postgresql/data

  # 如果使用Redis
  # redis:
  #   image: redis:6-alpine
  #   container_name: agent-harness-redis
  #   ports:
  #     - "6379:6379"
  #   volumes:
  #     - redis_data:/data

volumes:
  rabbitmq_data:
  # postgres_data:
  # redis_data:

然后，通过以下命令启动服务：

docker-compose up -d

---

8. 分步实现 (Step-by-Step Implementation)

现在我们开始分步实现Harness层消息队列系统。我们将按照以下步骤进行：

1. 配置应用和核心组件
2. 定义数据模型和模式
3. 实现数据库连接和迁移
4. 实现消息队列连接和交互
5. 实现服务层逻辑
6. 实现API接口
7. 实现Worker
8. 实现主应用入口

8.1 配置应用和核心组件

首先，我们来配置应用和核心组件，包括配置管理、数据库连接、消息队列连接等。

8.1.1 配置管理

在 app/core/config.py 文件中添加以下内容：

from pydantic import BaseSettings
from typing import Optional


class Settings(BaseSettings):
    """应用配置类"""
    # 应用配置
    app_name: str = "Agent Harness MQ"
    app_version: str = "0.1.0"
    debug: bool = True

    # 数据库配置
    database_url: str = "sqlite:///./agent_harness_mq.db"

    # Redis配置（可选）
    redis_url: Optional[str] = None

    # RabbitMQ配置
    rabbitmq_url: str = "amqp://guest:guest@localhost:5672/%2F"
    rabbitmq_task_queue: str = "agent_tasks"
    rabbitmq_result_queue: str = "task_results"
    rabbitmq_exchange: str = "agent_exchange"
    rabbitmq_routing_key: str = "agent.task"

    # API配置
    api_host: str = "0.0.0.0"
    api_port: int = 8000

    class Config:
        env_file = ".env"
        case_sensitive = False


# 创建全局配置实例
settings = Settings()

这个配置类使用了Pydantic的BaseSettings，它会自动从环境变量中加载配置，也可以从.env文件中加载配置。

8.1.2 数据库连接

在 app/core/database.py 文件中添加以下内容：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from app.core.config import settings

# 创建数据库引擎
engine = create_engine(
    settings.database_url,
    connect_args={"check_same_thread": False} if "sqlite" in settings.database_url else {}
)

# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建基类
Base = declarative_base()


def get_db():
    """获取数据库会话的依赖项"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

这个文件设置了数据库连接，包括创建数据库引擎、会话工厂和基类。get_db函数是一个FastAPI的依赖项，用于在API接口中获取数据库会话。

8.1.3 消息队列连接

在 app/core/mq.py 文件中添加以下内容：

import pika
import json
import logging
from typing import Callable, Optional
from app.core.config import settings

logger = logging.getLogger(__name__)


class MessageQueue:
    """消息队列类，用于与RabbitMQ交互"""

    def __init__(self):
        self.connection: Optional[pika.BlockingConnection] = None
        self.channel: Optional[pika.channel.Channel] = None

    def connect(self):
        """连接到RabbitMQ"""
        try:
            parameters = pika.URLParameters(settings.rabbitmq_url)
            self.connection = pika.BlockingConnection(parameters)
            self.channel = self.connection.channel()

            # 声明交换器
            self.channel.exchange_declare(
                exchange=settings.rabbitmq_exchange,
                exchange_type='direct',
                durable=True
            )

            # 声明任务队列
            self.channel.queue_declare(
                queue=settings.rabbitmq_task_queue,
                durable=True
            )

            # 声明结果队列
            self.channel.queue_declare(
                queue=settings.rabbitmq_result_queue,
                durable=True
            )

            # 绑定任务队列到交换器
            self.channel.queue_bind(
                exchange=settings.rabbitmq_exchange,
                queue=settings.rabbitmq_task_queue,
                routing_key=settings.rabbitmq_routing_key
            )

            logger.info("成功连接到RabbitMQ")
        except Exception as e:
            logger.error(f"连接到RabbitMQ失败: {e}")
            raise

    def disconnect(self):
        """断开与RabbitMQ的连接"""
        if self.connection and not self.connection.is_closed:
            self.connection.close()
            logger.info("已断开与RabbitMQ的连接")

    def publish_task(self, task_data: dict):
        """发布任务到任务队列"""
        try:
            message = json.dumps(task_data)
            self.channel.basic_publish(
                exchange=settings.rabbitmq_exchange,
                routing_key=settings.rabbitmq_routing_key,
                body=message,
                properties=pika.BasicProperties(
                    delivery_mode=2,  # 持久化消息
                    content_type='application/json'
                )
            )
            logger.info(f"成功发布任务: {task_data.get('task_id')}")
        except Exception as e:
            logger.error(f"发布任务失败: {e}")
            raise

    def publish_result(self, result_data: dict):
        """发布结果到结果队列"""
        try:
            message = json.dumps(result_data)
            self.channel.basic_publish(
                exchange='',  # 使用默认交换器
                routing_key=settings.rabbitmq_result_queue,
                body=message,
                properties=pika.BasicProperties(
                    delivery_mode=2,  # 持久化消息
                    content_type='application/json'
                )
            )
            logger.info(f"成功发布结果: {result_data.get('task_id')}")
        except Exception as e:
            logger.error(f"发布结果失败: {e}")
            raise

    def consume_tasks(self, callback: Callable):
        """消费任务队列中的任务"""
        try:
            self.channel.basic_qos(prefetch_count=1)  # 公平调度
            self.channel.basic_consume(
                queue=settings.rabbitmq_task_queue,
                on_message_callback=callback,
                auto_ack=False  # 手动确认
            )
            logger.info("开始消费任务队列")
            self.channel.start_consuming()
        except Exception as e:
            logger.error(f"消费任务队列失败: {e}")
            raise

    def consume_results(self, callback: Callable):
        """消费结果队列中的结果"""
        try:
            self.channel.basic_qos(prefetch_count=1)  # 公平调度
            self.channel.basic_consume(
                queue=settings.rabbitmq_result_queue,
                on_message_callback=callback,
                auto_ack=False  # 手动确认
            )
            logger.info("开始消费结果队列")
            self.channel.start_consuming()
        except Exception as e:
            logger.error(f"消费结果队列失败: {e}")
            raise


# 创建全局消息队列实例
mq = MessageQueue()

这个类封装了与RabbitMQ的交互，包括连接、断开连接、发布任务、发布结果、消费任务、消费结果等功能。

8.2 定义数据模型和模式

接下来，我们定义数据模型和模式，包括Agent模型、Task模型，以及对应的Pydantic模式。

8.2.1 Agent模型

在 app/models/agent.py 文件中添加以下内容：

from sqlalchemy import Column, String, DateTime, Text, Integer
from sqlalchemy.sql import func
from app.core.database import Base


class Agent(Base):
    """Agent模型"""
    __tablename__ = "agents"

    id = Column(String, primary_key=True, index=True)
    name = Column(String, index=True, nullable=False)
    type = Column(String, index=True, nullable=False)
    status = Column(String, index=True, nullable=False, default="inactive")
    capabilities = Column(Text, nullable=True)  # 存储JSON格式的能力描述
    registered_at = Column(DateTime(timezone=True), server_default=func.now())
    last_heartbeat_at = Column(DateTime(timezone=True), onupdate=func.now())
    current_task_id = Column(String, nullable=True, index=True)

这个模型定义了Agent的基本信息，包括ID、名称、类型、状态、能力、注册时间、最后心跳时间、当前任务ID等。

8.2.2 Task模型

在 app/models/task.py 文件中添加以下内容：

from sqlalchemy import Column, String, DateTime, Text, Integer, ForeignKey
from sqlalchemy.sql import func
from sqlalchemy.orm import relationship
from app.core.database import Base


class Task(Base):
    """任务模型"""
    __tablename__ = "tasks"

    id = Column(String, primary_key=True, index=True)
    type = Column(String, index=True, nullable=False)
    status = Column(String, index=True, nullable=False, default="pending")
    priority = Column(Integer, nullable=False, default=0)
    parameters = Column(Text, nullable=True)  # 存储JSON格式的任务参数
    result = Column(Text, nullable=True)  # 存储JSON格式的任务结果
    error_message = Column(Text, nullable=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
    started_at = Column(DateTime(timezone=True), nullable=True)
    completed_at = Column(DateTime(timezone=True), nullable=True)

    # 与Agent执行记录的关系
    executions = relationship("TaskExecution", back_populates="task")


class TaskExecution(Base):
    """任务执行记录模型"""
    __tablename__ = "task_executions"

    id = Column(String, primary_key=True, index=True)
    task_id = Column(String, ForeignKey("tasks.id"), nullable=False, index=True)
    agent