Agent 框架 07：AgentUniverse——金融级多智能体协同与 PEER 模式

在一般性的消费级 AI 场景中，智能体偶尔产生轻微幻觉或者忽略某个细节，用户往往可以一笑置之。但在金融、证券、风控与合规审查等企业核心领域，一次事实错误或逻辑遗漏就可能导致严重的决策失误与资产损失。单智能体面对复杂的长篇研报撰写或产业链尽职调查时，极易陷入上下文过载或推理退化。

AgentUniverse（简称 aU） 是由蚂蚁集团开源的企业级多智能体协作框架（GitHub 仓库：alipay/agentUniverse）。它脱胎于支付宝内部真实的大模型金融业务，其最核心的贡献是提出了标准化的 PEER（规划-执行-表达-审查）多智能体协作范式，并采用全套声明式 YAML 配置驱动与金融级全链路可观测性设计。

核心设计：PEER 多智能体协作模式

面对一个复杂的目标（如“深度分析某上市公司的财报异动与供应链潜在风险”），如果直接扔给单个大模型，模型往往只能给出泛泛而谈的套话。AgentUniverse 将复杂任务的推导过程拆解为四个互有分工、相互制衡的专职智能体角色：

PEER 四要素权责定义

角色 缩写全称 核心职责 在金融/复杂业务中的对应岗位
P Planning Agent 将高阶模糊任务拆解为相互独立、完全穷尽（MECE 原则）的子研究任务流 行业研究总监 / 资深投资经理
E Executing Agent 专注执行单一垂直子任务，定向调用特定数据库、知识库与工具 行业分析助理 / 数据挖掘专员
E Expressing Agent 综合所有 Executing Agent 的零散事实输入，提炼核心结论，组织严密的文章框架 资深研报主笔人
R Reviewing Agent 站在反方与合规视角对草案进行挑刺审查，验证论据充分性与数字一致性 风控审查委员会 / 合规质检员

声明式配置驱动架构

与大多数框架要求将 Agent、Prompt、Tool 和模型写死在 Python 代码中不同，AgentUniverse 借鉴了云原生 Kubernetes 的设计思想：一切能力由 YAML 文件声明描述，Python 代码仅作为通用运行时底座。

1. 声明 Agent 配置（research_agent.yaml）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# agent/research_agent.yaml
name: 'research_agent'
description: '专业负责行业宏观数据检索与整理的执行智能体'
profile:
  prompt_version: 'research_agent_cn'
  llm_model:
    name: 'qwen_max_llm'
    temperature: 0.2
    max_tokens: 3000
plan:
  planner:
    name: 'react_planner'
action:
  tool:
    - 'financial_data_tool'
    - 'web_search_tool'
  knowledge:
    - 'macro_industry_kb'
memory:
  name: 'session_memory'

2. 声明大模型后端配置（qwen_llm.yaml）

1
2
3
4
5
6
7
# llm/qwen_llm.yaml
name: 'qwen_max_llm'
description: '阿里云通义千问 Qwen-Max 接入端'
model_name: 'qwen-max'
max_tokens: 4000
temperature: 0.1
request_timeout: 60

这种配置化的好处显而易见：业务人员或提示词工程师修改 Prompt、切换基座模型或增加工具白名单时，完全不需要修改任何底层 Python 逻辑，甚至可以通过热加载即时生效。

安装与快速启动

通过 pip 安装 AgentUniverse 核心包：

1
pip install agentUniverse

编写最简启动代码，框架会自动扫描指定目录下的全部配置文件，完成依赖注入与拓扑组装：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from agentuniverse.base.agentuniverse import AgentUniverse
from agentuniverse.agent.agent_manager import AgentManager

# 1. 框架初始化：指定根配置文件，自动发现 tools / agents / llms 目录
AgentUniverse().start(config_path="./config/config.toml")

# 2. 从注册容器中按名称获取已编排好的 PEER 智能体实例
peer_agent = AgentManager().get_instance_obj("peer_agent")

# 3. 运行复杂端到端金融分析任务
query = "深度分析 2026 年生成式 AI 在银行业信贷审批中的降本增效表现及潜在合规挑战"
response = peer_agent.run(input=query)

print("=" * 60)
print("PEER 协同最终产出报告：\n", response.get_data("output"))
print("=" * 60)

