第 12 期 | Agent 协作：多 Agent 通信与任务委派

好的，这是为 Hermes Agent 教程撰写的第 12 期文章。

---

副标题：探索 Hermes 的 Agent-to-Agent 通信协议，如何搭建一个由多个专业化 Agent 组成的"AI团队"

学习目标

在本期课程中，你将深入学习 Hermes Agent 框架中最激动人心的功能之一：多 Agent 协作。完成本章后，你将能够：

1. 理解多 Agent 协作的必要性：明白为什么将复杂任务分解并委派给多个专业化 Agent 是构建高级 AI 应用的关键。
2. 掌握核心协作架构：深入了解 Hermes Agent 协作的三个核心组件：Agent Registry、HIACP 协议和内部消息总线。
3. 配置并启动多个 Agent：学习如何为每个 Agent 配置唯一的身份和通信端点，并让它们在网络中互相发现。
4. 实现任务的委派与响应：通过编写一个新的 Skill，实践如何在一个 Agent 中调用另一个 Agent 的能力，并处理返回结果。
5. 构建并运行一个 "AI 团队"：亲手搭建一个由“项目经理”、“研究员”和“程序员”三个 Agent 组成的团队，协同完成一个复杂任务。

---

核心概念讲解

随着我们构建的 Agent 应用越来越复杂，我们很快会遇到“单一全能 Agent” (Monolithic Agent) 的瓶颈。一个试图精通所有领域的 Agent 往往在任何领域都无法达到专家水平。就像在人类世界中，我们依赖团队协作一样，AI Agent 的未来也必然走向群体智能和协作。

Hermes Agent 框架为此提供了一套完整的 Agent-to-Agent (A2A) 通信与协作机制。其设计哲学借鉴了成熟的软件工程概念，如微服务架构 (Microservices Architecture) 和演员模型 (Actor Model)。

The "AI Team" Paradigm: AI 团队范式

想象一下，你要完成一个任务：“调研最新的 AI 技术‘Transformer’，编写一个简单的代码示例，并总结成一篇技术博客。”

一个单一的 Agent 处理这个任务的流程可能是：

1. 调用搜索工具，浏览大量网页。
2. 在内心（上下文）中消化和理解信息。
3. 切换到代码生成模式，编写代码。
4. 再次切换到写作模式，整理成文。

这个过程效率低下且容易出错。上下文切换成本高，且 Agent 可能会在不同任务模式间“精神分裂”。

而一个 "AI 团队" 的工作模式则完全不同：

• 项目经理 Agent (Project Manager Agent)：作为总入口，接收用户的高级指令。它不亲自执行具体任务，而是负责任务分解和协调。
• 研究员 Agent (Researcher Agent)：专门负责信息检索。它拥有强大的 web_search、read_document 等 Skills，并配置了专门的搜索引擎 API Key。它的核心使命是快速、准确地提供信息。
• 程序员 Agent (Coder Agent)：精通多种编程语言，拥有 write_code、debug_code、run_test 等 Skills。它接收明确的编程需求，并产出高质量的代码。
• 作家 Agent (Writer Agent)：擅长文本润色、格式化和风格转换。它负责将原始信息和代码整合成一篇通顺、专业的文章。

这种模式的优势是显而易见的：

• 专业化 (Specialization)：每个 Agent 都可以使用针对其特定任务最优化的模型、Prompt 和 Skills。
• 可扩展性 (Scalability)：可以轻松地为团队增加新的专家 Agent，如“数据分析师 Agent”或“设计师 Agent”。
• 鲁棒性 (Robustness)：单个 Agent 的失败不会导致整个系统崩溃。项目经理 Agent 可以将任务重新委派给备用 Agent。
• 并发性 (Concurrency)：多个任务可以并行执行，例如，在研究员 Agent 搜集资料的同时，程序员 Agent 可以开始搭建代码框架。

