1. 这篇文章真正要解决的问题如果你正在寻找一个能快速构建、测试和部署智能体Agent的框架并且对市面上那些要么过于复杂、要么功能简陋的解决方案感到失望那么agency-agents这个项目值得你花十分钟了解一下。它不是一个试图解决所有问题的“大而全”平台而是一个定位精准的“脚手架”和“工具箱”。这篇文章要解决的正是开发者在构建AI Agent应用时面临的几个核心痛点上手门槛高、调试过程黑盒、以及难以将原型快速转化为可部署的服务。很多框架要么要求你从零开始理解复杂的异步通信和状态管理要么将你锁定在特定的云服务或模型提供商中。agency-agents的核心价值在于它通过提供一套清晰、模块化的基础组件让你能专注于Agent的逻辑本身而不是底层的基础设施。它解决了“从想法到可运行原型”的效率问题特别适合需要快速验证多个Agent协作场景的开发者、研究者或是希望将AI能力以服务形式嵌入现有系统的工程师。读完本文你将能清晰地理解agency-agents的设计哲学掌握其核心概念并能够独立完成一个包含多个协作Agent的示例项目的搭建、运行和调试。更重要的是你会知道它适合什么场景不适合什么场景以及在实际工程化中需要注意哪些“坑”。2. 基础概念与核心原理在深入代码之前我们需要统一几个关键概念这是理解agency-agents设计思路的基础。Agent智能体在agency-agents的语境下Agent 是一个具有特定能力、可以接收消息、进行处理并返回响应的独立实体。它可以是一个调用大语言模型LLM的聊天助手也可以是一个执行代码的代码解释器或者一个查询数据库的工具。每个 Agent 都专注于一个明确的领域。Skill技能这是 Agent 能力的具象化。一个 Agent 可以拥有多个 Skill。例如一个“数据分析师”Agent 可能拥有“执行SQL查询”、“绘制图表”、“生成报告摘要”等多个 Skill。在实现上一个 Skill 通常对应一个或多个可以被调用的函数方法。Space空间这是整个框架最核心的抽象。你可以把 Space 理解为一个虚拟的“协作房间”或“消息总线”。所有的 Agent 都“生活”在同一个 Space 中。当一个 Agent 需要与另一个 Agent 通信时它不需要知道对方的具体地址或端口只需要向 Space 发送一条消息指定目标 Agent 的名称即可。Space 负责消息的路由、传递和生命周期管理。这种设计极大地降低了Agent间耦合度。Message消息Agent 之间通信的基本单位。一条消息通常包含发送者、接收者、消息内容以及可能的元数据如消息类型、优先级等。工作原理流程初始化创建一个 Space 实例并向其中添加多个配置好的 Agent。触发外部请求如HTTP API或某个Agent主动发起向Space中的某个目标Agent发送一条Message。路由Space 接收到消息根据消息头中的目标Agent标识将其放入对应Agent的消息队列。处理目标Agent从其消息队列中取出消息根据消息内容识别需要调用的Skill并执行相应的处理逻辑可能涉及调用LLM、执行代码、访问API等。响应处理完成后Agent生成响应消息。这个响应可以直接返回给最初的发送者也可以作为新的消息发送给Space内的另一个Agent从而形成工作流。交付Space 将最终响应交付给最初的请求方。这种基于消息传递的架构使得系统非常灵活。Agent之间是松耦合的你可以随时向Space中添加或移除Agent而不会影响其他Agent的正常工作。这也为模拟复杂的多智能体协作场景如辩论、评审、接力任务提供了天然的基础。3. 环境准备与前置条件开始实践前请确保你的开发环境满足以下要求。我们将以一个相对通用的Python环境为例进行说明。操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)均可。本文命令以Linux/macOS的bash shell为例Windows用户可在PowerShell或WSL中执行类似操作。Python版本agency-agents通常要求 Python 3.8 或更高版本。推荐使用 Python 3.9 以获得更好的兼容性。# 检查Python版本 python3 --version # 或 python --version依赖管理工具我们使用pip进行包管理。强烈建议使用虚拟环境venv来隔离项目依赖避免污染系统Python环境。# 创建并激活虚拟环境 (Linux/macOS) python3 -m venv agency-env source agency-env/bin/activate # 创建并激活虚拟环境 (Windows PowerShell) python -m venv agency-env .\agency-env\Scripts\Activate.ps1 # 如果遇到执行策略限制请先以管理员身份运行Set-ExecutionPolicy RemoteSigned -Scope CurrentUser激活虚拟环境后命令行提示符前通常会显示环境名称(agency-env)。安装核心框架使用pip从GitHub直接安装agency-agents。由于它可能处于活跃开发中我们安装主分支的最新版本。pip install githttps://github.com/msitarzewski/agency-agents.git可选但重要的依赖为了让我们后续的示例能够真正运行起来例如调用LLM我们还需要安装openai库。同时pydantic和httpx也是框架常用的依赖。pip install openai pydantic httpx获取API密钥如果示例中涉及调用OpenAI的模型你需要准备一个有效的 OpenAI API Key。请妥善保管不要将其直接硬编码在代码中。环境准备就绪后我们就可以开始构建第一个多Agent系统了。4. 核心流程拆解构建一个翻译校对系统让我们通过一个具体的场景来学习agency-agents构建一个“翻译-校对”双Agent系统。用户输入一句中文系统先将其翻译成英文然后由另一个Agent对英文翻译进行语法和流畅度校对。4.1 步骤一定义Agent和Skill首先我们需要定义两个AgentTranslator和Proofreader。每个Agent都是一个Python类继承自框架提供的基类通常是Agent。我们将在类中定义它们各自的Skill。# 文件translator_proofreader.py from agency_agents import Agent, Space from pydantic import BaseModel import openai import os # 设置OpenAI API Key (实践中应从环境变量读取) openai.api_key os.getenv(OPENAI_API_KEY) class TranslationRequest(BaseModel): 翻译请求的数据模型 text: str target_language: str English class Translator(Agent): 翻译官Agent name translator # Agent在Space中的唯一标识 def __init__(self): super().__init__() # 可以在这里初始化一些资源如模型客户端 async def translate(self, request: TranslationRequest) - str: 核心Skill将文本翻译成目标语言 print(f[Translator] 收到翻译请求: {request.text} - {request.target_language}) # 调用OpenAI API进行翻译 (简化示例) # 注意这是一个模拟调用实际应用中需要处理错误和异步 prompt f请将以下中文翻译成{request.target_language}{request.text} # 在实际项目中这里应使用异步客户端如 await openai.AsyncClient().chat.completions.create(...) # 为简化示例我们假设一个同步调用 try: # 模拟API调用返回 # response await openai_async_client.chat.completions.create(...) translated_text f[模拟翻译结果] {request.text} in {request.target_language} return translated_text except Exception as e: return f翻译失败: {str(e)} class Proofreader(Agent): 校对员Agent name proofreader async def check_grammar(self, text: str) - str: 核心Skill检查文本的语法和流畅度 print(f[Proofreader] 收到校对文本: {text}) # 模拟调用LLM进行校对 prompt f请检查以下英文句子的语法和流畅度并给出修改建议{text} # 模拟API调用 feedback f[模拟校对反馈] 句子结构良好但建议将‘simulated’改为‘demonstrated’以更正式。 return feedback关键点解析继承与命名每个Agent类必须继承Agent并设置一个唯一的name。这个name是Space内寻址的关键。Skill定义类中的异步方法如translate,check_grammar会自动被框架识别为该Agent的Skill。方法参数通常使用Pydantic模型来确保类型安全。异步支持框架基于异步async/await所以Skill方法通常是异步的以便高效处理I/O操作如网络请求。4.2 步骤二创建Space并注册Agent定义了Agent之后我们需要创建一个Space并将这些Agent的实例“添加”进去。# 接上文 translator_proofreader.py async def main(): # 1. 创建协作空间 space Space() # 2. 实例化Agent translator Translator() proofreader Proofreader() # 3. 将Agent添加到Space中 await space.add(translator) await space.add(proofreader) print(fSpace已启动包含Agent: {[agent.name for agent in space.agents.values()]}) # 4. 模拟一个用户请求先翻译后校对 user_input 今天天气真好我们一起去公园散步吧。 # 4.1 向translator发送翻译请求 translation_request TranslationRequest(textuser_input, target_languageEnglish) # 使用 run 方法同步等待一个异步Skill的执行结果 translated_result await space.run(translator, translate, translation_request.model_dump()) print(f翻译结果: {translated_result}) # 4.2 将翻译结果发送给proofreader进行校对 if isinstance(translated_result, str) and not translated_result.startswith(翻译失败): proofread_result await space.run(proofreader, check_grammar, translated_result) print(f校对反馈: {proofread_result}) else: print(翻译步骤出错跳过校对。) # 5. 关闭Space (清理资源) await space.close() if __name__ __main__: import asyncio asyncio.run(main())关键点解析Space创建Space()是核心容器。添加Agentawait space.add(agent)是注册Agent的标准方式。添加后Agent即可通过Space被寻址。调用Skillawait space.run(agent_name, skill_name, input_data)是调用Agent Skill的核心API。input_data需要是一个字典与Skill方法的Pydantic模型对应。异步执行整个流程必须在异步上下文asyncio.run中运行。4.3 步骤三运行与观察将以上两部分代码合并到一个文件translator_proofreader.py中并在终端运行。# 确保在虚拟环境中且已安装依赖 python translator_proofreader.py预期输出Space已启动包含Agent: [translator, proofreader] [Translator] 收到翻译请求: 今天天气真好我们一起去公园散步吧。 - English 翻译结果: [模拟翻译结果] 今天天气真好我们一起去公园散步吧。 in English [Proofreader] 收到校对文本: [模拟翻译结果] 今天天气真好我们一起去公园散步吧。 in English 校对反馈: [模拟校对反馈] 句子结构良好但建议将‘simulated’改为‘demonstrated’以更正式。至此你已经成功构建并运行了一个最简单的多Agent协作系统。虽然这里的LLM调用是模拟的但整个架构和通信流程是真实可用的。5. 完整示例集成真实LLM与HTTP服务上面的例子使用了模拟响应。现在我们将其升级集成真实的OpenAI API并暴露为HTTP服务使其成为一个可对外提供服务的应用。5.1 升级Agent集成OpenAI异步客户端我们将使用openai库的官方异步客户端。# 文件advanced_agents.py from agency_agents import Agent, Space from pydantic import BaseModel, Field from openai import AsyncOpenAI import os from contextlib import asynccontextmanager import uvicorn from fastapi import FastAPI, HTTPException # 初始化OpenAI异步客户端 client AsyncOpenAI(api_keyos.getenv(OPENAI_API_KEY)) class TranslationRequest(BaseModel): text: str Field(..., description待翻译的文本) source_lang: str Field(中文, description源语言) target_lang: str Field(英文, description目标语言) class Translator(Agent): name translator async def translate(self, request: TranslationRequest) - str: print(f[Translator] 翻译: {request.source_lang} - {request.target_lang}) prompt f请将以下{request.source_lang}文本翻译成{request.target_lang}{request.text} try: response await client.chat.completions.create( modelgpt-3.5-turbo, # 可根据需要更换模型 messages[{role: user, content: prompt}], temperature0.3, max_tokens500, ) translated response.choices[0].message.content.strip() return translated except Exception as e: print(fOpenAI调用失败: {e}) return f翻译服务暂时不可用: {str(e)} class Proofreader(Agent): name proofreader async def review(self, text: str) - dict: 校对Skill返回结构化的结果 print(f[Proofreader] 校对文本长度: {len(text)}) prompt f请对以下英文文本进行校对并返回一个JSON对象包含以下字段 - corrected_text: 修改后的文本。 - issues_found: 发现的语法或表达问题列表。 - overall_score: 整体质量评分 (1-10分)。 文本内容{text} try: response await client.chat.completions.create( modelgpt-3.5-turbo, messages[ {role: system, content: 你是一个专业的英文校对助手请始终返回有效的JSON。}, {role: user, content: prompt} ], temperature0.2, response_format{type: json_object}, # 要求返回JSON ) import json result json.loads(response.choices[0].message.content) return result except Exception as e: print(fProofreading失败: {e}) return {error: str(e)}5.2 创建FastAPI应用封装工作流我们将使用FastAPI创建一个HTTP端点接收用户请求在内部协调Translator和Proofreader完成工作。# 接上文 advanced_agents.py # 全局Space实例 space None asynccontextmanager async def lifespan(app: FastAPI): 管理应用生命周期启动时初始化Space关闭时清理。 global space # 启动 space Space() translator Translator() proofreader Proofreader() await space.add(translator) await space.add(proofreader) print(Agency Space 已启动。) yield # 关闭 await space.close() print(Agency Space 已关闭。) app FastAPI(lifespanlifespan) class UserQuery(BaseModel): chinese_text: str app.post(/translate-and-review) async def translate_and_review(query: UserQuery): 主要业务端点翻译并校对 if space is None: raise HTTPException(status_code503, detailService initializing) # 步骤1: 翻译 trans_request TranslationRequest(textquery.chinese_text) try: translated await space.run(translator, translate, trans_request.model_dump()) if 暂时不可用 in translated: raise HTTPException(status_code500, detailfTranslation failed: {translated}) except Exception as e: raise HTTPException(status_code500, detailf调用Translator失败: {e}) # 步骤2: 校对 try: review_result await space.run(proofreader, review, translated) if error in review_result: raise HTTPException(status_code500, detailfProofreading error: {review_result[error]}) except Exception as e: raise HTTPException(status_code500, detailf调用Proofreader失败: {e}) # 返回整合结果 return { original_text: query.chinese_text, translated_text: translated, proofreading_result: review_result } app.get(/health) async def health_check(): 健康检查端点 if space and len(space.agents) 0: return {status: healthy, agents: list(space.agents.keys())} return {status: unhealthy}, 503 if __name__ __main__: # 启动FastAPI服务器 uvicorn.run(app, host0.0.0.0, port8000)5.3 运行与测试设置环境变量在启动前确保设置了OPENAI_API_KEY。export OPENAI_API_KEYyour-api-key-here # Linux/macOS # set OPENAI_API_KEYyour-api-key-here # Windows CMD # $env:OPENAI_API_KEYyour-api-key-here # Windows PowerShell安装额外依赖pip install fastapi uvicorn openai启动服务python advanced_agents.py看到输出Agency Space 已启动。和Uvicorn running on http://0.0.0.0:8000即表示成功。测试API 使用curl或 Postman 等工具测试。curl -X POST http://localhost:8000/translate-and-review \ -H Content-Type: application/json \ -d {chinese_text: 人工智能正在深刻改变软件开发的方式。}预期响应示例{ original_text: 人工智能正在深刻改变软件开发的方式。, translated_text: Artificial intelligence is profoundly changing the way software is developed., proofreading_result: { corrected_text: Artificial intelligence is profoundly changing the way software is developed., issues_found: [], overall_score: 9 } }这个示例展示了一个更接近生产环境的用法集成了真实的AI服务、提供了结构化的API、并拥有了完整的应用生命周期管理。6. 运行结果与效果验证成功运行上述示例后你可以从以下几个维度验证系统的效果和状态服务可达性访问http://localhost:8000/health应返回{status:healthy,agents:[translator,proofreader]}。这证明Space和Agent已正常初始化。核心功能验证通过/translate-and-review接口发送请求。验证点包括HTTP状态码成功应为200 OK。响应结构返回的JSON应包含original_text,translated_text,proofreading_result三个字段。业务逻辑正确性translated_text应为合理的英文翻译。proofreading_result应是一个包含corrected_text、issues_found和overall_score的JSON对象。控制台日志观察服务端控制台应能看到[Translator] 翻译: ...和[Proofreader] 校对文本长度: ...的打印信息证明消息在Agent间正确流转。错误处理验证你可以通过以下方式测试系统的健壮性断开网络模拟OpenAI API调用失败观察返回的错误信息是否符合预期如“翻译服务暂时不可用”。传入空文本或超长文本检查服务是否崩溃或返回有意义的错误。停用一个Agent在代码中注释掉await space.add(proofreader)再次调用接口应收到关于调用Proofreader失败的明确错误。如果以上验证均通过说明你已成功部署了一个基于agency-agents的、具备基本容错能力的多Agent服务。7. 常见问题与排查思路在开发和部署过程中你可能会遇到以下典型问题。下表列出了现象、可能原因及解决方案。问题现象可能原因排查方式解决方案ModuleNotFoundError: No module named agency_agents1. 未正确安装agency-agents。2. 虚拟环境未激活或不对。1. 运行pip list | grep agency。2. 检查命令行提示符前是否有(agency-env)。1. 确保在正确的虚拟环境中执行pip install githttps://github.com/msitarzewski/agency-agents.git。2. 确认Python解释器路径。RuntimeError: Event loop is closed或异步相关错误1. 在非异步上下文中调用了await。2. 异步事件循环管理冲突常见于Jupyter或已有循环的环境。检查代码是否被asyncio.run(main())或类似方式正确启动。1. 确保入口函数是async def并用asyncio.run()调用。2. 在Jupyter中尝试使用nest_asyncio.apply()需安装nest_asyncio。调用space.run()时提示Agent xxx not found1. Agent名称拼写错误。2. Agent未被成功添加到Space。3.space.run在Agent完全初始化前被调用。1. 打印space.agents查看已注册的Agent名称。2. 检查await space.add(agent)是否执行成功且没有异常被吞没。1. 确保Agent类的name属性与调用时一致。2. 确保add操作在run之前完成考虑在lifespan或main函数中确保顺序。Skill方法未被调用或无响应1. Skill方法不是async异步方法。2. 传入space.run的input_data参数格式不对无法反序列化到Skill方法的Pydantic模型。1. 检查Skill方法定义是否有async。2. 在Skill方法内第一行打印日志看是否执行。3. 检查input_data确保它是字典且键名与Pydantic模型字段名匹配。1. 将所有Skill方法改为async def。2. 使用request.model_dump()来生成输入字典。3. 在Skill方法开始时添加print或日志语句用于调试。集成真实LLM时超时或报错1. API Key 未设置或无效。2. 网络问题。3. OpenAI服务端错误或额度不足。4. 未使用异步HTTP客户端在同步代码中调用了异步库。1. 检查环境变量OPENAI_API_KEY。2. 使用try...except捕获异常并打印详细错误。3. 检查OpenAI控制台用量和状态。1. 确保使用AsyncOpenAI客户端。2. 为API调用设置合理的超时参数如timeout30.0。3. 实现重试机制和降级策略如返回缓存或默认值。多个Agent协作时流程阻塞或顺序不对对await的理解有误。await space.run()会等待该次调用完成但多个space.run之间默认是顺序执行的。分析业务逻辑确认哪些步骤可以并行哪些必须串行。如果需要并行调用多个Agent使用asyncio.gather()。例如result1, result2 await asyncio.gather(space.run(agent1, skill1, data1), space.run(agent2, skill2, data2))。8. 最佳实践与工程建议将agency-agents用于实际项目时遵循以下最佳实践可以避免很多麻烦清晰的Agent与Skill边界单一职责每个Agent应只负责一个明确的领域如翻译、数据库查询、代码执行。避免创建“上帝Agent”。Skill粒度Skill应该足够细粒度。一个“处理用户请求”的Skill过于庞大应拆分为“解析意图”、“查询知识库”、“生成回复”等多个Skill。完善的错误处理与日志Skill内部每个Skill方法内部都应使用try...except捕获可能出现的异常如网络错误、API限额、数据格式错误并返回结构化的错误信息而不是让异常抛出导致整个Space崩溃。全局日志为Space和Agent配置统一的日志系统如Pythonlogging模块记录消息的流入流出、处理耗时和错误这对于调试分布式Agent交互至关重要。配置与密钥管理永远不要硬编码将API密钥、模型名称、超时时间等配置项放在环境变量或配置文件中如.env文件使用python-dotenv读取。为不同环境配置开发、测试、生产环境应使用不同的配置。性能与可观测性异步优化确保所有I/O操作网络请求、文件读写、数据库查询都是异步的以免阻塞整个事件循环。超时设置为所有对外部服务的调用如LLM API设置明确的超时。添加监控考虑在关键路径上添加指标如请求次数、处理延迟、错误率可以集成像Prometheus这样的监控系统。测试策略单元测试Skill将Skill作为纯函数或类方法进行单元测试模拟其依赖如LLM客户端。集成测试Space编写测试用例启动一个包含真实Agent的Space测试端到端的协作流程。模拟外部服务在测试中使用unittest.mock来模拟OpenAI等外部API的响应保证测试的稳定性和速度。安全考虑输入验证尽管Pydantic提供了基础的类型验证但对于来自外部的输入如API请求仍需进行严格的业务逻辑验证和清洗防止提示词注入等攻击。权限控制如果Space暴露给多用户需要考虑在Space层面或消息路由层面添加权限校验确保用户只能与授权的Agent交互。敏感信息Agent处理的数据可能包含敏感信息。确保日志不会记录敏感数据并且通信通道如果涉及网络是安全的。版本管理与部署将Agent类、Skill定义和主要工作流代码进行版本控制。使用Docker容器化你的多Agent应用确保环境一致性。考虑使用Kubernetes或类似的编排工具来管理多个Agent服务的部署、扩缩容和健康检查。agency-agents提供了一个优雅的抽象层但构建稳定、高效、安全的生产级多Agent系统仍然需要你在这些工程实践上投入精力。它更像是一个强大的“乐高底座”让你能快速拼接出想法原型而最终的“建筑”是否坚固则取决于你如何运用这些最佳实践。9. 总结与后续学习方向通过本文我们深入探讨了agency-agents这个专注于多智能体协作的Python框架。我们从其解决的核心痛点——降低多Agent系统开发门槛——入手逐步构建了一个从模拟到真实、从本地脚本到HTTP服务的完整示例。关键收获理解核心抽象掌握了Agent、Skill、Space和Message这四个核心概念理解了基于消息传递的松耦合架构优势。掌握开发流程学会了定义Agent和Skill、创建和配置Space、通过space.run进行Agent间调用的标准流程。完成集成实践成功将框架与真实的OpenAI API以及FastAPI Web框架集成构建了一个可对外提供服务的应用。识别潜在问题了解了在开发中可能遇到的常见错误及其排查方法。建立工程化思维获得了将原型发展为生产应用所需的最佳实践清单。后续可以深入的方向探索高级模式研究框架是否支持更复杂的交互模式如发布/订阅Pub/Sub、Agent的持久化状态管理、技能的动态注册与发现。可视化与调试工具寻找或开发能够可视化Space内消息流、Agent状态和性能指标的调试工具这对理解复杂协作至关重要。与其他框架对比将agency-agents与LangGraph、AutoGen、CrewAI等流行的多Agent框架进行对比分析各自在编程模型、性能、生态系统上的优劣以便为不同项目选择最合适的工具。设计复杂工作流尝试用其构建更复杂的业务场景如一个包含“需求分析Agent”、“架构设计Agent”、“代码生成Agent”和“单元测试Agent”的软件研发流水线。agency-agents项目目前可能仍处于快速迭代中建议你持续关注其GitHub仓库的更新了解最新的特性和API变化。对于希望快速搭建多智能体协作原型、并追求代码清晰度和控制力的开发者来说它是一个非常值得放入工具箱的选择。建议将本文的示例代码作为起点根据你的具体需求进行修改和扩展在实践中不断深化理解。