尧图网页设计 
1. 从“玩具”到“工程”为什么我们需要一个多智能体框架最近在折腾AI应用尤其是想搞点多智能体协作的东西比如让几个大模型扮演不同角色一起完成一个复杂的任务。一开始我的想法很简单不就是调几个API写几个函数让它们互相发消息吗我随手写了个脚本用OpenAI的接口让一个“产品经理”智能体生成需求再让一个“程序员”智能体写代码。看起来跑通了但问题很快就来了。当我想增加一个“测试员”智能体来审查代码时消息传递的逻辑立刻变得混乱。谁该在什么时候收到什么消息消息内容是什么格式一个智能体的输出如何结构化地成为另一个智能体的输入更头疼的是我想记录下整个对话过程用于复盘或者当某个智能体“胡说八道”时能快速定位和干预。我的脚本迅速膨胀成一团乱麻调试起来像在走迷宫。我相信很多开发者都有过类似的经历。我们用一个for循环和几个if-else就能快速验证一个多智能体协作的想法这很棒。但一旦你想把这个“玩具”变成一个健壮的、可维护的、可扩展的“工程化”应用时就会遇到一系列基础设施层面的挑战。这正是像AgentScope这样的框架要解决的问题。它不是一个束缚你创意的笼子而是一个为你处理所有脏活累活的基础设施团队让你能专注于智能体本身的行为逻辑和业务价值。简单说AgentScope的核心价值在于它提供了一套标准化的“脚手架”和“工具箱”专门用于构建和管理多智能体应用。它把智能体间的通信、状态管理、流程编排、可观测性这些繁琐但必需的工程问题封装成了清晰、易用的接口。这就好比你想盖房子框架为你准备好了标准化的钢筋、水泥和施工流程你不用再从烧制砖头开始。2. AgentScope的核心架构消息总线与智能体“公寓”要理解AgentScope不能只看它提供的几个Agent类得先看清它设计的“地基”。这个地基的核心我认为是它的消息传递模型和智能体生命周期管理。我们可以把它想象成一栋专门出租给“智能体”的公寓楼。2.1 消息总线公寓楼里的内部电话系统在这栋公寓楼里智能体们房客不能直接砸对方的门。所有通信必须通过一个中央的“内部电话系统”——这就是消息总线Message Bus。在AgentScope中这个角色主要由Message类和底层的分发机制承担。一个Message不是一个简单的字符串。它是一个结构化的数据对象通常包含name: 发送者标识。content: 消息内容可以是文本、字典、甚至任何Python对象。url: 可选关联的文件或资源链接。为什么需要结构化举个例子智能体A发送了一个包含代码和解释的消息。如果只是字符串智能体B需要费力地去解析哪里是代码哪里是注释。而通过结构化的content比如{code: “print(‘hello’)”, “explanation”: “这是一个示例”}接收者就能直接获取语义明确的字段。更重要的是消息总线管理着消息的路由。它确保消息能准确送达目标智能体并且可以方便地实现广播发给所有智能体、组播发给特定群组等模式。框架还处理了消息的序列化、持久化如果你想保存对话历史和潜在的消息队列问题这些都是你在裸写脚本时需要自己反复造轮子的部分。2.2 智能体Agent标准化的“房客”与自定义装修AgentScope提供了一些基础“房客”模板也就是内置的Agent类比如UserAgent: 代表人类用户通常用于模拟用户输入或作为交互接口。DialogAgent: 一个基于大语言模型的对话智能体它封装了与LLM如GPT、文心一言等的交互包括构造Prompt、调用API、解析响应。ServiceAgent: 可以调用外部工具或服务的智能体比如执行代码、查询数据库、调用API。这些内置智能体帮你解决了最通用的模式。但真正的力量在于“自定义装修”。你可以通过继承这些基础类来创建具有特定技能、记忆或决策逻辑的智能体。例如创建一个“代码评审专家”智能体from agentscope.agents import DialogAgent from agentscope.message import Msg class CodeReviewAgent(DialogAgent): def __init__(self, name, model_config_name): super().__init__(namename, model_config_namemodel_config_name) # 可以在这里初始化该智能体的专属知识库或规则 self.review_rules [检查语法错误, 评估代码效率, 确保符合PEP8规范] def reply(self, incoming_message: Msg) - Msg: # 在调用父类即LLM的回复逻辑前可以预处理输入 enhanced_prompt f 你是一个资深代码评审专家。请根据以下规则进行评审 {self.review_rules} 需要评审的代码 {incoming_message.content} 请给出详细的评审意见包括问题、风险和改进建议。 # 修改传入消息的内容再交给父类的LLM处理 modified_msg Msg(self.name, enhanced_prompt, roleuser) return super().reply(modified_msg)通过继承和重写reply方法你就能轻松地给智能体注入特定的行为模式而无需关心消息是如何收发的、LLM的token如何管理、上下文窗口如何维护——这些都由框架层处理好了。2.3 环境与流程编排公寓的公共空间与活动流程智能体们住在各自的房间里通过消息总线打电话。那它们怎么协同完成一个任务呢比如“一起写一份项目计划书”。这就需要流程编排Orchestration。AgentScope提供了Pipeline、Workflow等抽象来定义智能体间的协作流程。你可以把它看作是公寓的“公共活动流程表”。一个最简单的顺序管道示例from agentscope.pipelines import SequentialPipeline # 假设我们已经创建了三个智能体 product_manager DialogAgent(namePM, ...) programmer DialogAgent(nameDev, ...) tester DialogAgent(nameQA, ...) # 创建一个顺序执行的管道 pipeline SequentialPipeline( agents[product_manager, programmer, tester], pipeline_name软件研发流水线 ) # 运行管道PM生成需求 - Dev根据需求写代码 - QA测试代码 initial_message Msg(user, 我们需要一个计算斐波那契数列的函数。) final_result pipeline.run(initial_message)在这个管道里框架会自动将上一个智能体的输出消息作为下一个智能体的输入消息。你还可以构建更复杂的流程比如包含条件分支IfElsePipeline、循环ForLoopPipeline或者并行执行ParallelPipeline。这种声明式的编排方式让复杂的多智能体交互逻辑变得清晰、可视且易于维护。3. 从零搭建一个实战项目多智能体头脑风暴会议理论讲得再多不如亲手搭一个。我们来实战一个经典场景多智能体头脑风暴会议。目标是创建一个会议系统包含一个主持人Moderator、若干个领域专家如技术专家、市场专家、设计专家让它们针对一个议题进行多轮讨论最终由主持人汇总成一份创新方案。3.1 第一步环境搭建与模型配置首先安装AgentScope。推荐使用pip并指定版本以获得稳定体验。pip install agentscope接下来是模型配置。AgentScope支持多种模型后端OpenAI、Azure、DashScope、Ollama等。我们需要在项目根目录创建一个config.yaml或configs.py来集中管理配置。这是非常重要的一步它能将模型API密钥、基地址等敏感或易变信息从代码中分离出来。# config.yaml model_configs: # 定义一个名为“gpt-4”的配置 gpt-4: model_type: “openai” # 指定模型类型 config_name: “gpt-4” # 配置名称在代码中引用 model_name: “gpt-4-turbo-preview” # 具体的模型名称 api_key: “your-openai-api-key-here” # 你的API密钥 # 可选参数 max_tokens: 2000 temperature: 0.7 # 可以定义多个配置例如备用模型或不同能力的模型 qwen-max: model_type: “dashscope” config_name: “qwen-max” model_name: “qwen-max” api_key: “your-dashscope-api-key-here”在代码中我们通过配置名来加载模型而不是硬编码密钥import agentscope from agentscope.agents import DialogAgent # 初始化AgentScope加载配置文件 agentscope.init(model_configs./config.yaml) # 创建智能体时直接引用配置名 moderator DialogAgent(name主持人, model_config_namegpt-4) tech_expert DialogAgent(name技术专家, model_config_namegpt-4)关键经验务必使用配置文件管理模型密钥这不仅是安全最佳实践也极大方便了后续模型的切换和对比实验。比如你可以轻松地将“技术专家”的模型从gpt-4换成qwen-max只需修改配置名而无需改动智能体初始化代码。3.2 第二步创建定制化的智能体角色基础DialogAgent只是个“白板”我们需要赋予它们角色和个性。通过重写__init__和reply方法来实现。from agentscope.agents import DialogAgent from agentscope.message import Msg class ModeratorAgent(DialogAgent): 主持人智能体负责引导会议、总结发言。 def __init__(self, name, model_config_name, topic): super().__init__(namename, model_config_namemodel_config_name) self.topic topic self.discussion_rounds 0 self.summary_points [] def reply(self, incoming_message: Msg) - Msg: # 分析当前讨论状态 if “总结” in incoming_message.content or self.discussion_rounds 3: # 触发总结流程 summary_prompt f“基于以下讨论要点{self.summary_points}请撰写一份关于‘{self.topic}’的最终创新方案报告。” final_msg Msg(self.name, summary_prompt, role“user”) response super().reply(final_msg) return Msg(self.name, f“【最终方案】\n{response.content}”, role“assistant”) else: # 引导下一轮讨论 self.discussion_rounds 1 guide_prompt f“当前议题{self.topic}。这是第{self.discussion_rounds}轮讨论。请基于之前的对话提出一个新的、深入的见解或质疑以推动讨论。之前的讨论要点有{self.summary_points[-3:] if self.summary_points else ‘暂无’}” guide_msg Msg(self.name, guide_prompt, role“user”) return super().reply(guide_msg) def record_point(self, point): 记录讨论要点 self.summary_points.append(point) class ExpertAgent(DialogAgent): 领域专家基类 def __init__(self, name, model_config_name, expertise): super().__init__(namename, model_config_namemodel_config_name) self.expertise expertise # 如“区块链技术”、“用户体验设计” def reply(self, incoming_message: Msg) - Msg: # 在回复时强调自己的专家身份和视角 expert_prompt f“你是一名{self.expertise}专家。请从你的专业角度出发对以下内容发表看法\n\n{incoming_message.content}。你的回答应体现专业深度并可以适当质疑或补充其他观点。” expert_msg Msg(self.name, expert_prompt, role“user”) response super().reply(expert_msg) # 可以在这里简单提取观点关键词用于主持人记录实际项目可能需要更复杂的NLP if “观点” in response.content or “建议” in response.content: # 这里简化处理实际应更精细 self.moderator.record_point(f“{self.name}: {response.content[:100]}...”) return response这里的关键是我们通过定制reply方法在消息到达大模型之前为其穿上了“角色”的外衣即修改了Prompt。同时ModeratorAgent内部维护了讨论状态轮次、要点实现了简单的状态管理。3.3 第三步设计会议流程与消息循环有了智能体我们需要设计它们如何互动。这里我们不使用高级的Pipeline而是手动实现一个简单的循环以便更清晰地理解消息流。def run_brainstorming_session(topic, experts_config): 运行头脑风暴会议 Args: topic: 会议议题 experts_config: 专家列表如 [(AI伦理, gpt-4), (产品设计, gpt-4)] # 1. 创建智能体 moderator ModeratorAgent(name“主持人”, model_config_name“gpt-4”, topictopic) experts [] for expertise, model_cfg in experts_config: expert ExpertAgent(namef“{expertise}专家”, model_config_namemodel_cfg, expertiseexpertise) # 简单传递主持人引用用于记录观点生产环境需更优雅的设计如通过消息 expert.moderator moderator experts.append(expert) # 2. 初始化主持人开场 print(f“议题{topic}”) opening moderator.reply(Msg(“system”, “请宣布会议开始并介绍议题。”)) print(f“[主持人] {opening.content}”) # 3. 多轮讨论循环 for round_num in range(1, 4): # 进行3轮讨论 print(f“\n——— 第{round_num}轮讨论 ———”) for expert in experts: # 专家针对主持人上一轮的话或开场白进行发言 expert_response expert.reply(opening) print(f“[{expert.name}] {expert_response.content}”) # 主持人记录并准备引导下一轮 # 这里简化将专家回复直接作为主持人的新输入 opening Msg(expert.name, expert_response.content, role“user”) # 4. 总结 print(f“\n——— 会议总结 ———”) # 发送一个触发总结的信号给主持人 summary_trigger Msg(“system”, “讨论已进行三轮请进行总结。”) final_summary moderator.reply(summary_trigger) print(f“[主持人] {final_summary.content}”) return final_summary.content # 运行会议 if __name__ “__main__”: final_report run_brainstorming_session( topic“如何设计一个负责任且用户友好的AI健康助手” experts_config[(“AI伦理”, “gpt-4”), (“医疗健康”, “gpt-4”), (“用户体验”, “gpt-4”), (“数据安全”, “gpt-4”)] ) # 可以将final_report保存到文件或数据库 with open(“brainstorming_report.md”, “w”, encoding“utf-8”) as f: f.write(final_report)这个流程清晰地展示了多智能体应用的核心循环消息驱动。每个智能体的reply方法被触发产生新的Message这个Message又成为其他智能体或下一轮循环的输入。框架的价值在于它让Message的创建、传递和处理变得规范且可追踪。4. 深入原理AgentScope如何管理对话状态与上下文当我们使用DialogAgent与LLM对话时一个核心问题是LLM本身是无状态的。如何让智能体“记住”之前的对话这就是对话状态Dialog State或上下文管理Context Management要解决的问题。AgentScope在这方面做了精心的设计理解它对于高效使用和调试至关重要。4.1 记忆Memory模块智能体的“记事本”在AgentScope中每个Agent通常都有一个memory属性。这不是指LLM的参数权重而是指智能体与外界交互的历史记录。对于DialogAgent其memory默认是一个DialogMemory对象它就像一个“记事本”按顺序记录着这个智能体说过和听过的话。当你调用agent.reply(message)时背后发生了以下几步记忆写入将传入的incoming_message用户或其他智能体说的话添加到memory中。上下文构造框架从memory中提取最近的一些消息考虑LLM的token限制并按照一定的模板如ChatML格式、OpenAI格式组装成LLM能理解的“上下文”。LLM调用将组装好的上下文发送给LLM API。记忆写入与返回将LLM返回的响应包装成Message也添加到memory中然后返回给调用者。# 这是一个简化的内部逻辑示意 class DialogAgent: def reply(self, incoming_message): # 1. 记住收到的消息 self.memory.add(incoming_message) # 2. 从记忆中构建Prompt上下文 prompt_messages self._format_messages(self.memory.get_messages()) # 3. 调用LLM llm_response self.model.call(prompt_messages) # 4. 将响应包装成消息并记住它 response_message Msg(self.name, llm_response, role“assistant”) self.memory.add(response_message) return response_message4.2 上下文窗口与记忆管理策略LLM有token数限制不能把从盘古开天辟地以来的所有对话都塞进去。因此DialogMemory需要管理策略。常见策略有固定窗口只保留最近N条消息。Token数限制只保留总token数小于上限的最新消息。摘要压缩将较早的对话压缩成一段摘要。AgentScope允许你配置这些策略。例如在初始化DialogAgent时你可以指定一个自定义的memory对象。from agentscope.memory import TemporaryMemory # 创建一个临时记忆只保留最近5轮对话 temp_memory TemporaryMemory(store_frequency5) agent DialogAgent( name“短期记忆助手”, model_config_name“gpt-4”, memorytemp_memory # 使用自定义记忆 )踩坑提示上下文管理是成本与效果的平衡点。保留太多历史会浪费token增加费用和延迟甚至可能超出模型上下文长度导致报错。保留太少智能体可能“健忘”无法进行连贯的长对话。在实践中你需要根据任务性质来调整。对于头脑风暴会议可能需要较长的记忆来保持讨论连贯对于一个简单的问答工具可能只需要最近一两轮对话。4.3 共享记忆与独立记忆在我们的头脑风暴例子中每个专家智能体都有自己的独立记忆。但有时候我们需要智能体共享一部分记忆。例如一个“团队知识库”智能体它的记忆应该对所有其他智能体可见。AgentScope通过灵活的memory引用可以实现这一点。你可以让多个智能体指向同一个memory对象。from agentscope.memory import BufferMemory # 创建一个共享的记忆缓冲区 shared_team_memory BufferMemory() # 创建两个智能体共享同一份记忆 agent_a DialogAgent(name“A”, model_config_name“gpt-4”, memoryshared_team_memory) agent_b DialogAgent(name“B”, model_config_name“gpt-4”, memoryshared_team_memory) # 当A说了一句话这句话会被存入shared_team_memory msg_from_a agent_a.reply(Msg(“user”, “我们项目的目标是...”)) # 当B回复时它能“看到”A刚才说的话因为它们的记忆是共享的 msg_from_b agent_b.reply(msg_from_a)这种设计非常强大可以模拟出团队共识、公共白板等协作场景。理解记忆模块的工作原理是进行高级多智能体系统设计的基础。5. 高级特性与生产级考量当我们把Demo推向实际应用时会面临更多工程挑战。AgentScope提供了一系列特性来应对这些挑战。5.1 可观测性Observability与日志记录调试多智能体系统就像调试一个分布式系统日志至关重要。AgentScope内置了灵活的日志系统。import agentscope import logging # 设置日志级别和格式 agentscope.init( model_configs“./config.yaml”, # 设置项目名称用于日志区分 project“MultiAgent-Brainstorm”, # 配置日志 logging_levellogging.INFO, # 可以将日志保存到文件 save_dir“./logs” ) # 在代码中你可以使用标准的logging模块记录特定信息 logger logging.getLogger(__name__) logger.info(“头脑风暴会议开始...”)框架会自动记录关键事件如消息发送/接收、模型调用开始/结束、管道步骤执行等。查看这些日志你可以清晰地追踪到消息在智能体网络中的流动路径以及每个环节的耗时这对于性能分析和故障排查不可或缺。5.2 持久化Persistence与状态恢复对于长时间运行或重要的智能体会话你可能需要将会话状态主要是记忆保存下来以便后续恢复。AgentScope的memory对象通常支持序列化。import pickle import json # 保存某个智能体的记忆到文件 def save_agent_memory(agent, filepath): memory_data agent.memory.get_messages() # 获取原始消息列表 # 使用pickle保存二进制保存完整对象 with open(filepath, “wb”) as f: pickle.dump(memory_data, f) # 或者保存为JSON可读性好但可能丢失复杂对象信息 # with open(filepath, ‘w’, encoding‘utf-8’) as f: # json.dump([msg.to_dict() for msg in memory_data], f, ensure_asciiFalse, indent2) # 从文件加载记忆并恢复智能体状态 def load_agent_memory(agent, filepath): with open(filepath, “rb”) as f: memory_data pickle.load(f) agent.memory.clear() for msg_dict in memory_data: # 假设保存的是Msg对象的字典形式 agent.memory.add(Msg(**msg_dict))在生产环境中你可能会将会话状态存入数据库如Redis、PostgreSQL并为每个会话分配一个唯一的session_id。这样即使服务重启也能从断点继续。5.3 错误处理与智能体“熔断”多智能体系统中任何一个环节出错如LLM API调用失败、网络超时、智能体逻辑异常都可能导致整个流程崩溃。健壮的系统必须有错误处理机制。AgentScope的模型调用层通常已经包含了一些重试逻辑。但你需要在业务逻辑层也考虑容错。from openai import APIError, Timeout import time class RobustDialogAgent(DialogAgent): def reply(self, incoming_message, max_retries3): for attempt in range(max_retries): try: return super().reply(incoming_message) except (APIError, Timeout) as e: logger.warning(f“{self.name} 第{attempt1}次调用失败: {e}”) if attempt max_retries - 1: wait_time 2 ** attempt # 指数退避 time.sleep(wait_time) else: # 所有重试都失败返回一个降级响应 logger.error(f“{self.name} 所有重试均失败返回降级响应。”) return Msg(self.name, “抱歉我暂时无法处理您的请求请稍后再试或联系管理员。”, role“assistant”) except Exception as e: # 捕获其他未预期的异常 logger.error(f“{self.name} 发生未预期错误: {e}”, exc_infoTrue) return Msg(self.name, “系统内部出现错误。”, role“assistant”)此外对于管道Pipeline中的智能体你可以考虑实现“熔断”机制。如果某个智能体连续失败多次可以暂时将其从流程中跳过或替换防止单个故障点拖垮整个系统。5.4 性能优化异步与流式响应当智能体数量增多或任务复杂时同步顺序执行可能成为性能瓶颈。AgentScope支持异步Async操作允许智能体并行执行。import asyncio from agentscope.agents import DialogAgent from agentscope.message import Msg async def parallel_agent_task(agent, message): 异步执行单个智能体的回复任务 response await agent.areply(message) # 注意是 areply 异步方法 return response async def main_async(): agents [DialogAgent(namef“Agent{i}”, ...) for i in range(5)] initial_msg Msg(“user”, “请从你的视角分析这个问题。”) # 并行调用所有智能体 tasks [parallel_agent_task(agent, initial_msg) for agent in agents] responses await asyncio.gather(*tasks) for agent, resp in zip(agents, responses): print(f“{agent.name}: {resp.content[:50]}...”) # 运行异步主函数 asyncio.run(main_async())对于需要长时间生成内容的场景如撰写长报告流式响应Streaming可以提升用户体验。AgentScope的模型包装器通常也支持流式输出你可以逐步获取并处理生成的token而不是等待全部生成完毕。6. 超越基础构建复杂的智能体生态系统掌握了单机和基础协作后我们可以展望更复杂的架构。AgentScope能够支撑起一个智能体“生态系统”。6.1 分层架构管理者、执行者与工具智能体在一个大型系统中可以设计不同层级的智能体管理者/协调者Manager/Coordinator高层智能体负责分解任务、分配子任务给执行者、并汇总结果。它可能拥有更强的模型如GPT-4和更复杂的决策逻辑。执行者Worker专门负责某项具体任务的智能体如代码生成、数据分析、文本总结。它们可能使用性价比更高的模型如GPT-3.5-Turbo。工具智能体Tool Agent封装了特定工具或API调用的智能体如ServiceAgent。它们不一定依赖大模型而是执行确定性的操作运行代码、查询数据库。这种分层架构使得系统易于管理和扩展。管理者智能体可以通过Pipeline来编排多个执行者智能体的工作流。6.2 动态智能体创建与资源池在某些场景下你可能需要根据运行时需求动态创建或销毁智能体。例如一个客服系统中每来一个新用户会话就动态创建一个专属的客服智能体会话结束后销毁以释放资源。这需要结合工厂模式Factory Pattern和资源池Pool的概念。你可以维护一个智能体配置模板和模型连接池当需要时快速实例化避免每次创建都重新加载大模型等重量级资源。6.3 与外部系统的集成真正的生产力工具必须能与现有系统打通。AgentScope智能体可以轻松集成数据库通过ServiceAgent调用SQL查询或ORM。API服务智能体可以构造HTTP请求调用内部或外部的RESTful API。消息队列让智能体监听Kafka、RabbitMQ等消息队列实现事件驱动的响应。前端界面通过WebSocket或Server-Sent Events (SSE) 将智能体的流式响应推送到网页或App前端。例如创建一个可以查询天气的WeatherServiceAgentimport requests from agentscope.agents import ServiceAgent from agentscope.message import Msg class WeatherServiceAgent(ServiceAgent): def __init__(self, name, api_key): super().__init__(namename) self.api_key api_key self.base_url “https://api.weatherapi.com/v1” def reply(self, incoming_message: Msg) - Msg: # 解析消息内容提取城市名这里简化处理 city incoming_message.content.get(“city”, “Beijing”) try: response requests.get( f“{self.base_url}/current.json”, params{“key”: self.api_key, “q”: city} ) response.raise_for_status() data response.json() weather_info f“{city}当前天气{data[‘current’][‘condition’][‘text’]}温度{data[‘current’][‘temp_c’]}°C。” return Msg(self.name, weather_info, role“assistant”) except requests.RequestException as e: return Msg(self.name, f“获取天气信息失败{e}”, role“assistant”)通过这种方式你将大模型的推理能力与外部世界的实时数据和业务逻辑连接了起来构建出真正有用的智能应用。从最初手忙脚乱的脚本到如今能够设计并实现一个结构清晰、功能健壮的多智能体系统框架的价值在于它提供的不仅仅是代码工具更是一种工程化的思维模式。它强迫你思考智能体的边界、消息的契约、状态的归属和流程的编排。当你开始用AgentScope的视角去设计系统时你会发现构建复杂的AI协作应用不再是一个充满未知和混乱的探险而是一个有蓝图、有工具、有章可循的建造过程。剩下的就是充分发挥你的创意去解决那些真正有趣的问题了。 多智能体框架AgentScope：从消息总线到工程化实践
1. 从“玩具”到“工程”为什么我们需要一个多智能体框架最近在折腾AI应用尤其是想搞点多智能体协作的东西比如让几个大模型扮演不同角色一起完成一个复杂的任务。一开始我的想法很简单不就是调几个API写几个函数让它们互相发消息吗我随手写了个脚本用OpenAI的接口让一个“产品经理”智能体生成需求再让一个“程序员”智能体写代码。看起来跑通了但问题很快就来了。当我想增加一个“测试员”智能体来审查代码时消息传递的逻辑立刻变得混乱。谁该在什么时候收到什么消息消息内容是什么格式一个智能体的输出如何结构化地成为另一个智能体的输入更头疼的是我想记录下整个对话过程用于复盘或者当某个智能体“胡说八道”时能快速定位和干预。我的脚本迅速膨胀成一团乱麻调试起来像在走迷宫。我相信很多开发者都有过类似的经历。我们用一个for循环和几个if-else就能快速验证一个多智能体协作的想法这很棒。但一旦你想把这个“玩具”变成一个健壮的、可维护的、可扩展的“工程化”应用时就会遇到一系列基础设施层面的挑战。这正是像AgentScope这样的框架要解决的问题。它不是一个束缚你创意的笼子而是一个为你处理所有脏活累活的基础设施团队让你能专注于智能体本身的行为逻辑和业务价值。简单说AgentScope的核心价值在于它提供了一套标准化的“脚手架”和“工具箱”专门用于构建和管理多智能体应用。它把智能体间的通信、状态管理、流程编排、可观测性这些繁琐但必需的工程问题封装成了清晰、易用的接口。这就好比你想盖房子框架为你准备好了标准化的钢筋、水泥和施工流程你不用再从烧制砖头开始。2. AgentScope的核心架构消息总线与智能体“公寓”要理解AgentScope不能只看它提供的几个Agent类得先看清它设计的“地基”。这个地基的核心我认为是它的消息传递模型和智能体生命周期管理。我们可以把它想象成一栋专门出租给“智能体”的公寓楼。2.1 消息总线公寓楼里的内部电话系统在这栋公寓楼里智能体们房客不能直接砸对方的门。所有通信必须通过一个中央的“内部电话系统”——这就是消息总线Message Bus。在AgentScope中这个角色主要由Message类和底层的分发机制承担。一个Message不是一个简单的字符串。它是一个结构化的数据对象通常包含name: 发送者标识。content: 消息内容可以是文本、字典、甚至任何Python对象。url: 可选关联的文件或资源链接。为什么需要结构化举个例子智能体A发送了一个包含代码和解释的消息。如果只是字符串智能体B需要费力地去解析哪里是代码哪里是注释。而通过结构化的content比如{code: “print(‘hello’)”, “explanation”: “这是一个示例”}接收者就能直接获取语义明确的字段。更重要的是消息总线管理着消息的路由。它确保消息能准确送达目标智能体并且可以方便地实现广播发给所有智能体、组播发给特定群组等模式。框架还处理了消息的序列化、持久化如果你想保存对话历史和潜在的消息队列问题这些都是你在裸写脚本时需要自己反复造轮子的部分。2.2 智能体Agent标准化的“房客”与自定义装修AgentScope提供了一些基础“房客”模板也就是内置的Agent类比如UserAgent: 代表人类用户通常用于模拟用户输入或作为交互接口。DialogAgent: 一个基于大语言模型的对话智能体它封装了与LLM如GPT、文心一言等的交互包括构造Prompt、调用API、解析响应。ServiceAgent: 可以调用外部工具或服务的智能体比如执行代码、查询数据库、调用API。这些内置智能体帮你解决了最通用的模式。但真正的力量在于“自定义装修”。你可以通过继承这些基础类来创建具有特定技能、记忆或决策逻辑的智能体。例如创建一个“代码评审专家”智能体from agentscope.agents import DialogAgent from agentscope.message import Msg class CodeReviewAgent(DialogAgent): def __init__(self, name, model_config_name): super().__init__(namename, model_config_namemodel_config_name) # 可以在这里初始化该智能体的专属知识库或规则 self.review_rules [检查语法错误, 评估代码效率, 确保符合PEP8规范] def reply(self, incoming_message: Msg) - Msg: # 在调用父类即LLM的回复逻辑前可以预处理输入 enhanced_prompt f 你是一个资深代码评审专家。请根据以下规则进行评审 {self.review_rules} 需要评审的代码 {incoming_message.content} 请给出详细的评审意见包括问题、风险和改进建议。 # 修改传入消息的内容再交给父类的LLM处理 modified_msg Msg(self.name, enhanced_prompt, roleuser) return super().reply(modified_msg)通过继承和重写reply方法你就能轻松地给智能体注入特定的行为模式而无需关心消息是如何收发的、LLM的token如何管理、上下文窗口如何维护——这些都由框架层处理好了。2.3 环境与流程编排公寓的公共空间与活动流程智能体们住在各自的房间里通过消息总线打电话。那它们怎么协同完成一个任务呢比如“一起写一份项目计划书”。这就需要流程编排Orchestration。AgentScope提供了Pipeline、Workflow等抽象来定义智能体间的协作流程。你可以把它看作是公寓的“公共活动流程表”。一个最简单的顺序管道示例from agentscope.pipelines import SequentialPipeline # 假设我们已经创建了三个智能体 product_manager DialogAgent(namePM, ...) programmer DialogAgent(nameDev, ...) tester DialogAgent(nameQA, ...) # 创建一个顺序执行的管道 pipeline SequentialPipeline( agents[product_manager, programmer, tester], pipeline_name软件研发流水线 ) # 运行管道PM生成需求 - Dev根据需求写代码 - QA测试代码 initial_message Msg(user, 我们需要一个计算斐波那契数列的函数。) final_result pipeline.run(initial_message)在这个管道里框架会自动将上一个智能体的输出消息作为下一个智能体的输入消息。你还可以构建更复杂的流程比如包含条件分支IfElsePipeline、循环ForLoopPipeline或者并行执行ParallelPipeline。这种声明式的编排方式让复杂的多智能体交互逻辑变得清晰、可视且易于维护。3. 从零搭建一个实战项目多智能体头脑风暴会议理论讲得再多不如亲手搭一个。我们来实战一个经典场景多智能体头脑风暴会议。目标是创建一个会议系统包含一个主持人Moderator、若干个领域专家如技术专家、市场专家、设计专家让它们针对一个议题进行多轮讨论最终由主持人汇总成一份创新方案。3.1 第一步环境搭建与模型配置首先安装AgentScope。推荐使用pip并指定版本以获得稳定体验。pip install agentscope接下来是模型配置。AgentScope支持多种模型后端OpenAI、Azure、DashScope、Ollama等。我们需要在项目根目录创建一个config.yaml或configs.py来集中管理配置。这是非常重要的一步它能将模型API密钥、基地址等敏感或易变信息从代码中分离出来。# config.yaml model_configs: # 定义一个名为“gpt-4”的配置 gpt-4: model_type: “openai” # 指定模型类型 config_name: “gpt-4” # 配置名称在代码中引用 model_name: “gpt-4-turbo-preview” # 具体的模型名称 api_key: “your-openai-api-key-here” # 你的API密钥 # 可选参数 max_tokens: 2000 temperature: 0.7 # 可以定义多个配置例如备用模型或不同能力的模型 qwen-max: model_type: “dashscope” config_name: “qwen-max” model_name: “qwen-max” api_key: “your-dashscope-api-key-here”在代码中我们通过配置名来加载模型而不是硬编码密钥import agentscope from agentscope.agents import DialogAgent # 初始化AgentScope加载配置文件 agentscope.init(model_configs./config.yaml) # 创建智能体时直接引用配置名 moderator DialogAgent(name主持人, model_config_namegpt-4) tech_expert DialogAgent(name技术专家, model_config_namegpt-4)关键经验务必使用配置文件管理模型密钥这不仅是安全最佳实践也极大方便了后续模型的切换和对比实验。比如你可以轻松地将“技术专家”的模型从gpt-4换成qwen-max只需修改配置名而无需改动智能体初始化代码。3.2 第二步创建定制化的智能体角色基础DialogAgent只是个“白板”我们需要赋予它们角色和个性。通过重写__init__和reply方法来实现。from agentscope.agents import DialogAgent from agentscope.message import Msg class ModeratorAgent(DialogAgent): 主持人智能体负责引导会议、总结发言。 def __init__(self, name, model_config_name, topic): super().__init__(namename, model_config_namemodel_config_name) self.topic topic self.discussion_rounds 0 self.summary_points [] def reply(self, incoming_message: Msg) - Msg: # 分析当前讨论状态 if “总结” in incoming_message.content or self.discussion_rounds 3: # 触发总结流程 summary_prompt f“基于以下讨论要点{self.summary_points}请撰写一份关于‘{self.topic}’的最终创新方案报告。” final_msg Msg(self.name, summary_prompt, role“user”) response super().reply(final_msg) return Msg(self.name, f“【最终方案】\n{response.content}”, role“assistant”) else: # 引导下一轮讨论 self.discussion_rounds 1 guide_prompt f“当前议题{self.topic}。这是第{self.discussion_rounds}轮讨论。请基于之前的对话提出一个新的、深入的见解或质疑以推动讨论。之前的讨论要点有{self.summary_points[-3:] if self.summary_points else ‘暂无’}” guide_msg Msg(self.name, guide_prompt, role“user”) return super().reply(guide_msg) def record_point(self, point): 记录讨论要点 self.summary_points.append(point) class ExpertAgent(DialogAgent): 领域专家基类 def __init__(self, name, model_config_name, expertise): super().__init__(namename, model_config_namemodel_config_name) self.expertise expertise # 如“区块链技术”、“用户体验设计” def reply(self, incoming_message: Msg) - Msg: # 在回复时强调自己的专家身份和视角 expert_prompt f“你是一名{self.expertise}专家。请从你的专业角度出发对以下内容发表看法\n\n{incoming_message.content}。你的回答应体现专业深度并可以适当质疑或补充其他观点。” expert_msg Msg(self.name, expert_prompt, role“user”) response super().reply(expert_msg) # 可以在这里简单提取观点关键词用于主持人记录实际项目可能需要更复杂的NLP if “观点” in response.content or “建议” in response.content: # 这里简化处理实际应更精细 self.moderator.record_point(f“{self.name}: {response.content[:100]}...”) return response这里的关键是我们通过定制reply方法在消息到达大模型之前为其穿上了“角色”的外衣即修改了Prompt。同时ModeratorAgent内部维护了讨论状态轮次、要点实现了简单的状态管理。3.3 第三步设计会议流程与消息循环有了智能体我们需要设计它们如何互动。这里我们不使用高级的Pipeline而是手动实现一个简单的循环以便更清晰地理解消息流。def run_brainstorming_session(topic, experts_config): 运行头脑风暴会议 Args: topic: 会议议题 experts_config: 专家列表如 [(AI伦理, gpt-4), (产品设计, gpt-4)] # 1. 创建智能体 moderator ModeratorAgent(name“主持人”, model_config_name“gpt-4”, topictopic) experts [] for expertise, model_cfg in experts_config: expert ExpertAgent(namef“{expertise}专家”, model_config_namemodel_cfg, expertiseexpertise) # 简单传递主持人引用用于记录观点生产环境需更优雅的设计如通过消息 expert.moderator moderator experts.append(expert) # 2. 初始化主持人开场 print(f“议题{topic}”) opening moderator.reply(Msg(“system”, “请宣布会议开始并介绍议题。”)) print(f“[主持人] {opening.content}”) # 3. 多轮讨论循环 for round_num in range(1, 4): # 进行3轮讨论 print(f“\n——— 第{round_num}轮讨论 ———”) for expert in experts: # 专家针对主持人上一轮的话或开场白进行发言 expert_response expert.reply(opening) print(f“[{expert.name}] {expert_response.content}”) # 主持人记录并准备引导下一轮 # 这里简化将专家回复直接作为主持人的新输入 opening Msg(expert.name, expert_response.content, role“user”) # 4. 总结 print(f“\n——— 会议总结 ———”) # 发送一个触发总结的信号给主持人 summary_trigger Msg(“system”, “讨论已进行三轮请进行总结。”) final_summary moderator.reply(summary_trigger) print(f“[主持人] {final_summary.content}”) return final_summary.content # 运行会议 if __name__ “__main__”: final_report run_brainstorming_session( topic“如何设计一个负责任且用户友好的AI健康助手” experts_config[(“AI伦理”, “gpt-4”), (“医疗健康”, “gpt-4”), (“用户体验”, “gpt-4”), (“数据安全”, “gpt-4”)] ) # 可以将final_report保存到文件或数据库 with open(“brainstorming_report.md”, “w”, encoding“utf-8”) as f: f.write(final_report)这个流程清晰地展示了多智能体应用的核心循环消息驱动。每个智能体的reply方法被触发产生新的Message这个Message又成为其他智能体或下一轮循环的输入。框架的价值在于它让Message的创建、传递和处理变得规范且可追踪。4. 深入原理AgentScope如何管理对话状态与上下文当我们使用DialogAgent与LLM对话时一个核心问题是LLM本身是无状态的。如何让智能体“记住”之前的对话这就是对话状态Dialog State或上下文管理Context Management要解决的问题。AgentScope在这方面做了精心的设计理解它对于高效使用和调试至关重要。4.1 记忆Memory模块智能体的“记事本”在AgentScope中每个Agent通常都有一个memory属性。这不是指LLM的参数权重而是指智能体与外界交互的历史记录。对于DialogAgent其memory默认是一个DialogMemory对象它就像一个“记事本”按顺序记录着这个智能体说过和听过的话。当你调用agent.reply(message)时背后发生了以下几步记忆写入将传入的incoming_message用户或其他智能体说的话添加到memory中。上下文构造框架从memory中提取最近的一些消息考虑LLM的token限制并按照一定的模板如ChatML格式、OpenAI格式组装成LLM能理解的“上下文”。LLM调用将组装好的上下文发送给LLM API。记忆写入与返回将LLM返回的响应包装成Message也添加到memory中然后返回给调用者。# 这是一个简化的内部逻辑示意 class DialogAgent: def reply(self, incoming_message): # 1. 记住收到的消息 self.memory.add(incoming_message) # 2. 从记忆中构建Prompt上下文 prompt_messages self._format_messages(self.memory.get_messages()) # 3. 调用LLM llm_response self.model.call(prompt_messages) # 4. 将响应包装成消息并记住它 response_message Msg(self.name, llm_response, role“assistant”) self.memory.add(response_message) return response_message4.2 上下文窗口与记忆管理策略LLM有token数限制不能把从盘古开天辟地以来的所有对话都塞进去。因此DialogMemory需要管理策略。常见策略有固定窗口只保留最近N条消息。Token数限制只保留总token数小于上限的最新消息。摘要压缩将较早的对话压缩成一段摘要。AgentScope允许你配置这些策略。例如在初始化DialogAgent时你可以指定一个自定义的memory对象。from agentscope.memory import TemporaryMemory # 创建一个临时记忆只保留最近5轮对话 temp_memory TemporaryMemory(store_frequency5) agent DialogAgent( name“短期记忆助手”, model_config_name“gpt-4”, memorytemp_memory # 使用自定义记忆 )踩坑提示上下文管理是成本与效果的平衡点。保留太多历史会浪费token增加费用和延迟甚至可能超出模型上下文长度导致报错。保留太少智能体可能“健忘”无法进行连贯的长对话。在实践中你需要根据任务性质来调整。对于头脑风暴会议可能需要较长的记忆来保持讨论连贯对于一个简单的问答工具可能只需要最近一两轮对话。4.3 共享记忆与独立记忆在我们的头脑风暴例子中每个专家智能体都有自己的独立记忆。但有时候我们需要智能体共享一部分记忆。例如一个“团队知识库”智能体它的记忆应该对所有其他智能体可见。AgentScope通过灵活的memory引用可以实现这一点。你可以让多个智能体指向同一个memory对象。from agentscope.memory import BufferMemory # 创建一个共享的记忆缓冲区 shared_team_memory BufferMemory() # 创建两个智能体共享同一份记忆 agent_a DialogAgent(name“A”, model_config_name“gpt-4”, memoryshared_team_memory) agent_b DialogAgent(name“B”, model_config_name“gpt-4”, memoryshared_team_memory) # 当A说了一句话这句话会被存入shared_team_memory msg_from_a agent_a.reply(Msg(“user”, “我们项目的目标是...”)) # 当B回复时它能“看到”A刚才说的话因为它们的记忆是共享的 msg_from_b agent_b.reply(msg_from_a)这种设计非常强大可以模拟出团队共识、公共白板等协作场景。理解记忆模块的工作原理是进行高级多智能体系统设计的基础。5. 高级特性与生产级考量当我们把Demo推向实际应用时会面临更多工程挑战。AgentScope提供了一系列特性来应对这些挑战。5.1 可观测性Observability与日志记录调试多智能体系统就像调试一个分布式系统日志至关重要。AgentScope内置了灵活的日志系统。import agentscope import logging # 设置日志级别和格式 agentscope.init( model_configs“./config.yaml”, # 设置项目名称用于日志区分 project“MultiAgent-Brainstorm”, # 配置日志 logging_levellogging.INFO, # 可以将日志保存到文件 save_dir“./logs” ) # 在代码中你可以使用标准的logging模块记录特定信息 logger logging.getLogger(__name__) logger.info(“头脑风暴会议开始...”)框架会自动记录关键事件如消息发送/接收、模型调用开始/结束、管道步骤执行等。查看这些日志你可以清晰地追踪到消息在智能体网络中的流动路径以及每个环节的耗时这对于性能分析和故障排查不可或缺。5.2 持久化Persistence与状态恢复对于长时间运行或重要的智能体会话你可能需要将会话状态主要是记忆保存下来以便后续恢复。AgentScope的memory对象通常支持序列化。import pickle import json # 保存某个智能体的记忆到文件 def save_agent_memory(agent, filepath): memory_data agent.memory.get_messages() # 获取原始消息列表 # 使用pickle保存二进制保存完整对象 with open(filepath, “wb”) as f: pickle.dump(memory_data, f) # 或者保存为JSON可读性好但可能丢失复杂对象信息 # with open(filepath, ‘w’, encoding‘utf-8’) as f: # json.dump([msg.to_dict() for msg in memory_data], f, ensure_asciiFalse, indent2) # 从文件加载记忆并恢复智能体状态 def load_agent_memory(agent, filepath): with open(filepath, “rb”) as f: memory_data pickle.load(f) agent.memory.clear() for msg_dict in memory_data: # 假设保存的是Msg对象的字典形式 agent.memory.add(Msg(**msg_dict))在生产环境中你可能会将会话状态存入数据库如Redis、PostgreSQL并为每个会话分配一个唯一的session_id。这样即使服务重启也能从断点继续。5.3 错误处理与智能体“熔断”多智能体系统中任何一个环节出错如LLM API调用失败、网络超时、智能体逻辑异常都可能导致整个流程崩溃。健壮的系统必须有错误处理机制。AgentScope的模型调用层通常已经包含了一些重试逻辑。但你需要在业务逻辑层也考虑容错。from openai import APIError, Timeout import time class RobustDialogAgent(DialogAgent): def reply(self, incoming_message, max_retries3): for attempt in range(max_retries): try: return super().reply(incoming_message) except (APIError, Timeout) as e: logger.warning(f“{self.name} 第{attempt1}次调用失败: {e}”) if attempt max_retries - 1: wait_time 2 ** attempt # 指数退避 time.sleep(wait_time) else: # 所有重试都失败返回一个降级响应 logger.error(f“{self.name} 所有重试均失败返回降级响应。”) return Msg(self.name, “抱歉我暂时无法处理您的请求请稍后再试或联系管理员。”, role“assistant”) except Exception as e: # 捕获其他未预期的异常 logger.error(f“{self.name} 发生未预期错误: {e}”, exc_infoTrue) return Msg(self.name, “系统内部出现错误。”, role“assistant”)此外对于管道Pipeline中的智能体你可以考虑实现“熔断”机制。如果某个智能体连续失败多次可以暂时将其从流程中跳过或替换防止单个故障点拖垮整个系统。5.4 性能优化异步与流式响应当智能体数量增多或任务复杂时同步顺序执行可能成为性能瓶颈。AgentScope支持异步Async操作允许智能体并行执行。import asyncio from agentscope.agents import DialogAgent from agentscope.message import Msg async def parallel_agent_task(agent, message): 异步执行单个智能体的回复任务 response await agent.areply(message) # 注意是 areply 异步方法 return response async def main_async(): agents [DialogAgent(namef“Agent{i}”, ...) for i in range(5)] initial_msg Msg(“user”, “请从你的视角分析这个问题。”) # 并行调用所有智能体 tasks [parallel_agent_task(agent, initial_msg) for agent in agents] responses await asyncio.gather(*tasks) for agent, resp in zip(agents, responses): print(f“{agent.name}: {resp.content[:50]}...”) # 运行异步主函数 asyncio.run(main_async())对于需要长时间生成内容的场景如撰写长报告流式响应Streaming可以提升用户体验。AgentScope的模型包装器通常也支持流式输出你可以逐步获取并处理生成的token而不是等待全部生成完毕。6. 超越基础构建复杂的智能体生态系统掌握了单机和基础协作后我们可以展望更复杂的架构。AgentScope能够支撑起一个智能体“生态系统”。6.1 分层架构管理者、执行者与工具智能体在一个大型系统中可以设计不同层级的智能体管理者/协调者Manager/Coordinator高层智能体负责分解任务、分配子任务给执行者、并汇总结果。它可能拥有更强的模型如GPT-4和更复杂的决策逻辑。执行者Worker专门负责某项具体任务的智能体如代码生成、数据分析、文本总结。它们可能使用性价比更高的模型如GPT-3.5-Turbo。工具智能体Tool Agent封装了特定工具或API调用的智能体如ServiceAgent。它们不一定依赖大模型而是执行确定性的操作运行代码、查询数据库。这种分层架构使得系统易于管理和扩展。管理者智能体可以通过Pipeline来编排多个执行者智能体的工作流。6.2 动态智能体创建与资源池在某些场景下你可能需要根据运行时需求动态创建或销毁智能体。例如一个客服系统中每来一个新用户会话就动态创建一个专属的客服智能体会话结束后销毁以释放资源。这需要结合工厂模式Factory Pattern和资源池Pool的概念。你可以维护一个智能体配置模板和模型连接池当需要时快速实例化避免每次创建都重新加载大模型等重量级资源。6.3 与外部系统的集成真正的生产力工具必须能与现有系统打通。AgentScope智能体可以轻松集成数据库通过ServiceAgent调用SQL查询或ORM。API服务智能体可以构造HTTP请求调用内部或外部的RESTful API。消息队列让智能体监听Kafka、RabbitMQ等消息队列实现事件驱动的响应。前端界面通过WebSocket或Server-Sent Events (SSE) 将智能体的流式响应推送到网页或App前端。例如创建一个可以查询天气的WeatherServiceAgentimport requests from agentscope.agents import ServiceAgent from agentscope.message import Msg class WeatherServiceAgent(ServiceAgent): def __init__(self, name, api_key): super().__init__(namename) self.api_key api_key self.base_url “https://api.weatherapi.com/v1” def reply(self, incoming_message: Msg) - Msg: # 解析消息内容提取城市名这里简化处理 city incoming_message.content.get(“city”, “Beijing”) try: response requests.get( f“{self.base_url}/current.json”, params{“key”: self.api_key, “q”: city} ) response.raise_for_status() data response.json() weather_info f“{city}当前天气{data[‘current’][‘condition’][‘text’]}温度{data[‘current’][‘temp_c’]}°C。” return Msg(self.name, weather_info, role“assistant”) except requests.RequestException as e: return Msg(self.name, f“获取天气信息失败{e}”, role“assistant”)通过这种方式你将大模型的推理能力与外部世界的实时数据和业务逻辑连接了起来构建出真正有用的智能应用。从最初手忙脚乱的脚本到如今能够设计并实现一个结构清晰、功能健壮的多智能体系统框架的价值在于它提供的不仅仅是代码工具更是一种工程化的思维模式。它强迫你思考智能体的边界、消息的契约、状态的归属和流程的编排。当你开始用AgentScope的视角去设计系统时你会发现构建复杂的AI协作应用不再是一个充满未知和混乱的探险而是一个有蓝图、有工具、有章可循的建造过程。剩下的就是充分发挥你的创意去解决那些真正有趣的问题了。