为了实现这一范式，Hermes 提供了以下三大核心组件。

1. Agent Registry (代理注册中心)

Agent 如何知道网络中存在哪些其他的 Agent，以及它们分别擅长什么？答案就是 Agent Registry。

Agent Registry 是一个服务发现 (Service Discovery) 组件。当一个 Hermes Agent 启动时，如果开启了协作模式，它会向 Registry 注册自己的信息，通常包括：

• agent_id: 全局唯一的 Agent 名称，如 researcher-01。
• address: Agent 的网络地址和端口，如 127.0.0.1:8001。
• capabilities: 该 Agent 拥有的技能列表，如 ["web_search", "summarize_text"]。这使得其他 Agent 可以根据能力来查找服务。

当一个 Agent (如 PM Agent) 需要委派任务时，它会首先向 Registry 查询：“谁拥有 web_search 这个 Skill？” Registry 会返回所有匹配的 Agent 列表，PM Agent 就可以选择一个进行通信。

在 Hermes 中，任何一个 Agent 都可以被配置为兼任 Registry 的角色，以简化部署。在生产环境中，则建议部署一个独立的 Registry 服务。

2. Hermes Inter-Agent Communication Protocol (HIACP)

一旦 Agent 发现了彼此，它们需要一种共同的语言来交流。这就是 HIACP 协议的作用。HIACP 是一种标准化的 JSON 消息格式，定义了 Agent 之间交互的结构和语义。它的设计灵感来源于经典的 FIPA-ACL (Agent Communication Language)。

一个典型的 HIACP 消息结构如下：

{
  "protocol_version": "1.0",
  "message_id": "uuid-1234-abcd-5678",
  "conversation_id": "conv-xyz-987",
  "sender_id": "pm-agent-main",
  "receiver_id": "researcher-01",
  "performative": "DELEGATE_TASK",
  "content": {
    "task_name": "research_topic",
    "parameters": {
      "topic": "What is the Transformer architecture in AI?",
      "depth": "detailed"
    }
  },
  "timestamp": "2023-10-27T10:00:00Z"
}

关键字段解析：

• message_id: 每条消息的唯一标识符。
• conversation_id: 用于追踪一个完整的多步对话或任务流程。
• sender_id / receiver_id: 发送者和接收者的 agent_id。
• performative: 消息的意图，这是 HIACP 的核心。常见的 performative 包括：
  • REQUEST: 请求对方执行一个动作（通常是调用一个 Skill）。
  • INFORM: 发送一个陈述性事实或结果。
  • QUERY_REF: 查询某个信息。
  • DELEGATE_TASK: 将一个完整的子任务委派给对方。
  • ACCEPT / REFUSE: 同意或拒绝一个请求。
  • FAILURE: 通知任务执行失败。
• content: 消息的具体内容，其结构由任务本身决定。

通过遵循 HIACP，Hermes 确保了不同开发者、甚至不同版本的 Agent 之间都能够进行可预测、可靠的通信。

3. 内部消息总线 (Internal Message Bus)

HIACP 定义了“说什么”，而消息总线则解决了“怎么说”的问题。它负责消息的可靠传输。Hermes 内置了一个基于 ZeroMQ 的轻量级异步消息总线。

当你通过 Agent 的协作 API (如 agent.collaborate.send(...)) 发送一条 HIACP 消息时，实际上是将其推入到这个消息总线。总线会处理网络连接、消息序列化、超时和重试等底层细节，让 Skill 开发者可以专注于业务逻辑。

这种解耦的设计意味着，即使目标 Agent 暂时离线，消息总线也可以在一定时间内缓存消息，待其上线后再次投递，极大地提高了系统的韧性。

---

💻 实战演示

现在，让我们动手搭建一个由三个 Agent 组成的 "AI 团队"，来完成我们之前提到的任务：“请调研一下什么是 LangChain，并写一个简单的 Python 示例代码，最后整理成一篇报告。”

