设计#

分层设计：拆分为两个核心模块

• autogen-core：作为框架的底层基础，封装了与语言模型交互、消息传递等核心功能。它的存在保证了框架的稳定性和未来扩展性。
• autogen-agentchat：构建于 core 之上，提供了用于开发对话式智能体应用的高级接口，简化了多智能体应用的开发流程。 这种分层策略使得各组件职责明确，降低了系统的耦合度。 异步优先：在多智能体协作场景中，网络请求（如调用 LLM API）是主要耗时操作。异步模式提升了并发处理能力和系统资源的利用效率。

核心智能体组件#

• AssistantAgent (助理智能体)： 
  • 这是任务的主要解决者，其核心是封装了一个大型语言模型（LLM）。
  • 它的职责是根据对话历史生成富有逻辑和知识的回复，例如提出计划、撰写文章或编写代码。通过不同的系统消息（System Message），我们可以为其赋予不同的“专家”角色。
• UserProxyAgent (用户代理智能体)： 这是 AutoGen 中功能独特的组件。
  • 它扮演着双重角色：既是人类用户的“代言人”，负责发起任务和传达意图；
  • 又是一个可靠的“执行器”，可以配置为执行代码或调用工具，并将结果反馈给其他智能体。
  • 这种设计清晰地区分了“思考”（由 AssistantAgent 完成）与“行动”。

对话机制#

• 轮询群聊 (RoundRobinGroupChat)： 这是一种明确的、顺序化的对话协调机制。让参与的智能体按照预定义的顺序依次发言。这种模式非常适用于流程固定的任务，例如一个典型的软件开发流程：产品经理先提出需求，然后工程师编写代码，最后由代码审查员进行检查。
• 工作流：
  1. 首先，创建一个 RoundRobinGroupChat 实例，并将所有参与协作的智能体（如产品经理、工程师等）加入其中。
  2. 当一个任务开始时，群聊会按照预设的顺序，依次激活相应的智能体。
  3. 被选中的智能体根据当前的对话上下文进行响应。
  4. 群聊将新的回复加入对话历史，并激活下一个智能体。
  5. 这个过程会持续进行，直到达到最大对话轮次或满足预设的终止条件。

通过这种方式，AutoGen 将复杂的协作关系，简化为一个流程清晰、易于管理的自动化“圆桌会议”。开发者只需定义好每个团队成员的角色和发言顺序，剩下的协作流程便可由群聊机制自主驱动。

分层架构体系：#

• Memory（记忆管理）
• Model API（模型调用层）
• Tool（工具封装）

第二层：躯干 —— 智能体基础设施层（Agent-level Infrastructure）

• 内置智能体（Built-in Agents）：框架直接给你几个“半成品”Agent，比如能操作浏览器的、能做深度研究的、能做复杂任务规划的。你可以直接拿来用，或者二次改造。

• ReAct 范式实现：这是让 Agent “边想边做”的核心机制。框架帮你把“推理（Reason）”和“行动（Act）”的循环流程写好了，并额外支持：

  • 钩子（Hooks）
  • 并行调用工具：Agent 一次可以同时调用多个工具，提高效率。
  • 结构化输出：强制 Agent 按指定格式（如 JSON）输出，方便程序解析。
  • 状态/会话管理：统一管理 Agent 的内部状态，解决你之前关心的“状态持久化”问题。
  • 异步执行与实时控制：支持 Agent 在后台“边跑边等”，并且你可以在运行中“插手”调整它的方向（这是 AgentScope 的一大亮点）。

第三层：协作网 —— 多智能体协作层（Multi-Agent Cooperation） 解决“多个 Agent 怎么配合干活”：

• MsgHub（消息中心）：相当于一个“邮局”，负责所有 Agent 之间的消息转发、路由和状态同步。Agent 们不用互相知道对方地址，只跟 MsgHub 打交道就行。

• Pipeline（工作流管道）：提供“流水线”编排能力。你可以设定 Agent 们是按顺序干（串行）、还是一起干（并行）、或者更复杂的模式。这让你能把多个 Agent 组织成一个高效的“流水线工厂”。

第四层：交付与运维 —— 开发与部署层（Development & Deployment） 这一层负责“从写代码到上线

消息驱动#

AgentScope 的核心创新在于其消息驱动架构。在这个架构中，所有的智能体交互都被抽象为消息的发送和接收，而不是传统的函数调用。

1
from agentscope.message import Msg
2

3
# 消息的标准结构
4
message = Msg(
5
    name="Alice",           # 发送者名称
6
    content="Hello, Bob!",  # 消息内容
7
    role="user",           # 角色类型
8
    metadata={             # 元数据信息
9
        "timestamp": "2024-01-15T10:30:00Z",
10
        "message_type": "text",
11
        "priority": "normal"
12
    }
13
)

