多智能体协作

使用 AgenticX 构建多智能体系统。

多智能体协作

概述

AgenticX 从设计之初就面向多智能体系统。多个智能体可通过委派、并行执行与结构化通信协议协作完成复杂任务。

智能体团队

python

1from agenticx.runtime import AgentTeamManager
2from agenticx import Agent
3from agenticx.llms import OpenAIProvider
4
5llm = OpenAIProvider(model="gpt-4o")
6
7# Define team members
8researcher = Agent(id="researcher", name="Researcher", role="Information Gatherer",
9                   goal="Find accurate information", organization_id="team")
10analyst = Agent(id="analyst", name="Analyst", role="Data Analyst",
11                goal="Analyze and interpret data", organization_id="team")
12writer = Agent(id="writer", name="Writer", role="Content Writer",
13               goal="Produce clear written content", organization_id="team")
14
15# Team manager handles concurrency, session isolation, and agent lifecycle
16team = AgentTeamManager(agents=[researcher, analyst, writer], max_concurrency=3)

Meta-Agent 模式

Meta-Agent 充当 CEO/项目经理，将工作分派给专职子智能体：

1User Request
2    ↓
3Meta-Agent (analyzes, plans, delegates)
4    ↓
5┌───────────────────────────────┐
6│  Researcher │ Analyst │ Writer │  ← Sub-agents running concurrently
7└───────────────────────────────┘
8    ↓
9Meta-Agent (aggregates, synthesizes)
10    ↓
11Final Response to User

Meta-Agent 维护所有运行中子智能体的活跃快照，并在每轮对话中将其状态注入系统提示。

A2A 通信协议

智能体可通过 A2A（Agent-to-Agent）协议直接通信：

python

1from agenticx.protocols.a2a import A2AClient, A2AServer, AgentCard
2
3# Publish an agent as an A2A service
4card = AgentCard(
5    agent_id="researcher",
6    skills=["web_search", "document_analysis"],
7    endpoint="http://localhost:8001"
8)
9server = A2AServer(agent=researcher, card=card)
10server.start()
11
12# Another agent calls it
13client = A2AClient()
14result = client.invoke_skill(
15    agent_id="researcher",
16    skill="web_search",
17    params={"query": "latest AI papers"}
18)

并行执行

同时运行多个智能体：

python

1from agenticx.flow import ParallelExecutor
2
3executor = ParallelExecutor(max_workers=4)
4
5tasks = [
6    (researcher, Task(description="Research topic A")),
7    (researcher, Task(description="Research topic B")),
8    (researcher, Task(description="Research topic C")),
9]
10
11results = executor.run_all(tasks)

人在回路（Human-in-the-Loop）

暂停智能体执行以获取人工审批：

python

1from agenticx.runtime import HumanInTheLoop
2
3hitl = HumanInTheLoop(
4    trigger_on=["tool_call:delete_file", "tool_call:send_email"],
5    timeout_seconds=300,
6    default_action="reject"  # auto-reject if no human response
7)
8
9executor = AgentExecutor(agent=agent, llm=llm, human_in_the_loop=hitl)

会话隔离

每次智能体团队运行通过 owner_session_id 隔离，避免并发会话间相互污染。全局注册表支持跨会话查询智能体状态，便于监控。