我们的团队成员：

1. pm_agent: 项目经理，用户入口，负责任务分解和协调。它将兼任 Agent Registry。
2. research_agent: 研究员，只负责调用 web_search Skill。
3. coder_agent: 程序员，只负责根据需求编写 Python 代码。

Step 1: 环境准备与目录结构

首先，创建我们的项目根目录和三个 Agent 的子目录。

mkdir hermes_ai_team
cd hermes_ai_team

mkdir pm_agent
mkdir pm_agent/skills

mkdir research_agent
mkdir research_agent/skills

mkdir coder_agent
mkdir coder_agent/skills

Step 2: Agent 配置

我们需要为每个 Agent 创建一个 config.yml 文件。

1. pm_agent/config.yml

这个 Agent 是我们的核心，它将作为 Registry。

# pm_agent/config.yml
agent:
  name: ProjectManagerAgent
  description: The main agent that interacts with the user and delegates tasks.

# 使用本地的 Ollama 模型作为大脑
llm_provider:
  default: ollama
  ollama:
    model: llama3:8b
    api_base: http://localhost:11434

# 开启协作模式
collaboration:
  enabled: true
  agent_id: pm_agent_main  # 全局唯一的 ID
  host: 0.0.0.0
  port: 8000
  # 将此 Agent 设置为 Registry 服务
  registry:
    enabled: true
    host: 0.0.0.0
    port: 8080 # Registry 服务的端口

2. research_agent/config.yml

这个 Agent 是一个客户端，它需要知道 Registry 在哪里。

# research_agent/config.yml
agent:
  name: ResearcherAgent
  description: A specialized agent for searching the web and summarizing information.

llm_provider:
  default: ollama
  ollama:
    model: llama3:8b
    api_base: http://localhost:11434

collaboration:
  enabled: true
  agent_id: researcher_01
  host: 0.0.0.0
  port: 8001 # 自身服务的端口
  # 指向 PM Agent 提供的 Registry 服务
  registry_url: http://127.0.0.1:8080 

3. coder_agent/config.yml

与 research_agent 类似，也指向同一个 Registry。

# coder_agent/config.yml
agent:
  name: CoderAgent
  description: A specialized agent for writing and explaining code snippets.

llm_provider:
  default: ollama
  ollama:
    model: llama3:8b
    api_base: http://localhost:11434

collaboration:
  enabled: true
  agent_id: coder_01
  host: 0.0.0.0
  port: 8002 # 自身服务的端口
  registry_url: http://127.0.0.1:8080

Step 3: 创建专业化 Skills

现在，我们为 research_agent 和 coder_agent 添加它们的专业技能。

1. research_agent/skills/web_research.py

为了简化，我们这里用一个伪代码来模拟网络搜索。在真实场景中，你可以集成 Tavily, Serper 等搜索 API。

# research_agent/skills/web_research.py
from hermes_agent.skills import Skill

class WebResearchSkill(Skill):
    name = "web_research"
    description = "Performs web research on a given topic and returns a summary."

    def __init__(self, agent):
        super().__init__(agent)

    async def execute(self, topic: str) -> str:
        """
        Simulates web research.
        In a real-world scenario, this would use a search API.
        """
        self.agent.logger.info(f"Received research request for topic: {topic}")
        if "langchain" in topic.lower():
            return (
                "LangChain is a framework for developing applications powered by "
                "language models. It provides a standard interface for chains, "
                "a plethora of integrations with other tools, and end-to-end chains "
                "for common applications. Key components include Models, Prompts, "
                "Indexes, Chains, and Agents."
            )
        return f"Sorry, I couldn't find information on {topic}."

2. coder_agent/skills/code_generator.py

这个 Skill 接收需求并生成代码。

# coder_agent/skills/code_generator.py
from hermes_agent.skills import Skill

