Ohhnews

分类导航

$ cd ..
foojay原文

BoxLang AI 深度解析：多智能体编排与协作系统构建

#boxlang#人工智能#多智能体系统#流程编排#软件架构

BoxLang AI 深度解析 — 第 3 部分（共 7 部分）：多智能体编排 — 构建高效 AI 团队 🌲

目录

[LOADING...]

BoxLang AI 3.0 系列 · 第 3 部分（共 7 部分）

单个智能体很有用，但一组智能体协同工作则更加强大。

大多数多智能体框架的问题在于编排层是“外挂”式的——你需要手动管理智能体引用，手动在它们之间传递输出，并时刻担心是否引入了循环依赖。它们缺乏层级概念，没有循环检测，也无法回答“谁是负责人”或“我在树中的深度是多少”这类问题。

BoxLang AI 3.0 改变了这一点。AiAgent 现在可以跟踪其在完整层级树中的位置，子智能体会被自动挂载为工具——协调者无需知道如何进行委派，只需知道它有权委派即可。

🌲 智能体树

每个 AiAgent 都拥有一个 parentAgent 属性和一套完整的层级辅助方法。这种关系是双向的：addSubAgent() 在一次调用中即可将子智能体注册为可调用的工具，并设置父级引用。

coordinator = aiAgent( name: "coordinator" )
    .addSubAgent( aiAgent( name: "researcher" ) )
    .addSubAgent( aiAgent( name: "writer" ) )

// 层级查询
println( coordinator.isRootAgent() )           // true
println( coordinator.getAgentDepth() )         // 0

println( researcherAgent.isRootAgent() )       // false
println( researcherAgent.getAgentDepth() )     // 1
println( researcherAgent.getAgentPath() )      // /coordinator/researcher
println( researcherAgent.getAncestors() )      // [ coordinator ]

println( writerAgent.getRootAgent().getAgentName() ) // coordinator

源码中完整的层级 API 如下：

// 来自 AiAgent.bx
setParentAgent( parent )   // 分配父级 — 包含自我引用和循环防护
clearParentAgent()         // 从父级脱离
hasParentAgent()           // 布尔值
isRootAgent()              // 当深度 == 0 时返回 true
getRootAgent()             // 向上遍历，返回根节点
getAgentDepth()            // 0 = 根节点, 1 = 子节点, 2 = 孙节点, ...
getAgentPath()             // "/coordinator/researcher"
getAncestors()             // [直接父级, 祖父级, ..., 根节点]

内置循环检测

设置会导致循环的父级会立即抛出异常——绝不会出现静默的无限循环：

// 来自 AiAgent.bx — setParentAgent()
AiAgent function setParentAgent( required AiAgent parent ) {
    if ( arguments.parent == this ) {
        throw( type: "AiAgent.InvalidParent", message: "智能体不能作为自己的父级。" )
    }
    // 向上遍历拟定父级的祖先链并检查自身
    var ancestor = arguments.parent
    while ( ancestor.hasParentAgent() ) {
        ancestor = ancestor.getParentAgent()
        if ( ancestor == this ) {
            throw( type: "AiAgent.CyclicParent", message: "设置此父级将导致循环..." )
        }
    }
    variables.parentAgent = arguments.parent
    return this
}

🤖 将子智能体作为工具

addSubAgent() 的魔力在于，每个子智能体都会自动封装为一个父级可以调用的工具——无需手动配置，无需自定义回调代码。

// 来自 AiAgent.bx — createSubAgentTool()
private ITool function createSubAgentTool( required AiAgent subAgent ) {
    var agentName    = arguments.subAgent.getConfig().name
    var toolName     = "delegate_to_" & agentName.slugify()
    var toolDescription = "将任务委派给 '#agentName#' 子智能体。 "
        & "当任务符合子智能体的专业领域时使用此工具: #agentConfig.description#"

    return aiTool(
        name: toolName,
        description: toolDescription,
        callable: ( required task ) => {
            return subAgent.run( task )
        }
    ).describeArg( "task", "委派给子智能体的任务或问题" )
}

当调用 addSubAgent() 时，父级的 AiModel 会获得一个名为 delegate_to_researcher、delegate_to_writer 等的新工具。大模型（LLM）在其上下文中看到这些工具后，会决定何时调用它们——其方式与调用任何其他工具完全相同。

协调者不需要特殊的逻辑，它只是拥有了更多的工具。

🏢 AiAgent 现已完全无状态