自定义组件扩展

1. 自定义 Agent 行为

当标准 ReAct 逻辑无法满足企业特定流水线时，可以继承 Agent 基类重写输入输出契约：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from agentuniverse.agent.agent import Agent
from agentuniverse.agent.input_object import InputObject
from agentuniverse.agent.output_object import OutputObject

class FinancialAuditAgent(Agent):
    """自定义风控审计专用智能体"""

    def input_keys(self) -> list[str]:
        return ["input", "risk_level"]

    def output_keys(self) -> list[str]:
        return ["audit_status", "risk_items", "summary"]

    def parse_input(self, input_object: InputObject, agent_input: dict) -> dict:
        agent_input["input"] = input_object.get_data("input")
        agent_input["risk_level"] = input_object.get_data("risk_level", "medium")
        return agent_input

    def parse_result(self, agent_result: dict) -> OutputObject:
        return OutputObject({
            "audit_status": agent_result.get("audit_status", "REJECTED"),
            "risk_items": agent_result.get("risk_items", []),
            "summary": agent_result.get("summary", "审计未完成"),
        })

2. 自定义结构化金融工具

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from agentuniverse.agent.action.tool.tool import Tool

class StockMetricsTool(Tool):
    """获取企业实时财务比率指标"""
    name: str = "stock_metrics_tool"
    description: str = "输入企业股票代码，查询市盈率 PE、市净率 PB 及近一年净资产收益率 ROE"

    def execute(self, tool_input: str, **kwargs) -> str:
        # tool_input 是大模型规划传入的入参
        stock_code = tool_input.strip()
        # 实际业务中对接 Bloomberg、Wind 或内部量化接口
        mock_data = {
            "600519": "PE(TTM): 28.5, PB: 9.8, ROE: 32.4%",
            "000001": "PE(TTM): 4.6, PB: 0.52, ROE: 10.8%",
        }
        return mock_data.get(stock_code, f"未查询到标的代码 '{stock_code}' 的有效财务指标")

金融级可观测性：全链路 Tracing 与审计

在金融合规检查中，系统必须讲清楚“为什么做出这笔投资建议”，每个子任务由谁执行、花费多长时间、调用了哪些工具必须完全可审计。AgentUniverse 内置了全链路追踪支持：

1
2
3
4
5
6
7
8
9
10
11
12
13
from agentuniverse.base.tracing import Tracer

tracer = Tracer()
tracer.start_trace(agent_name="peer_financial_agent", input="任务输入")

# ... 触发智能体执行 ...

tracer.end_trace(output="执行结果")

# 导出本次决策完整拓扑链路
trace_info = tracer.get_trace()
for idx, step in enumerate(trace_info.steps, 1):
    print(f"步骤 {idx} | 智能体: {step.agent} | 动作: {step.action} | 耗时: {step.duration_ms}ms")

优缺点分析与工程选型边界

核心优势

  1. 工业级沉淀的 PEER 模式：相比松散的群聊讨论，PEER 提供了一套经过高压金融场景实测的严谨分工流，生成的文章结构性与准确度极高。
  2. 配置与代码解耦彻底：YAML 声明式语法便于企业大团队分工（业务专家管 Prompt/配置，架构师管底座插件）。
  3. 原生可观测性与审计保障：开箱即用的调用链追踪，天然契合政企风控与事后复盘要求。
  4. 国产模型生态友好：对阿里通义千问、百度文心一言、智谱 GLM 等国内大模型集成体验平滑。

现实痛点与妥协

  1. 学习与配置开销较大：项目包含大量的 TOML、YAML 配置文件和目录层级，初期上手容易被路径引用规则绕晕。
  2. 社区规模仍在成长期：主要在国内金融和企业级圈子内流行，海外社区讨论度与 GitHub 生态活跃度低于 LangGraph。
  3. 灵活性相对收敛：如果业务需要非常动态、不可预测的图分支跳转，固定配置式的驱动方式不如直接写 Python 代码来得灵活。

选型决策准则

  • 强烈推荐：金融证券、咨询研报、合规风控、政府智库等要求结论极度严谨、逻辑必须经由独立审核的深度研究型系统。
  • 不建议选择：轻量交互式闲聊机器人、个人极客的原型开发、或不需要严苛复核的简单工具调用 Agent。

关联导航