Home
Blog
OpenAgents x Milvus：如何构建共享内存的更智能的多代理系统

OpenAgents x Milvus：如何构建共享内存的更智能的多代理系统

Tutorials

November 24, 2025

Min Yin

大多数开发者都是从单个 Agents 开始开发代理系统的，后来才意识到他们基本上是在建造一个非常昂贵的聊天机器人。对于简单的任务，ReAct 类型的代理效果还不错，但很快就会达到极限：它无法并行运行步骤，无法跟踪较长的推理链，而且一旦添加太多工具，它就会分崩离析。多 Agents 设置有望解决这个问题，但它们也带来了自己的问题：协调开销、脆性交接，以及悄无声息地侵蚀模型质量的膨胀共享上下文。

OpenAgents是一个用于构建多代理系统的开源框架，在这个框架中，人工智能代理可以协同工作、共享资源，并在持久的社区内处理长期项目。OpenAgents 不需要单一的中央协调者，而是让代理以一种更加分布式的方式进行协作：它们可以相互发现、交流，并围绕共同目标进行协调。

与Milvus向量数据库搭配使用，该管道可获得一个可扩展的高性能长期内存层。Milvus 通过快速语义搜索、灵活的索引选择（如 HNSW 和 IVF）以及通过分区实现的干净隔离，为 Agents 内存提供动力，因此 Agents 可以存储、检索和重用知识，而不会淹没在上下文中或踩踏其他数据。

在这篇文章中，我们将介绍 OpenAgents 如何实现分布式多代理协作，Milvus 为什么是可扩展代理内存的重要基础，以及如何逐步组建这样一个系统。

构建真实世界代理系统的挑战

当今的许多主流 Agents 框架--LangChain、AutoGen、CrewAI 等--都是围绕以任务为中心的模型构建的。你启动一组 Agents，给它们布置任务，或许定义一个工作流程，然后让它们运行。这对于狭窄或短暂的用例非常有效，但在实际生产环境中，它暴露出三个结构性局限：

知识仍然是孤立的。Agents 的经验仅限于自己的部署。工程部门的代码审查 Agents 不会与评估可行性的产品团队 Agents 分享它所学到的知识。每个团队最终都要从头开始重建知识，既低效又脆弱。
协作僵化。即使在多 Agents 框架中，合作通常也依赖于事先定义的工作流程。当合作需要转变时，这些静态规则就无法适应，从而降低了整个系统的灵活性。
缺乏持久状态。大多数 Agents 都遵循一个简单的生命周期：启动 → 执行 → 关闭。它们会忘记运行之间的一切--上下文、关系、决策和交互历史。没有持久状态，Agent 就无法建立长期记忆或演化其行为。

这些结构性问题来自于将 Agents 视为孤立的任务执行者，而不是更广泛协作网络中的参与者。

OpenAgents 团队认为，未来的 Agents 系统需要的不仅仅是更强的推理能力--它们还需要一种机制，使 Agents 能够相互发现、建立关系、分享知识并动态地协同工作。重要的是，这种机制不应依赖于单一的中央控制器。互联网之所以行之有效，是因为它是分布式的--没有一个节点可以主宰一切，而且随着系统的发展，它也会变得更加强大和可扩展。多 Agents 系统也受益于同样的设计原则。这就是为什么 OpenAgents 摒弃了全能协调者的概念，转而支持分散的、网络驱动的合作。

什么是 OpenAgents？

OpenAgents 是一个用于构建人工智能代理网络的开源框架，可实现开放式合作，让人工智能代理协同工作、共享资源并解决长期项目。它为代理互联网提供了基础架构--在这个互联网中，代理与数百万其他代理在持久、不断发展的社区中公开协作。在技术层面，该系统围绕三个核心组件构建：Agents 网络、网络模块和运输工具。

1.Agents 网络：协作的共享环境

Agents 网络是一个共享环境，多个 Agents 可以在这个环境中连接、交流和协作，共同解决复杂的任务。其核心特征包括