class CodeGeneratorSkill(Skill):
    name = "generate_python_example"
    description = "Generates a simple Python code example based on a description."

    def __init__(self, agent):
        super().__init__(agent)

    async def execute(self, requirement: str) -> str:
        """
        Simulates code generation.
        In a real-world scenario, this would involve a call to an LLM with a specific prompt.
        """
        self.agent.logger.info(f"Received code generation request for: {requirement}")
        if "langchain" in requirement.lower() and "hello world" in requirement.lower():
            code = """
# main.py
from langchain_community.llms import Ollama
from langchain_core.prompts import ChatPromptTemplate

# Initialize the LLM
llm = Ollama(model="llama3:8b")

# Create a prompt template
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{input}")
])

# Create a chain
chain = prompt | llm 

# Invoke the chain
response = chain.invoke({"input": "Hello, LangChain!"})
print(response)
"""
            return f"Here is a simple 'Hello World' example for LangChain:\n```python\n{code.strip()}\n```"
        return "Sorry, I can't generate code for that requirement."

Step 4: 编写核心委派 Skill

这是本次实战的核心。我们在 pm_agent 中创建一个 Skill，它将负责调用其他两个 Agent。

pm_agent/skills/research_and_code_coordinator.py

# pm_agent/skills/research_and_code_coordinator.py
import asyncio
from hermes_agent.skills import Skill
from hermes_agent.collaboration import HIACPMessage, Performative

class ResearchAndCodeCoordinatorSkill(Skill):
    name = "coordinate_research_and_code"
    description = (
        "Coordinates research and coding tasks by delegating to specialized agents. "
        "Use this for complex queries involving both information retrieval and code generation."
    )

    def __init__(self, agent):
        super().__init__(agent)

    async def execute(self, topic: str) -> str:
        self.agent.logger.info(f"Starting coordination for topic: {topic}")

        # --- Step 1: Delegate research task ---
        self.agent.logger.info("Finding a research agent...")
        # Discover an agent that has the 'web_research' capability
        # The collaboration module handles querying the registry
        researcher_id = await self.agent.collaboration.discover_agent_by_skill("web_research")
        
        if not researcher_id:
            return "Error: Could not find any agent with 'web_research' capability."
        
        self.agent.logger.info(f"Found researcher: {researcher_id}. Delegating research task...")
        
        # Construct the HIACP message
        research_task = HIACPMessage(
            receiver_id=researcher_id,
            performative=Performative.REQUEST,
            content={"skill": "web_research", "parameters": {"topic": f"What is {topic}?"}}
        )
        
        # Send the message and wait for the response. The timeout is in seconds.
        research_response = await self.agent.collaboration.send_and_wait(research_task, timeout=30)
        
        if not research_response or research_response.performative == Performative.FAILURE:
            self.agent.logger.error("Research task failed or timed out.")
            return "Error: The research task failed."
        
        research_summary = research_response.content.get("result")
        self.agent.logger.info(f"Received research summary: {research_summary[:100]}...")

        # --- Step 2: Delegate coding task ---
        self.agent.logger.info("Finding a coder agent...")
        coder_id = await self.agent.collaboration.discover_agent_by_skill("generate_python_example")

        if not coder_id:
            return "Error: Could not find any agent with 'generate_python_example' capability."
            
        self.agent.logger.info(f"Found coder: {coder_id}. Delegating coding task...")

        coding_requirement = f"A simple 'hello world' style example for {topic}, based on this info: {research_summary}"
        
        code_task = HIACPMessage(
            receiver_id=coder_id,
            performative=Performative.REQUEST,
            content={"skill": "generate_python_example", "parameters": {"requirement": coding_requirement}}
        )
        
        code_response = await self.agent.collaboration.send_and_wait(code_task, timeout=30)
        
        if not code_response or code_response.performative == Performative.FAILURE:
            self.agent.logger.error("Coding task failed or timed out.")
            return "Error: The coding task failed."
            
        code_example = code_response.content.get("result")
        self.agent.logger.info("Received code example.")

        # --- Step 3: Synthesize the final report ---
        self.agent.logger.info("Synthesizing the final report...")
        final_prompt = f"""
        You are a tech writer. Based on the following information, please generate a final report.
        The report should include an introduction based on the research summary and a code section with the provided example.

        Research Summary:
        {research_summary}

        Code Example:
        {code_example}
        """

        # Use the agent's own LLM to generate the final output
        final_report = await self.agent.llm.invoke(final_prompt)
        
        return final_report

