1. 项目初探AgentScope一个面向多智能体应用的开源框架最近在折腾大模型应用开发特别是涉及到多个AI智能体Agent协同工作的场景时我遇到了不少麻烦。比如如何让不同的智能体比如一个负责写代码一个负责审核代码顺畅地对话如何管理它们之间的消息流如何优雅地处理不同模型API的调用差异这些问题在单个智能体的场景下还好一旦涉及到多智能体协作复杂度就指数级上升。就在我为此头疼的时候一个名为AgentScope的开源项目进入了我的视野。AgentScope项目地址agentscope-ai/agentscope是一个由智谱AI开源的、专门为构建多智能体应用而设计的框架。简单来说它不是一个单一的智能体工具而是一个“舞台导演”或者“编排系统”让你能轻松地定义多个拥有不同角色和能力的智能体并设计它们之间的交互规则从而完成复杂的任务。它的出现正是为了解决像我这样希望利用大模型能力构建更复杂、更智能应用却又苦于底层通信、状态管理和工具集成的开发者们的痛点。这个框架的核心价值在于“降低多智能体应用开发的门槛”。它提供了一套高层的、声明式的API让你可以像搭积木一样组合智能体而无需关心底层的网络通信、并发控制、消息序列化等繁琐细节。无论是想构建一个模拟辩论的AI系统还是一个自动化代码审查流水线或者是一个复杂的游戏NPC交互环境AgentScope都试图提供一个统一、高效的开发范式。接下来我将深入拆解它的架构、核心概念以及如何上手并分享一些在实际使用中的心得体会。2. AgentScope的核心架构与设计哲学要理解一个框架首先要看它的设计思路。AgentScope的架构清晰地反映了其“以智能体为中心以消息为纽带”的设计哲学。整个框架可以粗略地分为三层智能体层、服务层和基础设施层。2.1 智能体Agent能力的抽象与封装在AgentScope中Agent是最核心的抽象。每一个智能体都是一个独立的、具有特定角色、记忆和能力的实体。框架内置了几种常见的智能体类型这也是我们上手时最常打交道的部分DialogAgent对话智能体这是最基础的智能体它封装了一个大语言模型LLM的调用。你给它一个系统提示词System Prompt来定义其角色如“你是一个严谨的代码审查专家”它就能根据接收到的消息调用背后的LLM生成回复。它主要负责“思考”和“表达”。UserAgent用户智能体这个智能体代表真实用户。在交互式应用中它负责接收用户在终端或Web界面上的输入并将其转化为框架内部的标准消息格式。在自动化流程中它也可以被编程为按特定规则生成输入。ToolAgent工具智能体这是让智能体“动手”能力的关键。一个ToolAgent内部封装了一个或多个可执行的工具函数例如调用搜索引擎API、执行一段Python代码、查询数据库。当它收到一个请求时会尝试理解请求内容选择并调用合适的工具然后将工具执行的结果包装成消息回复出去。这极大地扩展了智能体的能力边界使其不再局限于文本生成。ActorAgent角色扮演智能体这是一个更灵活的抽象允许你自定义智能体的行为逻辑。你可以完全控制其reply函数实现任何你想要的反应模式比如基于规则的响应、调用其他服务等。这些智能体类型并不是孤立的你可以通过继承来创建复合型智能体。例如你可以创建一个既擅长对话又能使用特定工具的智能体。2.2 消息Message智能体间通信的通用语言智能体之间如何交流靠的是标准化的Message对象。AgentScope定义了一套清晰的消息格式每条消息通常包含name发送者、content内容等字段。这种设计的好处是解耦智能体无需知道消息来自哪个具体的智能体实例它只处理符合格式的消息。可追溯所有交互历史都以消息序列的形式保存便于调试、分析和重现。灵活性可以轻松地在消息流水线中插入过滤器、转换器或记录器。2.3 服务与编排让智能体动起来有了智能体和消息还需要一个“导演”来指挥它们如何互动。这就是AgentScope的服务层和编排能力。AgentService你可以将智能体包装成一个独立的服务例如一个HTTP服务。这使得智能体能力可以被远程调用方便构建分布式系统或微服务架构。流水线Pipeline与工作流这是实现复杂交互的核心。框架提供了直观的方式来定义智能体之间的交互顺序。最常见的是使用sequentialpipeline它让消息按顺序从一个智能体传递到下一个。对于更复杂的条件分支或循环交互你可以利用Python的控制流如if-else,for循环结合框架的API来灵活编排。2.4 设计哲学易用性与灵活性的平衡从上述架构可以看出AgentScope在努力做一件事把复杂留给自己把简单留给开发者。它通过高层次的抽象隐藏了多智能体系统中诸如并发安全、消息路由、错误处理等棘手问题。开发者只需要关注两件事1) 定义智能体的角色和能力2) 设计它们之间的交互逻辑。这种“约定大于配置”的思路能显著提升开发效率。3. 从零开始搭建你的第一个多智能体对话系统理论讲得再多不如动手一试。让我们来构建一个最简单的场景一个UserAgent模拟用户向一个DialogAgent扮演助手提问并得到回答。这个例子虽然简单但涵盖了最核心的流程。3.1 环境准备与安装首先确保你的Python环境在3.8及以上。然后通过pip安装AgentScopepip install agentscope安装过程会同时安装一些核心依赖。如果你想使用特定的模型比如OpenAI的GPT系列或智谱的GLM还需要安装对应的SDKpip install openai # 如需使用OpenAI # 或者 pip install zhipuai # 如需使用智谱GLM3.2 初始化模型与创建智能体AgentScope支持多种模型后端在使用前需要进行全局初始化。这里以使用OpenAI API为例你需要准备自己的API Keyimport agentscope from agentscope.agents import DialogAgent, UserAgent from agentscope.pipelines import sequentialpipeline # 第一步初始化模型配置 agentscope.init( model_configs[ { model_type: openai, # 指定模型类型 config_name: gpt-4, # 配置名称可自定义 model_name: gpt-4, # 实际调用的模型名 api_key: your-openai-api-key-here, # 替换为你的Key organization: your-org, # 可选组织ID } ] )初始化完成后我们就可以创建智能体了。创建时需要为它们指定一个名字和所使用的模型配置。# 第二步创建智能体 assistant_agent DialogAgent( nameAssistant, sys_prompt你是一个乐于助人且知识渊博的AI助手。请用中文回答用户的问题。, model_config_namegpt-4, # 使用上面初始化的配置 ) user_agent UserAgent(nameUser)DialogAgent的sys_prompt参数至关重要它定义了智能体的“人设”和行为准则。好的提示词能极大提升智能体的表现。model_config_name必须与init中定义的config_name一致。3.3 构建交互流水线并运行智能体创建好了现在需要定义它们如何交互。我们使用最简单的顺序流水线# 第三步定义并运行流水线 def main(): # sequentialpipeline 会按顺序执行传入的函数或智能体 # 这里先运行user_agent获取用户输入再将结果传给assistant_agent得到回复 pipeline sequentialpipeline([user_agent, assistant_agent]) print(多智能体对话系统已启动。输入‘退出’或‘quit’结束对话。) while True: # 运行一次流水线触发一轮对话 result pipeline() # 结果是一个消息列表这里我们简单打印最后一条消息助手的回复 if result and len(result) 0: last_msg result[-1] print(f\n[Assistant]: {last_msg[content]}\n) # 一个简单的退出检查实际中UserAgent可能会处理 # 这里为了演示我们检查用户输入 # 注意这是一个简化的逻辑真实场景需要更健壮的处理 user_input_msg result[0] if result else None if user_input_msg and user_input_msg[content].strip().lower() in [退出, quit]: print(对话结束。) break if __name__ __main__: main()运行这段代码你会在终端看到一个交互式对话界面。你输入问题UserAgent将其包装成消息sequentialpipeline驱动消息先经过UserAgent获取输入再传递给Assistant生成回复最后将结果返回。注意这个示例中的退出逻辑比较粗糙。在实际应用中更优雅的做法是在UserAgent内部或通过一个专门的ControllerAgent来管理对话状态和退出条件。这里为了突出核心流程做了简化。3.4 第一个项目的总结与常见坑点通过这个最小化的例子你应该已经感受到了AgentScope的便捷定义Agent - 组合Pipeline - 运行。三步就完成了一个可交互的多智能体系统雏形。在这个过程中最容易出错的点往往是模型配置错误init中的config_name和创建Agent时的model_config_name必须完全一致包括大小写。建议用一个常量来管理这些配置名。API Key等敏感信息管理切勿将API Key硬编码在代码中提交到版本库。务必使用环境变量或配置文件来管理。import os api_key os.getenv(OPENAI_API_KEY)提示词Prompt设计sys_prompt的质量直接决定智能体的表现。对于中文场景明确要求“用中文回答”很重要。更复杂的角色需要更细致、包含示例Few-shot的提示词。4. 进阶实战构建一个代码审查智能体协作系统单一的问答应答显然无法体现多智能体的威力。让我们设计一个更贴近实际开发的场景一个自动化的代码审查流水线。这个系统包含三个智能体CoderAgent程序员智能体负责提交待审查的代码片段。ReviewerAgent审查员智能体一个ToolAgent它不仅能理解代码还能调用一个静态代码分析工具例如pylint或bandit来检查代码中的潜在问题。SummarizerAgent总结员智能体负责汇总审查员的反馈并以清晰、友好的格式生成最终审查报告给程序员。这个例子将展示如何集成工具、如何处理更复杂的消息流。4.1 设计工具函数并创建ToolAgent首先我们需要为ReviewerAgent准备工具。这里我们模拟一个调用bandit一个Python安全漏洞扫描工具的工具函数。import subprocess import tempfile import agentscope from agentscope.agents import ToolAgent from agentscope.tools import tool # 定义一个工具函数使用tool装饰器注册 tool def run_bandit_security_scan(code_snippet: str) - str: 使用bandit对提供的Python代码片段进行安全扫描。 Args: code_snippet: 需要扫描的Python代码字符串。 Returns: 扫描结果的文本报告。如果扫描过程出错返回错误信息。 # 将代码片段写入临时文件 with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(code_snippet) temp_file_path f.name try: # 调用bandit命令进行扫描 # -q: 安静模式减少冗余输出 # -f: 指定输出格式为text result subprocess.run( [bandit, -q, -f, text, temp_file_path], capture_outputTrue, textTrue, timeout30 # 设置超时防止长时间运行 ) output result.stdout if result.returncode ! 0 and not output: output result.stderr or Bandit扫描过程出现未知错误。 return output except subprocess.TimeoutExpired: return 安全扫描超时代码可能过于复杂或存在死循环风险。 except FileNotFoundError: return 错误未找到bandit命令。请确保已通过‘pip install bandit’安装。 except Exception as e: return f执行安全扫描时发生异常{str(e)} finally: # 清理临时文件 import os os.unlink(temp_file_path)现在用这个工具函数来创建我们的ReviewerAgent# 假设模型配置已通过 agentscope.init() 初始化例如使用GPT-4 model_config_name gpt-4 # 与init中的config_name对应 reviewer_agent ToolAgent( nameSecurityReviewer, model_config_namemodel_config_name, tools[run_bandit_security_scan], # 传入工具函数列表 sys_prompt你是一个专业的代码安全审查员。你的任务是 1. 分析用户提供的Python代码理解其功能。 2. 自动调用安全扫描工具bandit对代码进行漏洞检测。 3. 结合工具扫描结果和你自身的代码安全知识生成一份全面的审查报告。 报告需包含 - 代码功能简述。 - 工具扫描发现的具体问题列出每个问题的严重等级、位置和描述。 - 基于最佳实践给出的修改建议。 - 如果扫描未发现问题也请说明。 请用清晰、有条理的中文撰写报告。 )ToolAgent的强大之处在于它能够根据用户的请求和工具的描述自动决定何时以及如何调用工具。你只需要把工具函数给它它会在对话中自主运用。4.2 创建其他智能体并设计工作流接下来创建CoderAgent和SummarizerAgent。CoderAgent可以是一个简单的UserAgent或者是一个能自动生成示例代码的DialogAgent。SummarizerAgent则是一个标准的DialogAgent负责润色报告。from agentscope.agents import DialogAgent, UserAgent # CoderAgent - 这里我们用一个能生成示例代码的DialogAgent来模拟 coder_agent DialogAgent( nameCoder, model_config_namemodel_config_name, sys_prompt你是一个Python程序员负责提交需要审查的代码片段。你会根据要求生成或提供一段包含潜在安全风险的Python代码示例。 ) # SummarizerAgent - 负责汇总和美化报告 summarizer_agent DialogAgent( nameReportSummarizer, model_config_namemodel_config_name, sys_prompt你是一个技术文档工程师。你将收到一份来自安全审查员的原始报告可能包含工具输出的原始文本。你的任务是将其提炼、总结生成一份格式优美、重点突出、对开发者友好的最终审查报告。使用Markdown格式包含标题、问题列表、严重性标签和具体的修复代码示例。 )现在我们来设计工作流。这次我们不使用简单的sequentialpipeline而是用更直观的函数调用来手动控制流程这提供了最大的灵活性。def code_review_pipeline(): 代码审查协作流水线 print( 代码审查系统启动 ) # 1. Coder 提交代码 print(\n[阶段1] Coder 提交代码...) # 模拟Coder生成一段有问题的代码 prompt_for_coder 请生成一段包含‘使用pickle反序列化用户输入’这种安全风险的Python代码片段用于审查。 code_submission coder_agent(prompt_for_coder) submitted_code code_submission[content] print(fCoder 提交的代码\npython\n{submitted_code}\n) # 2. Reviewer 进行审查包括工具调用 print(\n[阶段2] Reviewer 进行安全审查...) review_prompt f请对以下Python代码进行全面的安全审查 {‘submitted_code’} 请严格按照你的职责执行先理解代码然后调用安全扫描工具最后结合结果生成审查报告。 review_report reviewer_agent(review_prompt) print(fReviewer 生成的原始报告\n{review_report[content]}\n) # 3. Summarizer 美化报告 print(\n[阶段3] Summarizer 生成最终报告...) summary_prompt f这是安全审查员生成的原始报告 {review_report[‘content’]} 请将其整理成一份给开发者的最终版Markdown格式报告。 final_report summarizer_agent(summary_prompt) print(\n *50) print(【最终代码审查报告】) print(final_report[content]) print(*50) return final_report # 运行流水线 if __name__ __main__: # 确保先执行 agentscope.init(...) final_result code_review_pipeline()在这个流水线中我们清晰地定义了三个阶段每个阶段调用一个智能体并将上一个智能体的输出作为下一个的输入。这种显式控制流在复杂逻辑中非常有用。4.3 实战中的经验与优化点通过构建这个系统我深刻体会到多智能体协作的优势关注点分离和能力复用。ReviewerAgent只需要聚焦于“分析代码和调用工具”SummarizerAgent只关心“格式化和美化”它们各司其职。在实际操作中有几点需要特别注意工具调用的稳定性与错误处理工具函数如run_bandit_security_scan必须非常健壮。网络超时、依赖缺失、异常输入都要考虑周全。ToolAgent在调用工具失败时其行为取决于模型和提示词最好在工具函数内部返回清晰的错误信息以便智能体能将其纳入回复。消息内容的裁剪与上下文管理大语言模型有上下文长度限制。当传递的代码或报告很长时可能会导致后续智能体调用失败。在实际项目中需要考虑对长消息进行摘要、裁剪或分块处理。AgentScope的消息对象可以方便地附加元数据可以用来管理上下文。智能体状态的持久化上述例子中智能体对话是无状态的每轮都是新的。对于多轮对话需要利用AgentScope提供的记忆Memory功能让智能体记住之前的交互历史。这通常通过初始化智能体时传入一个memory参数来实现。异步与并发执行在我们的流水线中智能体是按顺序执行的。如果ReviewerAgent的工具调用很耗时会阻塞整个流程。对于性能要求高的场景可以考虑使用AgentScope的异步接口或并发执行能力让多个智能体同时工作。5. 深入原理AgentScope如何管理智能体与消息流当我们调用agent(message)或者运行pipeline时框架底层做了大量工作来确保一切井然有序。理解这些原理有助于我们更好地调试和构建更复杂的系统。5.1 消息的生命周期与处理流程一条消息在AgentScope中的旅程大致如下消息创建通常由UserAgent或上一个智能体的reply方法创建。它是一个符合特定模式的Python字典。消息预处理在智能体的reply方法内部消息可能会被格式化以符合背后LLM的输入要求。例如将对话历史、系统提示词和当前消息拼接成ChatML等标准格式。模型调用对于DialogAgent或ToolAgent框架会调用配置的模型API如OpenAI API。这里处理了重试、超时、API密钥轮换等底层细节。响应解析与后处理收到模型响应后框架会解析出文本内容并将其封装成新的Message对象。对于ToolAgent如果响应中包含工具调用请求框架会拦截这个请求转而执行对应的工具函数并将工具结果再次发送给模型形成“模型-工具”的多轮对话直到模型决定返回最终答案给用户。消息传递在流水线中这个新生成的消息会被传递给下一个智能体开启新一轮生命周期。5.2 智能体的执行模型与Hook机制每个智能体核心的reply方法可以被看作一个黑盒输入消息输出消息。但框架提供了“钩子”Hooks机制允许我们在执行前后插入自定义逻辑。pre_process在智能体处理消息前执行。可以用来过滤消息、记录日志、修改消息内容。post_process在智能体生成回复后执行。可以用来验证回复格式、将回复存入数据库、触发其他事件。例如我们可以给所有智能体添加一个记录执行时间的钩子import time from agentscope.agents import AgentBase from agentscope.message import Msg def add_timing_hook(agent: AgentBase): original_reply agent.reply def wrapped_reply(msg: Msg): start_time time.time() response original_reply(msg) end_time time.time() print(f[Hook] Agent {agent.name} 处理耗时: {end_time - start_time:.2f}秒) return response agent.reply wrapped_reply return agent # 使用钩子 reviewer_agent add_timing_hook(reviewer_agent)这种机制非常强大可以实现监控、审计、缓存、限流等跨智能体的通用功能。5.3 配置管理与可观测性AgentScope的init函数是配置的入口。除了模型配置还可以在这里设置项目配置如项目名称用于日志和追踪。日志配置控制日志级别和输出格式对于调试复杂交互至关重要。分布式设置为将来智能体以服务形式部署预留了配置项。良好的日志是调试多智能体系统的生命线。建议在开发时将日志级别设为DEBUG可以清晰地看到消息的流动、模型的请求与响应、工具调用的触发等信息。import agentscope import logging agentscope.init( projectMyCodeReviewApp, model_configs[...], # 你的模型配置 logging_levellogging.DEBUG, # 开启DEBUG日志 )6. 生产环境考量部署、监控与最佳实践将多智能体应用从实验脚本变为可靠的服务还需要考虑很多工程问题。AgentScope提供了一些基础支持但更多的需要开发者自己设计。6.1 智能体服务化部署AgentService是将智能体包装成Web服务的利器。你可以将一个复杂的ToolAgent暴露为HTTP API供其他系统调用。from agentscope.service import AgentService from fastapi import FastAPI import uvicorn # 创建我们之前定义的SecurityReviewer智能体 reviewer ... # 创建ToolAgent的代码 # 将其包装成服务 service AgentService(agentreviewer, namecode-review-service) # 使用FastAPI创建应用AgentService基于FastAPI app FastAPI() app.include_router(service.router) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)启动后你就可以通过POST /service/code-review-service/call接口来调用这个智能体了。这对于构建微服务架构的多智能体系统非常有用。6.2 性能优化与成本控制多智能体应用的核心成本是LLM API调用。以下是一些控制成本和提升性能的策略缓存对频繁出现的、结果确定的查询进行缓存。可以在pre_process钩子中实现一个简单的内存缓存如functools.lru_cache或集成Redis。注意缓存键需要包含消息内容和智能体配置。模型分级并非所有任务都需要最强大、最昂贵的模型。可以用小模型如GPT-3.5-turbo处理简单的路由、分类任务用大模型如GPT-4处理核心的推理、创作任务。在AgentScope中只需为不同的智能体配置不同的model_config_name即可轻松实现。异步与非阻塞调用如果流水线中的智能体间没有严格的先后依赖可以考虑使用asyncio并发执行。AgentScope的部分接口支持异步可以显著减少总等待时间。设置合理的超时与重试在模型配置中可以为API调用设置超时和重试策略防止因单个请求挂起导致整个系统停滞。6.3 监控与可观测性对于生产系统必须监控其健康度和效果。基础指标每个智能体的调用次数、平均响应时间、失败率。可以通过钩子函数收集这些数据并推送到监控系统如Prometheus。业务指标根据应用场景定义。例如在代码审查系统中可以监控“平均每个审查发现的问题数”、“高严重性问题占比”、“审查报告生成时间”等。链路追踪为每一轮用户会话或每一个任务分配一个唯一的trace_id并让这个ID在所有智能体的消息中传递。这样可以在日志中完整追溯一个请求在所有智能体间的流转路径对于排查问题至关重要。效果评估多智能体系统的最终输出质量如何评估可能需要设计一套人工或自动的评估流程定期抽样检查并根据反馈优化智能体的提示词或工作流。6.4 团队协作与代码组织当项目变大时良好的代码结构很重要。我建议采用如下组织方式my_agentscope_project/ ├── agents/ # 存放自定义智能体类 │ ├── __init__.py │ ├── expert_reviewer.py │ └── creative_writer.py ├── tools/ # 存放所有工具函数 │ ├── __init__.py │ ├── code_analysis.py │ └── web_search.py ├── pipelines/ # 存放定义好的工作流 │ ├── __init__.py │ └── code_review_workflow.py ├── configs/ # 配置文件模型配置、API密钥等 │ └── model_configs.yaml ├── services/ # AgentService定义 │ └── review_service.py ├── utils/ # 工具函数如钩子、监控 │ └── hooks.py └── main.py # 应用入口通过模块化的组织不同职责的代码清晰分离便于团队协作和维护。经过这几个章节的探讨我们从概念、上手、进阶到原理和生产实践完整地梳理了AgentScope这个多智能体框架。它的出现确实为构建复杂的AI协作应用提供了一个强有力的脚手架。当然框架本身还在快速发展中社区和生态是它未来能否成功的关键。对于开发者而言现在正是深入探索、贡献想法和代码的好时机。无论是用它来快速验证一个多AI协作的创意还是构建严肃的企业级应用AgentScope都提供了一个值得投入的起点。
AgentScope：开源多智能体框架，降低大模型协作应用开发门槛
1. 项目初探AgentScope一个面向多智能体应用的开源框架最近在折腾大模型应用开发特别是涉及到多个AI智能体Agent协同工作的场景时我遇到了不少麻烦。比如如何让不同的智能体比如一个负责写代码一个负责审核代码顺畅地对话如何管理它们之间的消息流如何优雅地处理不同模型API的调用差异这些问题在单个智能体的场景下还好一旦涉及到多智能体协作复杂度就指数级上升。就在我为此头疼的时候一个名为AgentScope的开源项目进入了我的视野。AgentScope项目地址agentscope-ai/agentscope是一个由智谱AI开源的、专门为构建多智能体应用而设计的框架。简单来说它不是一个单一的智能体工具而是一个“舞台导演”或者“编排系统”让你能轻松地定义多个拥有不同角色和能力的智能体并设计它们之间的交互规则从而完成复杂的任务。它的出现正是为了解决像我这样希望利用大模型能力构建更复杂、更智能应用却又苦于底层通信、状态管理和工具集成的开发者们的痛点。这个框架的核心价值在于“降低多智能体应用开发的门槛”。它提供了一套高层的、声明式的API让你可以像搭积木一样组合智能体而无需关心底层的网络通信、并发控制、消息序列化等繁琐细节。无论是想构建一个模拟辩论的AI系统还是一个自动化代码审查流水线或者是一个复杂的游戏NPC交互环境AgentScope都试图提供一个统一、高效的开发范式。接下来我将深入拆解它的架构、核心概念以及如何上手并分享一些在实际使用中的心得体会。2. AgentScope的核心架构与设计哲学要理解一个框架首先要看它的设计思路。AgentScope的架构清晰地反映了其“以智能体为中心以消息为纽带”的设计哲学。整个框架可以粗略地分为三层智能体层、服务层和基础设施层。2.1 智能体Agent能力的抽象与封装在AgentScope中Agent是最核心的抽象。每一个智能体都是一个独立的、具有特定角色、记忆和能力的实体。框架内置了几种常见的智能体类型这也是我们上手时最常打交道的部分DialogAgent对话智能体这是最基础的智能体它封装了一个大语言模型LLM的调用。你给它一个系统提示词System Prompt来定义其角色如“你是一个严谨的代码审查专家”它就能根据接收到的消息调用背后的LLM生成回复。它主要负责“思考”和“表达”。UserAgent用户智能体这个智能体代表真实用户。在交互式应用中它负责接收用户在终端或Web界面上的输入并将其转化为框架内部的标准消息格式。在自动化流程中它也可以被编程为按特定规则生成输入。ToolAgent工具智能体这是让智能体“动手”能力的关键。一个ToolAgent内部封装了一个或多个可执行的工具函数例如调用搜索引擎API、执行一段Python代码、查询数据库。当它收到一个请求时会尝试理解请求内容选择并调用合适的工具然后将工具执行的结果包装成消息回复出去。这极大地扩展了智能体的能力边界使其不再局限于文本生成。ActorAgent角色扮演智能体这是一个更灵活的抽象允许你自定义智能体的行为逻辑。你可以完全控制其reply函数实现任何你想要的反应模式比如基于规则的响应、调用其他服务等。这些智能体类型并不是孤立的你可以通过继承来创建复合型智能体。例如你可以创建一个既擅长对话又能使用特定工具的智能体。2.2 消息Message智能体间通信的通用语言智能体之间如何交流靠的是标准化的Message对象。AgentScope定义了一套清晰的消息格式每条消息通常包含name发送者、content内容等字段。这种设计的好处是解耦智能体无需知道消息来自哪个具体的智能体实例它只处理符合格式的消息。可追溯所有交互历史都以消息序列的形式保存便于调试、分析和重现。灵活性可以轻松地在消息流水线中插入过滤器、转换器或记录器。2.3 服务与编排让智能体动起来有了智能体和消息还需要一个“导演”来指挥它们如何互动。这就是AgentScope的服务层和编排能力。AgentService你可以将智能体包装成一个独立的服务例如一个HTTP服务。这使得智能体能力可以被远程调用方便构建分布式系统或微服务架构。流水线Pipeline与工作流这是实现复杂交互的核心。框架提供了直观的方式来定义智能体之间的交互顺序。最常见的是使用sequentialpipeline它让消息按顺序从一个智能体传递到下一个。对于更复杂的条件分支或循环交互你可以利用Python的控制流如if-else,for循环结合框架的API来灵活编排。2.4 设计哲学易用性与灵活性的平衡从上述架构可以看出AgentScope在努力做一件事把复杂留给自己把简单留给开发者。它通过高层次的抽象隐藏了多智能体系统中诸如并发安全、消息路由、错误处理等棘手问题。开发者只需要关注两件事1) 定义智能体的角色和能力2) 设计它们之间的交互逻辑。这种“约定大于配置”的思路能显著提升开发效率。3. 从零开始搭建你的第一个多智能体对话系统理论讲得再多不如动手一试。让我们来构建一个最简单的场景一个UserAgent模拟用户向一个DialogAgent扮演助手提问并得到回答。这个例子虽然简单但涵盖了最核心的流程。3.1 环境准备与安装首先确保你的Python环境在3.8及以上。然后通过pip安装AgentScopepip install agentscope安装过程会同时安装一些核心依赖。如果你想使用特定的模型比如OpenAI的GPT系列或智谱的GLM还需要安装对应的SDKpip install openai # 如需使用OpenAI # 或者 pip install zhipuai # 如需使用智谱GLM3.2 初始化模型与创建智能体AgentScope支持多种模型后端在使用前需要进行全局初始化。这里以使用OpenAI API为例你需要准备自己的API Keyimport agentscope from agentscope.agents import DialogAgent, UserAgent from agentscope.pipelines import sequentialpipeline # 第一步初始化模型配置 agentscope.init( model_configs[ { model_type: openai, # 指定模型类型 config_name: gpt-4, # 配置名称可自定义 model_name: gpt-4, # 实际调用的模型名 api_key: your-openai-api-key-here, # 替换为你的Key organization: your-org, # 可选组织ID } ] )初始化完成后我们就可以创建智能体了。创建时需要为它们指定一个名字和所使用的模型配置。# 第二步创建智能体 assistant_agent DialogAgent( nameAssistant, sys_prompt你是一个乐于助人且知识渊博的AI助手。请用中文回答用户的问题。, model_config_namegpt-4, # 使用上面初始化的配置 ) user_agent UserAgent(nameUser)DialogAgent的sys_prompt参数至关重要它定义了智能体的“人设”和行为准则。好的提示词能极大提升智能体的表现。model_config_name必须与init中定义的config_name一致。3.3 构建交互流水线并运行智能体创建好了现在需要定义它们如何交互。我们使用最简单的顺序流水线# 第三步定义并运行流水线 def main(): # sequentialpipeline 会按顺序执行传入的函数或智能体 # 这里先运行user_agent获取用户输入再将结果传给assistant_agent得到回复 pipeline sequentialpipeline([user_agent, assistant_agent]) print(多智能体对话系统已启动。输入‘退出’或‘quit’结束对话。) while True: # 运行一次流水线触发一轮对话 result pipeline() # 结果是一个消息列表这里我们简单打印最后一条消息助手的回复 if result and len(result) 0: last_msg result[-1] print(f\n[Assistant]: {last_msg[content]}\n) # 一个简单的退出检查实际中UserAgent可能会处理 # 这里为了演示我们检查用户输入 # 注意这是一个简化的逻辑真实场景需要更健壮的处理 user_input_msg result[0] if result else None if user_input_msg and user_input_msg[content].strip().lower() in [退出, quit]: print(对话结束。) break if __name__ __main__: main()运行这段代码你会在终端看到一个交互式对话界面。你输入问题UserAgent将其包装成消息sequentialpipeline驱动消息先经过UserAgent获取输入再传递给Assistant生成回复最后将结果返回。注意这个示例中的退出逻辑比较粗糙。在实际应用中更优雅的做法是在UserAgent内部或通过一个专门的ControllerAgent来管理对话状态和退出条件。这里为了突出核心流程做了简化。3.4 第一个项目的总结与常见坑点通过这个最小化的例子你应该已经感受到了AgentScope的便捷定义Agent - 组合Pipeline - 运行。三步就完成了一个可交互的多智能体系统雏形。在这个过程中最容易出错的点往往是模型配置错误init中的config_name和创建Agent时的model_config_name必须完全一致包括大小写。建议用一个常量来管理这些配置名。API Key等敏感信息管理切勿将API Key硬编码在代码中提交到版本库。务必使用环境变量或配置文件来管理。import os api_key os.getenv(OPENAI_API_KEY)提示词Prompt设计sys_prompt的质量直接决定智能体的表现。对于中文场景明确要求“用中文回答”很重要。更复杂的角色需要更细致、包含示例Few-shot的提示词。4. 进阶实战构建一个代码审查智能体协作系统单一的问答应答显然无法体现多智能体的威力。让我们设计一个更贴近实际开发的场景一个自动化的代码审查流水线。这个系统包含三个智能体CoderAgent程序员智能体负责提交待审查的代码片段。ReviewerAgent审查员智能体一个ToolAgent它不仅能理解代码还能调用一个静态代码分析工具例如pylint或bandit来检查代码中的潜在问题。SummarizerAgent总结员智能体负责汇总审查员的反馈并以清晰、友好的格式生成最终审查报告给程序员。这个例子将展示如何集成工具、如何处理更复杂的消息流。4.1 设计工具函数并创建ToolAgent首先我们需要为ReviewerAgent准备工具。这里我们模拟一个调用bandit一个Python安全漏洞扫描工具的工具函数。import subprocess import tempfile import agentscope from agentscope.agents import ToolAgent from agentscope.tools import tool # 定义一个工具函数使用tool装饰器注册 tool def run_bandit_security_scan(code_snippet: str) - str: 使用bandit对提供的Python代码片段进行安全扫描。 Args: code_snippet: 需要扫描的Python代码字符串。 Returns: 扫描结果的文本报告。如果扫描过程出错返回错误信息。 # 将代码片段写入临时文件 with tempfile.NamedTemporaryFile(modew, suffix.py, deleteFalse) as f: f.write(code_snippet) temp_file_path f.name try: # 调用bandit命令进行扫描 # -q: 安静模式减少冗余输出 # -f: 指定输出格式为text result subprocess.run( [bandit, -q, -f, text, temp_file_path], capture_outputTrue, textTrue, timeout30 # 设置超时防止长时间运行 ) output result.stdout if result.returncode ! 0 and not output: output result.stderr or Bandit扫描过程出现未知错误。 return output except subprocess.TimeoutExpired: return 安全扫描超时代码可能过于复杂或存在死循环风险。 except FileNotFoundError: return 错误未找到bandit命令。请确保已通过‘pip install bandit’安装。 except Exception as e: return f执行安全扫描时发生异常{str(e)} finally: # 清理临时文件 import os os.unlink(temp_file_path)现在用这个工具函数来创建我们的ReviewerAgent# 假设模型配置已通过 agentscope.init() 初始化例如使用GPT-4 model_config_name gpt-4 # 与init中的config_name对应 reviewer_agent ToolAgent( nameSecurityReviewer, model_config_namemodel_config_name, tools[run_bandit_security_scan], # 传入工具函数列表 sys_prompt你是一个专业的代码安全审查员。你的任务是 1. 分析用户提供的Python代码理解其功能。 2. 自动调用安全扫描工具bandit对代码进行漏洞检测。 3. 结合工具扫描结果和你自身的代码安全知识生成一份全面的审查报告。 报告需包含 - 代码功能简述。 - 工具扫描发现的具体问题列出每个问题的严重等级、位置和描述。 - 基于最佳实践给出的修改建议。 - 如果扫描未发现问题也请说明。 请用清晰、有条理的中文撰写报告。 )ToolAgent的强大之处在于它能够根据用户的请求和工具的描述自动决定何时以及如何调用工具。你只需要把工具函数给它它会在对话中自主运用。4.2 创建其他智能体并设计工作流接下来创建CoderAgent和SummarizerAgent。CoderAgent可以是一个简单的UserAgent或者是一个能自动生成示例代码的DialogAgent。SummarizerAgent则是一个标准的DialogAgent负责润色报告。from agentscope.agents import DialogAgent, UserAgent # CoderAgent - 这里我们用一个能生成示例代码的DialogAgent来模拟 coder_agent DialogAgent( nameCoder, model_config_namemodel_config_name, sys_prompt你是一个Python程序员负责提交需要审查的代码片段。你会根据要求生成或提供一段包含潜在安全风险的Python代码示例。 ) # SummarizerAgent - 负责汇总和美化报告 summarizer_agent DialogAgent( nameReportSummarizer, model_config_namemodel_config_name, sys_prompt你是一个技术文档工程师。你将收到一份来自安全审查员的原始报告可能包含工具输出的原始文本。你的任务是将其提炼、总结生成一份格式优美、重点突出、对开发者友好的最终审查报告。使用Markdown格式包含标题、问题列表、严重性标签和具体的修复代码示例。 )现在我们来设计工作流。这次我们不使用简单的sequentialpipeline而是用更直观的函数调用来手动控制流程这提供了最大的灵活性。def code_review_pipeline(): 代码审查协作流水线 print( 代码审查系统启动 ) # 1. Coder 提交代码 print(\n[阶段1] Coder 提交代码...) # 模拟Coder生成一段有问题的代码 prompt_for_coder 请生成一段包含‘使用pickle反序列化用户输入’这种安全风险的Python代码片段用于审查。 code_submission coder_agent(prompt_for_coder) submitted_code code_submission[content] print(fCoder 提交的代码\npython\n{submitted_code}\n) # 2. Reviewer 进行审查包括工具调用 print(\n[阶段2] Reviewer 进行安全审查...) review_prompt f请对以下Python代码进行全面的安全审查 {‘submitted_code’} 请严格按照你的职责执行先理解代码然后调用安全扫描工具最后结合结果生成审查报告。 review_report reviewer_agent(review_prompt) print(fReviewer 生成的原始报告\n{review_report[content]}\n) # 3. Summarizer 美化报告 print(\n[阶段3] Summarizer 生成最终报告...) summary_prompt f这是安全审查员生成的原始报告 {review_report[‘content’]} 请将其整理成一份给开发者的最终版Markdown格式报告。 final_report summarizer_agent(summary_prompt) print(\n *50) print(【最终代码审查报告】) print(final_report[content]) print(*50) return final_report # 运行流水线 if __name__ __main__: # 确保先执行 agentscope.init(...) final_result code_review_pipeline()在这个流水线中我们清晰地定义了三个阶段每个阶段调用一个智能体并将上一个智能体的输出作为下一个的输入。这种显式控制流在复杂逻辑中非常有用。4.3 实战中的经验与优化点通过构建这个系统我深刻体会到多智能体协作的优势关注点分离和能力复用。ReviewerAgent只需要聚焦于“分析代码和调用工具”SummarizerAgent只关心“格式化和美化”它们各司其职。在实际操作中有几点需要特别注意工具调用的稳定性与错误处理工具函数如run_bandit_security_scan必须非常健壮。网络超时、依赖缺失、异常输入都要考虑周全。ToolAgent在调用工具失败时其行为取决于模型和提示词最好在工具函数内部返回清晰的错误信息以便智能体能将其纳入回复。消息内容的裁剪与上下文管理大语言模型有上下文长度限制。当传递的代码或报告很长时可能会导致后续智能体调用失败。在实际项目中需要考虑对长消息进行摘要、裁剪或分块处理。AgentScope的消息对象可以方便地附加元数据可以用来管理上下文。智能体状态的持久化上述例子中智能体对话是无状态的每轮都是新的。对于多轮对话需要利用AgentScope提供的记忆Memory功能让智能体记住之前的交互历史。这通常通过初始化智能体时传入一个memory参数来实现。异步与并发执行在我们的流水线中智能体是按顺序执行的。如果ReviewerAgent的工具调用很耗时会阻塞整个流程。对于性能要求高的场景可以考虑使用AgentScope的异步接口或并发执行能力让多个智能体同时工作。5. 深入原理AgentScope如何管理智能体与消息流当我们调用agent(message)或者运行pipeline时框架底层做了大量工作来确保一切井然有序。理解这些原理有助于我们更好地调试和构建更复杂的系统。5.1 消息的生命周期与处理流程一条消息在AgentScope中的旅程大致如下消息创建通常由UserAgent或上一个智能体的reply方法创建。它是一个符合特定模式的Python字典。消息预处理在智能体的reply方法内部消息可能会被格式化以符合背后LLM的输入要求。例如将对话历史、系统提示词和当前消息拼接成ChatML等标准格式。模型调用对于DialogAgent或ToolAgent框架会调用配置的模型API如OpenAI API。这里处理了重试、超时、API密钥轮换等底层细节。响应解析与后处理收到模型响应后框架会解析出文本内容并将其封装成新的Message对象。对于ToolAgent如果响应中包含工具调用请求框架会拦截这个请求转而执行对应的工具函数并将工具结果再次发送给模型形成“模型-工具”的多轮对话直到模型决定返回最终答案给用户。消息传递在流水线中这个新生成的消息会被传递给下一个智能体开启新一轮生命周期。5.2 智能体的执行模型与Hook机制每个智能体核心的reply方法可以被看作一个黑盒输入消息输出消息。但框架提供了“钩子”Hooks机制允许我们在执行前后插入自定义逻辑。pre_process在智能体处理消息前执行。可以用来过滤消息、记录日志、修改消息内容。post_process在智能体生成回复后执行。可以用来验证回复格式、将回复存入数据库、触发其他事件。例如我们可以给所有智能体添加一个记录执行时间的钩子import time from agentscope.agents import AgentBase from agentscope.message import Msg def add_timing_hook(agent: AgentBase): original_reply agent.reply def wrapped_reply(msg: Msg): start_time time.time() response original_reply(msg) end_time time.time() print(f[Hook] Agent {agent.name} 处理耗时: {end_time - start_time:.2f}秒) return response agent.reply wrapped_reply return agent # 使用钩子 reviewer_agent add_timing_hook(reviewer_agent)这种机制非常强大可以实现监控、审计、缓存、限流等跨智能体的通用功能。5.3 配置管理与可观测性AgentScope的init函数是配置的入口。除了模型配置还可以在这里设置项目配置如项目名称用于日志和追踪。日志配置控制日志级别和输出格式对于调试复杂交互至关重要。分布式设置为将来智能体以服务形式部署预留了配置项。良好的日志是调试多智能体系统的生命线。建议在开发时将日志级别设为DEBUG可以清晰地看到消息的流动、模型的请求与响应、工具调用的触发等信息。import agentscope import logging agentscope.init( projectMyCodeReviewApp, model_configs[...], # 你的模型配置 logging_levellogging.DEBUG, # 开启DEBUG日志 )6. 生产环境考量部署、监控与最佳实践将多智能体应用从实验脚本变为可靠的服务还需要考虑很多工程问题。AgentScope提供了一些基础支持但更多的需要开发者自己设计。6.1 智能体服务化部署AgentService是将智能体包装成Web服务的利器。你可以将一个复杂的ToolAgent暴露为HTTP API供其他系统调用。from agentscope.service import AgentService from fastapi import FastAPI import uvicorn # 创建我们之前定义的SecurityReviewer智能体 reviewer ... # 创建ToolAgent的代码 # 将其包装成服务 service AgentService(agentreviewer, namecode-review-service) # 使用FastAPI创建应用AgentService基于FastAPI app FastAPI() app.include_router(service.router) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8000)启动后你就可以通过POST /service/code-review-service/call接口来调用这个智能体了。这对于构建微服务架构的多智能体系统非常有用。6.2 性能优化与成本控制多智能体应用的核心成本是LLM API调用。以下是一些控制成本和提升性能的策略缓存对频繁出现的、结果确定的查询进行缓存。可以在pre_process钩子中实现一个简单的内存缓存如functools.lru_cache或集成Redis。注意缓存键需要包含消息内容和智能体配置。模型分级并非所有任务都需要最强大、最昂贵的模型。可以用小模型如GPT-3.5-turbo处理简单的路由、分类任务用大模型如GPT-4处理核心的推理、创作任务。在AgentScope中只需为不同的智能体配置不同的model_config_name即可轻松实现。异步与非阻塞调用如果流水线中的智能体间没有严格的先后依赖可以考虑使用asyncio并发执行。AgentScope的部分接口支持异步可以显著减少总等待时间。设置合理的超时与重试在模型配置中可以为API调用设置超时和重试策略防止因单个请求挂起导致整个系统停滞。6.3 监控与可观测性对于生产系统必须监控其健康度和效果。基础指标每个智能体的调用次数、平均响应时间、失败率。可以通过钩子函数收集这些数据并推送到监控系统如Prometheus。业务指标根据应用场景定义。例如在代码审查系统中可以监控“平均每个审查发现的问题数”、“高严重性问题占比”、“审查报告生成时间”等。链路追踪为每一轮用户会话或每一个任务分配一个唯一的trace_id并让这个ID在所有智能体的消息中传递。这样可以在日志中完整追溯一个请求在所有智能体间的流转路径对于排查问题至关重要。效果评估多智能体系统的最终输出质量如何评估可能需要设计一套人工或自动的评估流程定期抽样检查并根据反馈优化智能体的提示词或工作流。6.4 团队协作与代码组织当项目变大时良好的代码结构很重要。我建议采用如下组织方式my_agentscope_project/ ├── agents/ # 存放自定义智能体类 │ ├── __init__.py │ ├── expert_reviewer.py │ └── creative_writer.py ├── tools/ # 存放所有工具函数 │ ├── __init__.py │ ├── code_analysis.py │ └── web_search.py ├── pipelines/ # 存放定义好的工作流 │ ├── __init__.py │ └── code_review_workflow.py ├── configs/ # 配置文件模型配置、API密钥等 │ └── model_configs.yaml ├── services/ # AgentService定义 │ └── review_service.py ├── utils/ # 工具函数如钩子、监控 │ └── hooks.py └── main.py # 应用入口通过模块化的组织不同职责的代码清晰分离便于团队协作和维护。经过这几个章节的探讨我们从概念、上手、进阶到原理和生产实践完整地梳理了AgentScope这个多智能体框架。它的出现确实为构建复杂的AI协作应用提供了一个强有力的脚手架。当然框架本身还在快速发展中社区和生态是它未来能否成功的关键。对于开发者而言现在正是深入探索、贡献想法和代码的好时机。无论是用它来快速验证一个多AI协作的创意还是构建严肃的企业级应用AgentScope都提供了一个值得投入的起点。