持久操作符：一旦创建，网络就会保持在线，不受任何单一任务或工作流的影响。
动态代理：代理可使用网络 ID 随时加入；无需预先注册。
多协议支持：统一的抽象层支持通过 WebSocket、gRPC、HTTP 和 libp2p 进行通信。
自主配置：每个网络都有自己的权限、管理和资源。

只需一行代码，您就可以创建一个网络，任何 Agents 都可以通过标准接口立即加入。

2.网络模块：协作的可插拔扩展

Mods 提供了一个模块化的协作功能层，与核心系统保持分离。您可以根据自己的具体需求混合和匹配 Mods，从而实现针对每个用例量身定制的协作模式。

模块	目的	使用案例
工作区消息传递	实时信息交流	流式回复、即时反馈
论坛	异步讨论	提案审查、多轮审议
维基	共享知识库	知识整合、文件协作
社交	关系图谱	专家路由、信任网络

所有 Mods 都在统一的事件系统上操作，便于随时扩展框架或引入自定义行为。

3.传输：与协议无关的通信渠道

传输协议是允许异构代理在 OpenAgents 网络中连接和交换信息的通信协议。OpenAgents 支持可在同一网络内同时运行的多种传输协议，包括

HTTP/REST用于广泛的跨语言集成
用于低延迟双向通信的WebSocket
用于适合大规模集群的高性能RPC的gRPC
用于分散式点对点网络的libp2p
A2A，一种专为 Agents 对 Agents 通信设计的新兴协议

所有传输都通过统一的基于事件的消息格式操作，实现协议之间的无缝转换。您无需担心对等代理使用哪种协议，框架会自动处理。以任何语言或框架构建的 Agents 都可以加入 OpenAgents 网络，而无需重写现有代码。

将 OpenAgents 与 Milvus 集成，实现长期代理记忆

OpenAgents 解决了 Agents 如何交流、相互发现和协作的难题，但仅有协作是不够的。代理会产生见解、决策、对话历史、工具结果和特定领域的知识。如果没有持久内存层，所有这些都会在 Agents 关闭的那一刻烟消云散。

这就是Milvus的重要性所在。Milvus 提供高性能向量存储和语义检索，将 Agents 的交互转化为持久、可重复使用的内存。当集成到 OpenAgents 网络中时，它具有三大优势：

1.语义搜索

Milvus 使用 HNSW 和 IVF_FLAT 等索引算法提供快速语义搜索。Agents 可以根据意义而非关键字检索最相关的历史记录，使他们能够

回顾之前的决策或计划、
避免重复工作
在不同会话中保持长视距上下文。

这就是Agents 记忆的支柱：快速、相关、上下文检索。

2.十亿级水平可扩展性

真实的 Agents 网络会产生海量数据。Milvus 可在这种规模下轻松操作，提供以下功能

数十亿向量的存储和搜索、
< 即使在高吞吐量 Top-K 检索情况下，延迟时间也小于 30 毫秒、
完全分布式架构，可随需求增长线性扩展。

无论你有十几个 Agents 还是成千上万个并行工作的 Agents，Milvus 都能保持检索的快速性和一致性。

3.多租户隔离

Milvus 通过Partition Key 提供细粒度的多租户隔离，这是一种轻量级分区机制，可在单个 Collections 内分割内存。这允许

不同团队、项目或 Agents 社区维护独立的内存空间、
与维护多个 Collections 相比，大大降低了开销、
需要共享知识时，可选择跨分区检索。

这种隔离对于大型多代理部署至关重要，因为在这种部署中，必须在不影响检索速度的情况下尊重数据边界。

OpenAgents 通过直接调用 Milvus API 的自定义模块与 Milvus 连接。Agents 消息、工具输出和交互日志会自动嵌入向量并存储在 Milvus 中。开发人员可以自定义

Embeddings 模型、
存储 Schema 和元数据、
和检索策略（如混合搜索、分区搜索）。

