AgentScope 2.0 是目前把大模型应用从单轮问答推进到多智能体协作时值得优先尝试的开源框架之一。它解决的是一个很现实的问题复杂任务不靠一个提示词模板就能完成而是需要多个角色分工、调用外部工具、边生成边输出并且在关键动作执行前保留人工审核入口。这篇文章基于 AgentScope 2.0 从零搭建一个多智能体最小示例覆盖流式输出、自定义工具和人工介入三个核心能力。读完你可以在自己的项目里复刻一套可运行的协作流程并且知道出问题时应该从哪一个环节开始排查。下面先解释 AgentScope 2.0 是怎么协调多个智能体工作的因为不理解消息传递模型后面写出的代码很容易变成“多个独立脚本拼接”。1. 先理解 AgentScope 2.0 的多智能体运行机制1.1 多智能体系统解决的问题是什么单 Agent 应用里所有逻辑都压在一条提示词里既要做理解、又要做规划、还要调用工具、最后生成答案。任务简单时没有问题一旦涉及数据查询、代码执行、内容评审等多步骤就会出现三类问题提示词越来越长模型在长上下文里迷失重点回答漂移。工具调用和回复生成混在一起很难定位是哪个环节返回了错误。关键操作没有闸门模型一旦决定执行写库或外发消息系统只能照做。多智能体系统的思路是把一个大任务拆给多个角色。每个 Agent 只负责一个相对窄的职责比如规划、检索、执行、审核彼此通过消息传递结果。这样单个系统的复杂度下降了问题边界清楚了也方便在任意两个角色之间插入人工审核节点。需要说明的是多智能体不是“多个提示词模板的拼接”。拼接只是把几段话先后丢给同一个模型角色之间没有真正的消息上下文也没有工具调用的调度能力。多智能体的核心在于通信、编排和状态管理这正是 AgentScope 这类框架存在的意义。1.2 三个核心抽象Agent、Message、PipelineAgentScope 2.0 里最常用的是三个抽象概念。Agent 是一个独立执行单元。它接收消息调用模型推理决定是直接回复还是调用工具然后把结果封装成新消息发出去。你不需要自己维护每个角色的历史记录框架会在内部管理会话上下文。Message 是 Agent 之间的通信载体。它不像普通方法参数那样只传一个字符串而是携带发送者、接收者、正文内容、工具调用标识、元数据等字段。多智能体协作的本质就是消息在生产者和消费者之间流转。Pipeline 负责编排执行顺序。它可以定义多个 Agent 是依次执行、并行执行还是走到某个条件时分支出不同路径。固定流程用 Pipeline 声明式配置会更清晰需要精细控制时也可以手动把消息逐个传给 Agent。抽象职责典型场景Agent接收消息、调用模型、调用工具、生成回复规划者、执行者、审核者Message封装通信内容和链路元数据Agent 间传递任务、结果、审批状态Pipeline编排执行顺序与分支先规划后执行、并行检索、条件分流对初学者来说最需要记住的是消息不只是“一句话”它是一条带上下文的记录。因此在设计 Agent 时要关注它接收什么消息、产出什么消息而不是只关注提示词怎么写。1.3 流式输出、工具调用与人工介入在链路中的位置流式输出、自定义工具、人工介入这三个能力分布在多智能体链路的不同层次。流式输出位于用户交互层。模型推理时token 逐批生成系统把已经生成的片段立刻返回给前端用户可以边等边看不必看到白屏等待十几秒。Web 场景下通常用 SSE 协议把流式片段推给浏览器。自定义工具位于能力扩展层。模型本身不能访问外部系统它只能输出一个“想调用某个函数”的结构化结果。框架负责把结果翻译成本地函数调用再把返回值作为额外消息送回给模型。这个机制让 Agent 能查数据库、调接口、执行计算。人工介入位于安全治理层。Agent 在行动前尤其是执行写操作、支付、外发通知这类高风险动作时需要暂停下来等一个人批准。人工介入不一定要用户全程参与它可以只出现在关键节点审批通过后流程继续自动运行。理解这三层的分工后下面的实操会顺理成章先搭消息链路再开流式输出接着挂工具最后加审批闸门。2. 环境准备与最小项目结构2.1 环境要求与版本确认AgentScope 2.0 是 Python 生态的框架建议使用 Python 3.10 及以上版本。安装前先确认两个信息一是本机 Python 版本二是准备接入的大模型服务是否提供兼容接口。模型服务要支持流式返回因为流式输出依赖服务端逐 token 下发。一个常见问题是环境里已经装了旧版本依赖。AgentScope 依赖 openai、pydantic 等库如果这些库版本过旧或过新可能在运行时出现类型校验或协议不兼容的报错。推荐用虚拟环境隔离。环境项要求说明Python3.10 及以上过低版本可能不支持部分语法和依赖pip / uv任意一种用于安装 agentscope 及其依赖模型服务兼容流式接口需要支持增量返回 tokenAPI Key有效且已授权通过环境变量注入避免写死如果原始项目没有指定框架版本落地前先去 PyPI 或官方仓库确认当前稳定版本号再执行安装。不同小版本的 API 名称可能有差异后面代码里的导入路径以安装版本的文档为准。2.2 安装 AgentScope 2.0 并配置模型安装命令如下python -m venv .venv source .venv/bin/activate pip install -U agentscope如果你使用 uv可以更快完成依赖解析uv venv .venv source .venv/bin/activate uv pip install -U agentscope安装完成后检查版本python -c import agentscope; print(agentscope.__version__)模型配置建议放到独立配置文件里。AgentScope 支持从 JSON 读取模型配置这样切换模型或更新 API Key 时不需要改业务代码。下面是一个示例实际使用时要替换成你自己的 endpoint 和 Key[ { config_name: primary_model, model_type: openai_chat, model_name: gpt-4o, api_key: ${env:MODEL_API_KEY}, base_url: https://your-endpoint.example.com/v1, generate_args: { temperature: 0.7 } } ]在代码入口初始化模型管理器from agentscope.manager import ModelManager ModelManager.get().init_model( model_configs[ { config_name: primary_model, model_type: openai_chat, model_name: gpt-4o, api_key: sk-xxx, } ] )这里要注意环境变量和明文 Key 的选择。本地学习可以直接填入占位 Key生产环境必须通过环境变量或密钥管理服务注入不要把 Key 提交到代码仓库。2.3 推荐的项目目录结构多智能体项目规模不大时把所有代码写进一个 main.py 也能跑通。但一旦工具数量超过三个建议按角色和工具拆目录方便维护和测试。下面是一个适合中期项目的结构agent_workshop/ ├── main.py ├── requirements.txt ├── configs/ │ └── model_configs.json ├── agents/ │ ├── __init__.py │ ├── planner.py │ └── executor.py └── tools/ ├── __init__.py ├── stock_tool.py ├── search_tool.py └── approval_tool.py把工具独立成模块的价值在于工具函数不依赖 Agent 和模型方便单独写单元测试同时多个 Agent 可以复用同一份工具不需要复制粘贴。后面的示例会沿用这个结构。3. 搭建带流式输出的双 Agent 协作流程3.1 用 ReActAgent 创建两个角色ReActAgent 是 AgentScope 里兼顾推理和行动的基础 Agent 类型。它会按照 Thought思考- Action行动- Observation观察的循环工作适合需要工具调用的场景。先创建两个角色规划者负责拆解用户需求执行者负责具体落地。from agentscope.agent import ReActAgent planner ReActAgent( nameplanner, model_config_nameprimary_model, system_prompt( 你是任务规划者。收到用户需求后 先拆解成不超过三个步骤 每个步骤都以清晰的任务描述发给执行者。 ), streamTrue, ) executor ReActAgent( nameexecutor, model_config_nameprimary_model, system_prompt( 你是执行者。根据规划者给出的任务步骤 依次完成并返回结果。 ), streamTrue, )这里有两个容易忽略的点。第一两个 Agent 可以共用同一个模型配置但在生产环境中最好给不同角色配置不同模型或不同 temperature规划任务可以更低温、执行任务可以稍高一些。第二system_prompt 要明确“你负责什么、收到什么、产出什么”角色边界越清楚协作越稳定。3.2 让模型按流式方式逐字返回streamTrue 是开启流式输出的关键参数。它告诉 Agent 在调用模型时使用流式接口让 token 分批返回而不是等完整文本生成完毕后再一次性返回。在终端里验证流式效果时需要关闭标准输出的缓冲def print_stream(chunk: str) - None: print(chunk, end, flushTrue)如果不调用 flush控制台可能等整段文本结束后才一次性打印看起来像没有流式效果。实际上数据已经分批到内存了只是没被刷新到终端。Web 场景下流式数据通常通过 SSE 推送。后端每次拿到新的 token 片段就格式化成一个 SSE 事件发送给前端。前端如果用 Vue3 消费可以用 fetch 配合 ReadableStream 读取或者使用 EventSource。EventSource 只支持 GET而很多流式接口需要 POST 请求所以更常见的做法是用 fetch 读取响应体中的流。const response await fetch(/api/agent/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ question: 分析本周风险 }), }); const reader response.body.getReader(); const decoder new TextDecoder(); while (true) { const { done, value } await reader.read(); if (done) break; console.log(decoder.decode(value)); }这段代码演示了前端如何消费流式接口真正接入线上接口时URL、请求参数和事件格式都要以后端协议为准。3.3 两种消息传递方式Pipeline 与手动转发第一种方式是使用 Pipeline 编排固定流程from agentscope.pipeline import Pipeline pipeline Pipeline( agents[planner, executor] ) response pipeline.run(帮我分析本周的股票持仓风险) print(response)Pipeline 的好处是声明式、可读性强适合流程固定不变的多步任务。缺点是不够灵活比如“根据规划结果决定走哪个执行分支”时写出条件逻辑会比较别扭。第二种方式是手动传递消息。先构造一条用户消息然后把规划者的输出继续传给执行者from agentscope.message import Msg request Msg( nameuser, content帮我分析本周的股票持仓风险, ) planner_reply planner(request, streamTrue) executor_reply executor(planner_reply, streamTrue) print(executor_reply.content)手动方式的好处是每一步都可见方便在中间插入人工审核、改消息、加日志。对初学者我推荐先用手动方式跑通链路理解消息是怎么从上一个 Agent 流到下一个 Agent 的再切换到 Pipeline。3.4 运行与验证把完整脚本保存为 main.py 后运行python main.py正常现象是控制台先出现规划者的思考过程和任务拆解然后出现执行者的处理过程文字按块状逐段打印最终输出分析结论。检查点有三个规划者和执行者的输出都出现了说明消息传递链路正常。输出是分批出现的而不是等待十几秒后一次性全部打印说明流式开关生效。system_prompt 没有互相串场规划者没有跑去执行执行者没有重新规划。如果终端一次性输出优先检查是否调用了 flush以及模型服务是否真的支持流式返回。4. 给 Agent 接入自定义工具4.1 工具调用为什么不是“函数直接执行”很多第一次接触 Agent 的人会问既然要查股票价格为什么不在代码里直接get_stock_price(600519)然后拼进提示词原因是模型需要自行决定“什么时候调用哪个工具、传什么参数”。你在提示词里写死一个查询结果模型只是被动使用这段文字它不知道在另一个场景下还可以调用另一个工具。工具调用的正确流程是模型根据用户问题判定需要调用函数。模型输出结构化的函数名和参数。框架解析这个消息在本地执行对应函数。函数返回值作为新的消息送回给模型。模型基于工具返回结果生成最终答案。所以工具不只是“函数”它是模型能力边界的扩展。模型通过函数描述了解工具的作用和使用方式这个描述越规范调用越准确。4.2 用 function_to_tool 注册一个查询工具AgentScope 提供了装饰器可以把普通函数转换成 Agent 可识别的工具。函数签名、类型注解和 docstring 会被自动解析成模型需要看到的工具说明。下面模拟一个股票查询工具from agentscope.tools import function_to_tool function_to_tool def get_stock_price(code: str) - str: 获取指定股票代码的当前价格。 Args: code: 股票代码例如 600519。 # 模拟行情数据生产环境替换为真实行情接口 price_map { 600519: 1680.0, 000001: 11.5, 601318: 45.2, } price price_map.get(code) if price is None: return f未找到股票 {code} 的价格 return f股票 {code} 当前价格为 {price} 元docstring 里的 Args 说明非常关键。模型不会真正查看函数内部代码它只能通过这份描述判断参数含义。描述含糊时模型可能传错参数例如把“600519”写成“贵州茅台”。如果需要返回结构化数据给模型做进一步计算可以让函数返回 JSON 字符串或字典。模型对 JSON 的解析稳定性较高比自由文本更适合后续处理。4.3 把工具挂载到 Agent 并观察调用链路工具注册好后在创建 Agent 时通过 tools 参数挂载from agentscope.agent import ReActAgent planner ReActAgent( nameplanner, model_config_nameprimary_model, system_prompt你是任务规划者负责拆解用户需求。, tools[get_stock_price], streamTrue, )运行后控制台通常会出现类似下面的调用链日志片段Thought: 用户需要查询这只股票的价格我调用股票查询工具。 Action: get_stock_price({code: 600519}) Observation: 股票 600519 当前价格为 1680.0 元看到 Action 和 Observation 成对出现说明工具调用链路是通的。如果只出现 Thought 没有 Action通常是工具描述不清晰模型没有判断出该用工具。如果 Action 后没有 Observation可能是函数内部抛了异常需要检查返回值是否合法。4.4 多工具的参数设计要点当 Agent 同时挂载多个工具时参数设计的差别会直接影响调用成功率。场景工具函数参数设计返回值查询行情get_stock_price(code)单一必填参数枚举值写清楚格式化字符串搜索资料web_search(keyword, limit)keyword 必填limit 可选带默认值JSON 列表写入记录create_order(order_id, amount)参数要带业务约束说明操作结果状态多工具场景下要注意三点第一参数名要语义化。model 能理解user_id是用户编号但a、b、x这类名字会明显降低调用准确率。第二可选参数要给出明确默认值。limit: int 5比limit: int更容易被模型正确处理。第三工具之间职责要分离。一个工具只做一件事不要写一个万能函数既能查价格又能下单否则模型会困惑该传什么参数。5. 人工介入给 Agent 加一道审批闸门5.1 人工介入在企业场景中的必要性多智能体自动跑了几个流程后发现一个规律模型在低风险任务上表现不错但遇到写库、支付、外发消息、删除数据这类动作时完全放权非常危险。人工介入也叫 Human-in-the-Loop是指在自动化链路中保留人工审核节点。它不是让用户全程参与而是只在关键节点暂停。节点通过后继续自动执行节点被拒后进入修正或终止分支。企业级系统里人工介入通常服务于三类诉求合规审计要求关键操作有责任人业务经验能纠正模型在该场景下的误判风险控制要求高风险操作不能由模型单独决定。5.2 方案一审批工具阻塞等待最简单的实现是把“人工审批”也封装成工具。Agent 决定执行关键动作前先调用审批工具函数内部阻塞等待操作员输入from agentscope.tools import function_to_tool function_to_tool def request_human_approval(action_desc: str) - str: 在执行关键动作前请求人工审批。 Args: action_desc: 待审批动作的完整描述包含对象和后果。 answer input( f[人工审批] 是否允许执行下列操作\n{action_desc}\n f请输入 yes 或 no ) answer answer.strip().lower() if answer in (yes, y): return APPROVED return REJECTED把该工具挂到执行者 Agent 上executor ReActAgent( nameexecutor, model_config_nameprimary_model, system_prompt你是执行者执行关键操作前必须申请人工审批。, tools[request_human_approval], streamTrue, )这种方式代码量最小适合命令行原型和内部工具。缺点是 input 会阻塞整个进程在 Web 服务或多线程环境中会占住线程不适合直接上线。5.3 方案二UserAgent 作为审核节点更贴合真实系统的方式是把人工审核设计成一个独立节点。AgentScope 提供 UserAgent 类型专门用于和真实用户交互。from agentscope.agent import UserAgent reviewer UserAgent(namereviewer) # 流程中先执行再送审 executor_output executor(planner_reply, streamTrue) review_reply reviewer(executor_output) if 拒绝 in review_reply.content or no in review_reply.content.lower(): # 进入人工修正分支 print(人工已拒绝流程终止) else: # 审批通过继续后续动作 final_result executor( Msg(planner, 审批已通过请完成最终结果整理, review_reply) )UserAgent 的优点是可以复用同一套消息机制审核结果仍然是一条 Message能被后续节点读取和记录。Web 化改造时可以把 UserAgent 的输入来源从命令行替换成前端页面上的“同意 / 拒绝”按钮。5.4 审批状态与超时设计无论选择哪种方案都要考虑审批状态管理。最简单的方式是在 Message 的 metadata 里带一个状态字段{ status: pending, operator: admin, review_time: 2025-06-01 10:30:00, review_note: 同意执行 }流程中需要处理的审批分支通常有三种结果通过继续执行后续 Agent。拒绝停止当前动作并告诉模型“用户拒绝了请生成替代方案或终止”。超时长时间无人处理时建议按拒绝处理并记录超时原因。超时场景很容易被忽略。如果审批节点无人点击流程会一直挂着占用资源。生产系统应该给每个审批节点配置超时时间超时后主动回退或转人工。import time deadline time.time() 600 # 10 分钟超时 while time.time() deadline: reply reviewer(...) # 轮询或回调 if reply is not None: break time.sleep(5)这种轮询只是演示思路。真实系统更推荐异步回调或消息队列的方案审批动作触发后系统把审批任务写入队列前端展示待办用户点击后通过接口回调流程继续往下走。6. 常见问题与排查路径6.1 流式输出为空或只输出一次现象设置了 streamTrue但控制台没有任何增量输出或只输出一次完整结果。可能原因和检查顺序模型服务不支持流式接口。检查模型配置里是否关闭了流式参数或者服务商的接口协议是否兼容 OpenAI 的流式格式。终端缓冲未刷新。确认生成回调里调用 flush。网络层做了缓冲。如果经过 Nginx 或其他网关要把代理缓冲关闭否则 SSE 数据会被攒到连接结束才返回。前端用了 EventSource 但接口是 POST。改为 fetch 流式读取。6.2 工具参数格式不匹配现象Action 里出现了明显错误的参数例如把股票代码写成“茅台”或者把数字参数传成字符串。原因通常是工具描述不够清晰。模型只能根据函数签名和 docstring 猜测参数含义。解决方式在 docstring 里写清楚参数格式、单位、可接受示例值。参数名用业务可读的名称不要用简称。如果模型反复传错给参数加一个格式说明比如code: 6 位数字股票代码。另外要检查工具函数的异常处理。函数内部如果抛出未捕获异常Agent 很可能把报错信息当成 Observation 返回导致后续推理混乱。建议工具函数统一返回状态结构而不是抛异常。6.3 人工介入流程卡死现象流程运行到审批节点后一直等待没有任何日志也没有超时处理。人工介入卡死最常见的原因是阻塞式 input 跑在了异步或 Web 环境里。input 会一直等待标准输入在 Web 服务进程里根本没有终端输入源于是永久阻塞。排查路径确认审批节点的输入来源是命令行、页面还是队列回调。检查是否配置超时。检查审批通过后流程是否能正确继续拒绝分支是否被写死。生产环境不要使用 input 方式优先使用异步审批接口加轮询或回调。6.4 Agent 反复调用工具进入循环现象日志里出现大量相同或相似的 Action/ObservationAgent 一直没有生成最终答案。常见原因有三个工具返回值没有真正回答模型的问题模型认为信息不足就一直重试。system_prompt 没有限制最大尝试次数。工具返回的数据格式模型无法理解例如返回了未经解析的二进制或过长文本。建议在创建 Agent 时设置最大迭代次数并在工具返回结果里包含足够的决策信息。框架层面的循环控制是最后一道防线不要只依赖模型自觉。6.5 排查顺序建议多智能体问题排查不要直接从日志尾部开始建议按下面顺序走输入是否正确发给第一个 Agent 的消息是否符合预期。消息链路是否连通每个 Agent 是否收到、是否产出消息。模型调用是否成功是否出现限流、超时、上下文超长。工具调度是否正确工具是否注册成功参数是否合法。审批与分支是否正常人工节点是否被触发结果如何流转。问题现象常见原因检查方式处理建议流式输出不生效模型或网关不支持流式查看网关配置和模型接口文档关闭网关缓冲确认流式参数工具参数错误工具描述不清晰查看 Action 中实际传参完善 docstring给示例值审批卡死使用了阻塞 input检查运行环境是否有终端输入改异步审批接口Agent 死循环工具结果模型无法理解查看 Observation 内容返回结构化结果并限制迭代次数模型报上下文超长历史消息累积过多查看请求体大小对历史消息做裁剪或摘要7. 生产环境落地从 Demo 到真正的企业级7.1 学习环境与生产环境的差异Demo 能跑通和生产环境可用是两回事。差异主要体现在这几个方面维度学习环境生产环境配置代码内硬编码配置中心、环境变量、密钥管理日志print 输出结构化日志、链路追踪人工介入input 阻塞审批流程、消息队列、超时回退工具调用本地模拟数据真实接口、幂等、限流、鉴权可靠性失败直接退出重试、降级、熔断、回滚安全无鉴权用户鉴权、操作审计、数据脱敏生产环境里模型返回的文本只是系统的一部分更重要的是把它嵌入到已有工程体系里让每一次调用都可追溯、可回滚、可审计。7.2 日志、追踪与审计多智能体链路横跨多个 Agent 和多次模型调用排查问题时最怕的是“只看到最终回复看不到中间过程”。建议从第一天就约定日志规范每个会话生成唯一 session_id所有 Agent 日志都带上这个 ID。记录每条 Message 的 from、to、内容摘要、时间戳。记录每次模型调用的输入、输出、耗时、token 消耗。记录每个工具的调用参数和返回结果参数中包含敏感信息时要脱敏。人工审批记录要单独留存包含操作人、审批时间、审批结果、备注。有了这些记录线上出问题时才能回答三个问题这个结果是谁生成的、它调用过什么工具、谁批准了这次操作。7.3 发布前检查清单把一个多智能体服务发布到生产环境前至少过一遍下面的清单模型配置是否外置API Key 是否通过环境变量注入。工具调用是否做了超时限制和异常兜底。高风险操作是否都有人工审批节点。审批超时策略是否配置拒绝分支是否可正常退出。是否有 session_id 贯穿全部日志。是否记录 token 消耗和接口调用量方便成本控制。是否设置最大迭代次数防止 Agent 死循环烧 token。历史消息是否做了长度控制。是否验证过模型服务不可用时的降级方案。是否对用户输入做了脱敏和内容安全过滤。其中第 7 条最容易被忽略。Agent 一旦陷入工具调用循环消耗的是真实 token 和接口调用费必须在框架层设上限。7.4 扩展方向当前这套双 Agent 加工具加审批的骨架可以往几个方向扩展。第一个方向是接入 MCP。MCP 提供了标准化工具接入协议让 Agent 可以按统一方式发现和调用外部工具服务。如果团队维护了大量内部工具与其一个个写装饰器不如通过 MCP 把工具服务统一暴露出来。第二个方向是分布式化。单机多 Agent 的瓶颈在于并发和资源隔离。用户量上来后可以把不同 Agent 部署成独立服务通过消息队列或 RPC 通信每个 Agent 独立扩容。第三个方向是记忆和知识库增强。给 Agent 增加持久化记忆让它能在多轮任务里记住用户偏好和历史决策知识库检索则能减少模型在专业领域里的凭空生成。第四个方向是前端的完整流式体验。当前后端流式接口已经跑通前端可以用 Vue3 按块渲染文本并结合用户反馈按钮把人工审批从命令行搬进页面。如果你想继续深入最值得练的题目是把当前示例里的命令行审批改成接口审批再给整个流程加上 session 维度的日志追踪。这两个改造做完这个 Demo 就具备接入真实业务系统的基础了。对新手来说还有一条练习路径值得走先把手动消息传递改成 Pipeline再用自定义工具替代模拟数据最后把审批工具替换成异步队列。每换一步都会遇到一类新问题解决完这些问题的过程就是真正理解多智能体编排的过程。