这是 3.0 版本中最重要的架构变更之一：AiAgent 不再将 userId 或 conversationId 作为实例状态保存，而是按调用进行解析。

// 来自 AiAgent.bx — run()
public any function run( any input = "", struct params = {}, struct options = {} ) {
    // 按调用解析 — 无共享状态
    var threadId       = arguments.options.threadId ?: variables.options.threadId ?: createUUID()
    var userId         = arguments.options.userId ?: variables.options.userId ?: ""
    var conversationId = arguments.options.conversationId ?: variables.options.conversationId ?: ""
    // ...
}

这意味着一个智能体实例可以安全地为多个并发用户提供服务——没有竞态条件，没有跨用户污染，也不需要为每个用户创建智能体工厂。

// 一个共享智能体 — 多个并发用户
sharedAgent = aiAgent( name: "support", memory: aiMemory( "cache" ) )

// 每次调用都是完全隔离的
sharedAgent.run( "你好",           {}, { userId: "alice", conversationId: "sess-1" } )
sharedAgent.run( "我刚才说了什么？", {}, { userId: "alice", conversationId: "sess-1" } ) // 记住 alice
sharedAgent.run( "你好",           {}, { userId: "bob",   conversationId: "sess-2" } ) // 与 alice 隔离

🧠 基于内存的每调用身份路由

所有内存类型都遵循相同的模式——add()、getAll()、clear() 和 trim() 都接受可选的 userId 和 conversationId 参数：

// 一个内存实例，多个租户
sharedMemory = aiMemory( "cache" )

sharedMemory.add( message, userId: "alice", conversationId: "conv-1" )
sharedMemory.add( message, userId: "bob",   conversationId: "conv-2" )

// 每次检索都是租户隔离的
aliceHistory = sharedMemory.getAll( userId: "alice", conversationId: "conv-1" )
bobHistory   = sharedMemory.getAll( userId: "bob",   conversationId: "conv-2" )

当智能体在内部调用 loadMemoryMessages() 时，它会将解析后的 userId 和 conversationId 传递给所有关联的内存。内存自然实现了租户隔离，无需任何额外的配置。

🏗️ 智能体运行生命周期

了解 run() 内部发生了什么，对于调试或构建中间件（第 4 部分将深入讨论）非常有用。以下是执行顺序：

  1. 解析 threadId / userId / conversationId（按调用解析，非实例状态）。
  2. 构建用户消息结构。
  3. 构建系统消息（描述 + 指令 + 技能 + 工具 + MCP 服务器）。
  4. 加载该 userId/conversationId 的内存消息。
  5. 组装：[system, ...memory, userMessage]。
  6. 触发 beforeAgentRun 中间件（前向传递）。
  7. 触发 BoxAnnounce 事件 "beforeAIAgentRun"（全局事件）。
  8. 通过 AiModel.run() 执行 — 处理工具调用、重试、流式传输。
  9. 如果挂起（HITL）：检查点存储并返回。
  10. 将助理响应存储在所有内存中（以 userId/conversationId 为作用域）。
  11. 触发 afterAgentRun 中间件（反向传递）。
  12. 触发 BoxAnnounce 事件 "afterAIAgentRun"（全局事件）。
  13. 返回响应。

系统消息也会被缓存和指纹化——如果描述、指令和技能池自上次调用以来没有改变，则使用缓存版本，而不是重新构建。这对于同一智能体处理大量请求的高吞吐量场景至关重要。

🌊 多智能体团队的流式传输

流式传输在多智能体设置中工作方式相同——每个智能体都可以独立进行流式传输：

coordinator = aiAgent( name: "coordinator" )
    .addSubAgent( aiAgent( name: "researcher" ) )
    .addSubAgent( aiAgent( name: "writer" ) )

// 流式处理协调者的输出
coordinator.stream(
    onChunk : chunk => writeOutput( chunk.choices?.first()?.delta?.content ?: "" ),
    input   : "研究并撰写一篇关于 BoxLang JVM 互操作性的 500 字文章",
    options : { userId: "user-123", conversationId: "session-abc" }
)

当协调者决定委派给研究员时，该子调用在工具调用内部同步发生——流式传输的协调者会收到研究员的结果作为工具响应，然后继续流式传输。

🔄 挂起与恢复

当 HumanInTheLoopMiddleware（第 4 部分涵盖）挂起智能体时，状态需要被保留。checkpointer 属性负责处理此过程：

agent = aiAgent(
    name        : "finance-bot",
    middleware  : new HumanInTheLoopMiddleware(
        mode                  : "web",
        toolsRequiringApproval: [ "transferFunds" ]
    ),
    checkpointer: aiMemory( "cache" )
)