这就为每个 Agents 社区提供了一个可扩展、持久并针对语义推理进行了优化的存储层。

如何使用 OpenAgent 和 Milvus 构建多代理聊天机器人

为了将事情具体化，让我们通过一个演示来了解一下：建立一个开发人员支持社区，让多个专业 Agents（Python 专家、数据库专家、DevOps 工程师等）合作回答技术问题。每个专家都会提供特定领域的推理，系统会自动将查询路由到最合适的 Agents，而不是依赖一个工作过度的通用代理。

本示例演示了如何将Milvus集成到 OpenAgents 部署中，为技术问答提供长期记忆。Agents 对话、过去的解决方案、故障排除日志和用户查询都被转换成向量嵌入并存储在 Milvus 中，从而使网络具备以下能力：

记住以前的答案、
重复使用先前的技术解释
在不同会话中保持一致，并
随着交互次数的增加而不断改进。

前提条件

python3.11 以上
conda
Openai-key

1.定义依赖关系

定义项目所需的 Python 软件包：

# Core framework
openagents>=0.6.11
# Vector database
pymilvus>=2.5.1
# Embedding model
sentence-transformers>=2.2.0
# LLM integration
openai>=1.0.0
# Environment config
python-dotenv>=1.0.0

2.环境变量

这里是您的环境配置模板：

# LLM configuration (required)
OPENAI_API_KEY=your_openai_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o
# Milvus configuration
MILVUS_URI=./multi_agent_memory.db
# Embedding model configuration
EMBEDDING_MODEL=text-embedding-3-large
EMBEDDING_DIMENSION=3072
# Network configuration
NETWORK_HOST=localhost
NETWORK_PORT=8700
STUDIO_PORT=8050

3.配置 OpenAgents 网络

定义 Agents 网络结构及其通信设置：

# Network transport protocol (HTTP on port 8700)
# Multi-channel messaging system (general, coordination, expert channels)
# Agent role definitions (coordinator, python_expert, etc.)
# Milvus integration settings
network:
  name: "Multi-Agent Collaboration Demo"
  transports:
    - type: "http"
      config:
        port: 8700
        host: "localhost"
  mods:
    - name: "openagents.mods.workspace.messaging"
      config:
        channels:
          - name: "general"          # User question channel
          - name: "coordination"     # Coordinator channel
          - name: "python_channel"   # Python expert channel
          - name: "milvus_channel"   # Milvus expert channel
          - name: "devops_channel"   # DevOps expert channel
  agents:
    coordinator:
      type: "coordinator"
      description: "Coordinator Agent, responsible for analyzing queries and dispatching tasks to expert agents"
      channels: ["general", "coordination"]
    python_expert:
      type: "expert"
      domain: "python"

4.实现多代理协作

下面显示的是核心代码片段（不是完整的实现）。

# SharedMemory: Milvus’s SharedMemory system
# CoordinatorAgent: Coordinator Agent, responsible for analyzing queries and dispatching tasks to expert agents
# PythonExpertAgent: Python Expert
# MilvusExpertAgent: Milvus Expert
# DevOpsExpertAgent: DevOps Expert
import os
import asyncio
import json
from typing import List, Dict
from dotenv import load_dotenv
from openagents.agents.worker_agent import WorkerAgent
from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType
import openai
load_dotenv()
class SharedMemory:
    """SharedMemory in Milvus for all Agents"""
    def __init__(self):
        connections.connect(uri="./multi_agent_memory.db")
        self.setup_collections()
        self.openai_client = openai.OpenAI(
            api_key=os.getenv("OPENAI_API_KEY"),
            base_url=os.getenv("OPENAI_BASE_URL")
        )
    def setup_collections(self):
        """Create memory collections: expert knowledge, collaboration history, problem solutions"""
        collections = {
            "expert_knowledge": "expert knowledge",
            "collaboration_history": "collaboration history", 
            "problem_solutions": "problem solutions"
        }
        # Code to create vector collections...
    async def search_knowledge(self, query: str, collection_name: str):
        """Search for relevant knowledge"""
        # Vector search implementation...
    async def store_knowledge(self, agent_id: str, content: str, metadata: dict, collection_name: str):
        """Store knowledge"""
        # Store into the vector database...