Step 5: 启动并测试 "AI 团队"

现在，激动人心的时刻到了。我们需要打开 三个 终端窗口。

终端 1: 启动 PM Agent (兼 Registry)

hermes start --config hermes_ai_team/pm_agent/config.yml

你会看到日志显示 ProjectManagerAgent 已启动，并且 Registry 服务正在 0.0.0.0:8080 上监听。

终端 2: 启动 Researcher Agent

hermes start --config hermes_ai_team/research_agent/config.yml

在 pm_agent 的日志中，你应该能看到一条类似 [Registry] Agent researcher_01 registered 的消息。

终端 3: 启动 Coder Agent

hermes start --config hermes_ai_team/coder_agent/config.yml

同样，pm_agent 的日志中会显示 [Registry] Agent coder_01 registered。

终端 4: 与 PM Agent 交互

现在我们的 AI 团队已经准备就绪。打开第四个终端，使用 Hermes CLI 与 pm_agent 对话。

# PM Agent 运行在 8000 端口
hermes chat --port 8000

进入聊天界面后，输入我们的任务指令：

> 请调研一下什么是 LangChain，并写一个简单的 Python 示例代码，最后整理成一篇报告。

现在，观察所有三个 Agent 的终端日志。你会看到一个精彩的协作流程：

1. PM Agent 日志:

   • INFO: Starting coordination for topic: LangChain
   • INFO: Finding a research agent...
   • INFO: Found researcher: researcher_01. Delegating research task...
   • INFO: Received research summary: LangChain is a framework...
   • INFO: Finding a coder agent...
   • INFO: Found coder: coder_01. Delegating coding task...
   • INFO: Received code example.
   • INFO: Synthesizing the final report...

2. Researcher Agent 日志:

   • INFO: Received research request for topic: What is LangChain?
   • (日志显示任务完成并返回结果)

3. Coder Agent 日志:

   • INFO: Received code generation request for: A simple 'hello world' style example for LangChain...
   • (日志显示任务完成并返回结果)

最终，pm_agent 会将整合后的完整报告返回给你，内容类似：

当然，这是关于 LangChain 的调研报告和代码示例。

### 简介

LangChain 是一个用于开发由语言模型驱动的应用程序的框架。它为链（Chains）提供了一个标准接口，与各种工具有大量的集成，并为常见应用提供了端到端的链。其核心组件包括模型（Models）、提示（Prompts）、索引（Indexes）、链（Chains）和代理（Agents）。

### Python 代码示例

这是一个使用 LangChain 和 Ollama 的简单“Hello World”示例：

```python
# main.py
from langchain_community.llms import Ollama
from langchain_core.prompts import ChatPromptTemplate

# Initialize the LLM
llm = Ollama(model="llama3:8b")

# Create a prompt template
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{input}")
])

# Create a chain
chain = prompt | llm 

# Invoke the chain
response = chain.invoke({"input": "Hello, LangChain!"})
print(response)

恭喜！你已成功搭建并指挥了一个由多个专业化 Agent 组成的 AI 团队！

---

## 涉及命令

*   `mkdir <directory_name>`: 创建新目录。
*   `hermes start --config <path_to_config.yml>`: 根据指定的配置文件启动一个 Hermes Agent 实例。
*   `hermes chat --port <port_number>`: 通过命令行与在指定端口运行的 Hermes Agent 进行交互。

---

##