// 首次调用 — 挂起等待批准
result = agent.run( "转账 $500 到账户 #12345" )
// result.isSuspended() == true
// 智能体自动保存带有 threadId 的检查点

// 稍后，在人类通过 Web UI 批准后
threadId = result.getData().threadId
agent.resume( "approve", threadId )

// 或者拒绝：
agent.resume( "reject", threadId )

// 或者修改参数：
agent.resume( "edit", threadId, { correctedArgs: { amount: 100, account: "#12345" } } )

resume() 的实现会从保存的检查点重新运行，将人类的决定注入中间件上下文：

// 来自 AiAgent.bx — resume()
any function resume( required string decision, required string threadId, struct editedData = {} ) {
    var savedState = variables.checkpointer.loadState( arguments.threadId )
    // 清除状态，防止被重复恢复
    variables.checkpointer.clearState( arguments.threadId )
    // 使用注入到 options 中的恢复上下文重新运行
    return run(
        savedState.input,
        savedState.params,
        savedState.options.append( { _resumeContext: {
            resumeDecision : arguments.decision,
            editedData     : arguments.editedData,
            suspendData    : savedState.suspendData
        } } )
    )
}

🔍 自省

getConfig() 方法让你能够全面了解智能体的状态——对于调试、监控仪表板和日志记录非常有用：

config = agent.getConfig()

// 层级
println( config.agentDepth )    // 0, 1, 2, ...
println( config.agentPath )     // "/coordinator/researcher"
println( config.parentAgent )   // "coordinator" (名称字符串)

// 能力
println( config.toolCount )     // 总工具数（包括子智能体委派工具）
println( config.tools )         // [{ name, description }]
println( config.mcpServers )    // [{ url, toolNames }]

// 内存
println( config.memoryCount )
println( config.memories )

// 技能
println( config.activeSkillCount )
println( config.availableSkillCount )
println( config.skills )        // { activeSkills: [...], availableSkills: [...] }

// 中间件
println( config.middlewareCount )
println( config.middleware )    // [{ name, description }]

🚀 完整的多智能体示例

这是一个实用的编排案例：一个协调者将研究任务委派给专业研究员，将写作任务委派给专业写作者，两者都拥有各自的技能和工具。

// 专业智能体
researchAgent = aiAgent(
    name        : "researcher",
    description : "擅长从多个来源查找、分析和总结信息的专家",
    instructions: "始终引用来源。优先考虑最新信息。清晰总结调查结果。",
    tools       : [ "searchWeb@tools", "fetchURL@tools" ],
    skills      : [ aiSkill( ".ai/skills/research-methodology/SKILL.md" ) ]
)

writerAgent = aiAgent(
    name        : "writer",
    description : "将研究转化为润色、引人入胜的散文的专家",
    instructions: "为技术受众撰写。简明扼要。不要废话。",
    skills      : [ aiSkill( ".ai/skills/writing-style/SKILL.md" ) ]
)

// 协调者，将以上两者作为子智能体
coordinator = aiAgent(
    name        : "content-coordinator",
    description : "编排研究和写作以产出完整文章",
    instructions: "将复杂请求分解为研究和写作阶段。相应地进行委派。",
    subAgents   : [ researchAgent, writerAgent ],
    memory      : aiMemory( "cache" )
)

// 检查层级
println( coordinator.isRootAgent() )               // true
println( researchAgent.getAgentPath() )            // /content-coordinator/researcher
println( coordinator.getConfig().toolCount )       // 2 (delegate_to_researcher + delegate_to_writer)

// 运行 — 协调者决定何时委派以及委派给谁
response = coordinator.run(
    "撰写一篇关于 BoxLang AI 功能的综合文章",
    {},
    { userId: "luis", conversationId: "article-session-1" }
)

驱动协调者的大模型看到了两个工具：delegate_to_researcher 和 delegate_to_writer。它决定首先调用研究员，获得详细摘要，然后将该摘要和原始请求发送给写作者，最后将写作者的输出合成为最终响应。你无需编写任何此类逻辑——大模型通过工具描述自行推导出了执行方案。

后续展望

在第 4 部分中，我们将探讨中间件系统——包括六个内置中间件类、钩子生命周期如何工作、如何编写自己的中间件，以及使 AI 智能体在 CI 中可测试的 FlightRecorderMiddleware。

📖 完整文档 📦 立即安装：install-bx-module bx-ai 🫶 专业支持