class CoordinatorAgent(WorkerAgent):
    """Coordinator Agent - analyzes questions and coordinates other Agent"""
    def __init__(self):
        super().__init__(agent_id="coordinator")
        self.expert_agents = {
            "python": "python_expert",
            "milvus": "milvus_expert", 
            "devops": "devops_expert"
        }
    async def analyze_question(self, question: str) -> List[str]:
        """Determine which experts are needed for the question"""
        keywords = {
            "python": ["python", "django", "flask", "async"],
            "milvus": ["milvus", "vector", "index", "performance"],
            "devops": ["deployment", "docker", "kubernetes", "operations"]
        }
        # Keyword matching logic...
        return needed_experts
    async def coordinate_experts(self, question: str, needed_experts: List[str]):
        """Coordinate collaboration among expert Agent"""
        # 1. Notify experts to begin collaborating
        # 2. Dispatch tasks to each expert
        # 3. Collect expert responses
        # 4. Return expert opinions
    async def on_channel_post(self, context):
        """Main logic for handling user questions"""
        content = context.incoming_event.payload.get('content', {}).get('text', '')
        if content and not content.startswith('🎯'):
            # 1. Analyze question → 2. Coordinate experts → 3. Merge answers → 4. Reply to user
class PythonExpertAgent(WorkerAgent):
    """Python Expert Agent"""
    async def analyze_python_question(self, question: str) -> str:
        """Analyze Python-related questions and provide expert advice"""
        # 1. Search for relevant experience
        # 2. Use LLM to generate expert response
        # 3. Store result in collaboration history
        return answer
# Start all Agens
async def run_multi_agent_demo():
    coordinator = CoordinatorAgent()
    python_expert = PythonExpertAgent()
    milvus_expert = MilvusExpertAgent()
    devops_expert = DevOpsExpertAgent()
    # Connect to the OpenAgents network
    await coordinator.async_start(network_host="localhost", network_port=8700)
    # ... Start other Agent
    while True:
        await asyncio.sleep(1)
if __name__ == "__main__":
    asyncio.run(run_multi_agent_demo())

5.创建并激活虚拟环境

conda create -n openagents
conda activate openagents

安装依赖项

pip install -r requirements.txt

配置 API 密钥

cp .env.example .env

启动 OpenAgents 网络

openagents network start .

启动多代理服务

python multi_agent_demo.py

启动 OpenAgents Studio

openagents studio -s

访问工作室

http://localhost:8050

检查 Agents 和网络的状态：

结论

OpenAgents 提供了一个协调层，让 Agents 可以相互发现、交流和协作，而 Milvus 则解决了如何存储、共享和重用知识这一同样重要的问题。通过提供高性能向量记忆层，Milvus 使 Agents 能够建立持久的上下文，回忆过去的交互，并随着时间的推移积累专业知识。它们共同推动人工智能系统超越孤立模型的限制，走向真正的多 Agents 网络更深层次的协作潜力。

当然，任何多代理架构都需要权衡利弊。并行运行 Agents 可能会增加令牌消耗，错误可能会在 Agents 之间串联，同时决策可能会导致偶尔的冲突。这些都是需要积极研究和不断改进的领域，但它们并不会降低构建能够协调、记忆和进化的系统的价值。

准备好让你的 Agents 拥有长期记忆了吗？

了解Milvus并尝试将其与您自己的工作流程整合。

有问题或想深入了解任何功能？加入我们的 Discord 频道或在 GitHub 上提交问题。您还可以通过 Milvus Office Hours 预订 20 分钟的一对一课程，以获得见解、指导和问题解答。