将消息作为交互的基础单元，带来了几个关键优势：

• 异步解耦: 消息的发送方和接收方在时间上解耦，无需相互等待，天然支持高并发场景。
• 位置透明: 智能体无需关心另一个智能体是在本地进程还是在远程服务器上，消息系统会自动处理路由。
• 可观测性: 每一条消息都可以被记录、追踪和分析，极大地简化了复杂系统的调试与监控。

智能体生命周期管理#

在 AgentScope 中，每个智能体都有明确的生命周期（初始化、运行、暂停、销毁等），并基于一个统一的基类 AgentBase 来实现。开发者通常只需要关注其核心的 reply 方法。

1
from agentscope.agents import AgentBase
2

3
class CustomAgent(AgentBase):
4
    def __init__(self, name: str, **kwargs):
5
        super().__init__(name=name, **kwargs)
6
        # 智能体初始化逻辑
7

8
    def reply(self, x: Msg) -> Msg:
9
        # 智能体的核心响应逻辑
10
        response = self.model(x.content)
11
        return Msg(name=self.name, content=response, role="assistant")
12

13
    def observe(self, x: Msg) -> None:
14
        # 智能体的观察逻辑（可选）
15
        self.memory.add(x)

消息传递机制#

AgentScope 内置了一个消息中心 (MsgHub)，它是整个消息驱动架构的中枢。

全局状态（State）#

整个图的执行过程都围绕一个共享的状态对象进行。所有的节点都能读取和更新这个中心状态。

1
from typing import TypedDict, List
2

3
# 定义全局状态的数据结构
4
class AgentState(TypedDict):
5
    messages: List[str]      # 对话历史
6
    current_task: str        # 当前任务
7
    final_answer: str        # 最终答案
8
    # ... 任何其他需要追踪的状态

节点（Nodes）#

每个节点都是一个接收当前状态作为输入、并返回一个更新后的状态作为输出的 Python 函数。节点是执行具体工作的单元。

1
# 定义一个“规划者”节点函数
2
def planner_node(state: AgentState) -> AgentState:
3
    """根据当前任务制定计划，并更新状态。"""
4
    current_task = state["current_task"]
5
    # ... 调用LLM生成计划 ...
6
    plan = f"为任务 '{current_task}' 生成的计划..."
7

8
    # 将新消息追加到状态中
9
    state["messages"].append(plan)
10
    return state
11

12
# 定义一个“执行者”节点函数
13
def executor_node(state: AgentState) -> AgentState:
14
    """执行最新计划，并更新状态。"""
15
    latest_plan = state["messages"][-1]
16
    # ... 执行计划并获得结果 ...
17
    result = f"执行计划 '{latest_plan}' 的结果..."
18

19
    state["messages"].append(result)
20
    return state

边（Edges）#

边负责连接节点，定义工作流的方向。最简单的边是常规边，它指定了一个节点的输出总是流向另一个固定的节点。而 LangGraph 最强大的功能在于条件边。它通过一个函数来判断当前的状态，然后动态地决定下一步应该跳转到哪个节点。这正是实现循环和复杂逻辑分支的关键。

1
def should_continue(state: AgentState) -> str:
2
    """条件函数：根据状态决定下一步路由。"""
3
    # 假设如果消息少于3条，则需要继续规划
4
    if len(state["messages"]) < 3:
5
        # 返回的字符串需要与添加条件边时定义的键匹配
6
        return "continue_to_planner"
7
    else:
8
        state["final_answer"] = state["messages"][-1]
9
        return "end_workflow"

在定义了状态、节点和边之后，我们可以像搭积木一样将它们组装成一个可执行的工作流。

1
from langgraph.graph import StateGraph, END
2

3
# 初始化一个状态图，并绑定我们定义的状态结构
4
workflow = StateGraph(AgentState)
5

6
# 将节点函数添加到图中
7
workflow.add_node("planner", planner_node)
8
workflow.add_node("executor", executor_node)
9

10
# 设置图的入口点
11
workflow.set_entry_point("planner")
12

13
# 添加常规边，连接 planner 和 executor
14
workflow.add_edge("planner", "executor")
15

16
# 添加条件边，实现动态路由
17
workflow.add_conditional_edges(
18
    # 起始节点
19
    "executor",
20
    # 判断函数
21
    should_continue,
22
    # 路由映射：将判断函数的返回值映射到目标节点
23
    {
24
        "continue_to_planner": "planner", # 如果返回"continue_to_planner"，则跳回planner节点
25
        "end_workflow": END               # 如果返回"end_workflow"，则结束流程
26
    }
27
)
28

29
# 编译图，生成可执行的应用
30
app = workflow.compile()
31

32
# 运行图
33
inputs = {"current_task": "分析最近的AI行业新闻", "messages": []}
34
for event in app.stream(inputs):
35
    print(event)