最近在探索AI智能体开发时发现一个非常有趣且实用的开源项目——agency-agents。它不是一个独立的框架而是一个精心设计的示例集合展示了如何将多个流行的AI智能体框架如LangChain、AutoGen、CrewAI等与agency这个轻量级库进行集成。对于想要快速上手、对比不同框架特性或者需要一个清晰、可运行的“样板间”来启动自己智能体项目的开发者来说这个仓库无疑是一个宝藏。本文将带你深入解析agency-agents项目从环境搭建到核心代码解读再到实战演练手把手教你如何利用这些示例构建你自己的多智能体系统。1. 项目背景与核心概念在深入代码之前我们首先要理解几个关键概念agency库和它所集成的各个智能体框架。agency是什么agency是一个Python库它的核心目标是简化AI智能体的构建与管理。你可以把它想象成一个智能体的“运行环境”或“编排层”。它不强制你使用特定的AI模型或通信协议而是提供了一套统一的抽象如Agent、Space让你能更专注于智能体的行为逻辑而不是底层的通信细节。agency支持智能体之间的消息传递、并发执行并能方便地暴露智能体能力为API或WebSocket服务。为什么需要agency-agents虽然agency提供了基础架构但实际开发中我们常常希望利用更上层的框架如LangChain的工具链、AutoGen的对话模式、CrewAI的角色分工来快速赋予智能体强大的能力。然而如何将这些框架“安装”到agency的架构中对于新手来说可能是个挑战。agency-agents项目正是为了解决这个问题而生。它提供了多个即插即用的示例每个示例都演示了如何将一个特定的框架封装成agency兼容的智能体让你可以开箱即用或者以此为蓝本进行定制。常见应用场景快速原型验证当你有一个新想法需要快速验证多个智能体协作的可行性时可以直接克隆并修改这些示例。框架对比学习通过运行不同的示例你能直观感受LangChain、AutoGen、CrewAI等框架在agency环境下的工作方式和差异。生产项目起点这些示例代码结构清晰包含了依赖管理、配置读取、日志记录等工程化要素可以作为正式项目的脚手架。2. 环境准备与版本说明为了顺利运行agency-agents项目我们需要准备一个干净的Python环境。以下步骤将确保所有依赖正确安装。操作系统: 本文示例在 Ubuntu 22.04 / macOS Ventura 和 Windows 11 WSL2 下测试通过其他主流Linux发行版和macOS也应兼容。Python版本: 要求 Python 3.10 或更高版本。这是许多AI框架如LangChain的推荐版本。第一步克隆项目仓库打开终端执行以下命令获取项目代码git clone https://github.com/msitarzewski/agency-agents.git cd agency-agents第二步创建并激活虚拟环境强烈建议使用虚拟环境来隔离项目依赖避免与系统或其他项目的Python包冲突。# 创建虚拟环境 python -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows venv\Scripts\activate激活后终端提示符前通常会显示(venv)字样。第三步安装项目依赖项目根目录下的requirements.txt文件列出了所有示例可能需要的依赖。由于示例较多一次性安装所有依赖可能会很庞大。你可以选择安装全部或按需安装特定示例的依赖。# 安装全部依赖推荐以备探索所有示例 pip install -r requirements.txt如果安装过程缓慢可以考虑使用国内镜像源例如清华源pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple关键依赖版本说明以写作时最新稳定版为例agency: 1.0.0 - 核心编排库。openai: 1.0.0 - 用于访问GPT等模型多数示例需要。langchain: 0.1.0 - LangChain框架。autogen: 0.2.0 - AutoGen框架。crewai: 0.1.0 - CrewAI框架。litellm: - 可选的模型调用代理支持多种后端。重要提示AI生态发展迅速依赖版本可能频繁更新。如果运行示例时遇到ImportError或API不兼容错误请首先检查requirements.txt中的版本或尝试使用pip install --upgrade package_name更新到最新版。本文重点在于架构和集成思路具体API调用细节需根据你使用的框架版本进行调整。第四步配置API密钥绝大多数示例都需要访问大语言模型如OpenAI GPT。你需要准备相应的API密钥。在项目根目录下复制环境变量示例文件cp .env.example .env使用文本编辑器打开.env文件填入你的API密钥OPENAI_API_KEYsk-your-openai-api-key-here # 其他可能的密钥如ANTHROPIC_API_KEY, GROQ_API_KEY等根据示例需要填写确保.env文件已被加载。示例代码通常使用python-dotenv库来自动加载该文件。3. 项目结构与核心原理拆解让我们先浏览一下agency-agents的项目结构这有助于理解其设计哲学。一个典型的目录结构可能如下所示agency-agents/ ├── README.md ├── requirements.txt ├── .env.example ├── agents/ # 核心各类智能体的实现 │ ├── __init__.py │ ├── base_agent.py # 可能存在的基类 │ ├── langchain_agent.py │ ├── autogen_agent.py │ └── crewai_agent.py ├── spaces/ # Agency Space 定义智能体活动的“空间” │ └── main_space.py ├── tools/ # 自定义工具供智能体使用 │ └── calculator.py ├── examples/ # 可运行的示例脚本 │ ├── 01_langchain_chat.py │ ├── 02_autogen_group_chat.py │ └── 03_crewai_task_execution.py └── config/ # 配置文件 └── settings.py核心集成原理agency定义了一个Agent基类。任何想要在agency空间中运行的智能体都需要继承这个类并实现其__call__或相应的方法。agency-agents项目的核心工作就是为每个第三方框架如LangChain编写一个“适配器”Wrapper。这个适配器的主要职责是初始化加载第三方框架的智能体或链Chain并配置好必要的工具、模型。消息桥接将agency空间收到的消息通常是字典或特定对象转换成第三方框架能理解的输入格式。调用执行调用第三方框架的运行时获取结果。结果返回将第三方框架的输出重新封装成agency空间能识别的消息格式并返回。通过这种方式LangChainAgent、AutoGenAgent等就变成了agency空间中的一等公民可以和其他原生agency智能体或其它框架的智能体无缝通信和协作。4. 实战案例构建一个LangChain智能体并与之对话我们以最常用的LangChain为例详细走一遍从零开始利用agency-agents的示例创建一个具备网络搜索能力的对话智能体。4.1 理解示例代码langchain_agent.py首先查看agents/langchain_agent.py假设路径如此请以实际项目为准。这个文件是集成的关键。# agents/langchain_agent.py import os from typing import Any, Dict from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.tools import Tool from langchain import hub from agency import Agent class LangChainAgent(Agent): 一个将LangChain Agent封装为Agency Agent的适配器。 def __init__(self, id: str, tools: list[Tool], llmNone, **kwargs): super().__init__(id, **kwargs) # 1. 初始化LLM self.llm llm or ChatOpenAI(modelgpt-4o-mini, temperature0, api_keyos.getenv(OPENAI_API_KEY)) # 2. 从LangChain Hub拉取提示词模板或使用本地模板 self.prompt hub.pull(hwchase17/openai-tools-agent) # 3. 使用LangChain创建Agent self.agent create_openai_tools_agent(llmself.llm, toolstools, promptself.prompt) # 4. 创建执行器 self.agent_executor AgentExecutor(agentself.agent, toolstools, verboseTrue) async def __call__(self, message: Dict[str, Any]) - Dict[str, Any]: 处理来自Agency Space的消息。 # 提取人类输入的问题 human_input message.get(content, ) if not human_input: return {content: Received an empty message.} # 调用LangChain Agent执行器 try: response await self.agent_executor.ainvoke({input: human_input}) output response.get(output, No output generated.) except Exception as e: output fAn error occurred: {str(e)} # 将结果封装成Agency消息格式返回 return {content: output}代码解读继承LangChainAgent继承自agency.Agent。初始化在__init__中它完成了LangChain智能体的全套初始化准备LLM、工具、提示词模板并最终创建出AgentExecutor。消息处理__call__方法是agency智能体的入口。它接收一个消息字典从中提取用户输入然后调用agent_executor.ainvoke异步方法来执行LangChain智能体。最后将结果包装后返回。关键点tools参数需要传入一个LangChainTool对象的列表。这是智能体能力的来源。4.2 创建自定义工具为了让智能体更强大我们给它添加一个自定义工具。查看tools/calculator.py。# tools/calculator.py from langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Type class CalculatorInput(BaseModel): a: float Field(descriptionThe first number) b: float Field(descriptionThe second number) operator: str Field(descriptionThe operator, one of [, -, *, /]) class CalculatorTool(BaseTool): name calculator description Useful for performing basic arithmetic calculations. Input must be a JSON string with keys a, b, and operator. args_schema: Type[BaseModel] CalculatorInput def _run(self, a: float, b: float, operator: str) - str: 执行计算 try: if operator : result a b elif operator -: result a - b elif operator *: result a * b elif operator /: if b 0: return Error: Division by zero. result a / b else: return fError: Unsupported operator {operator}. Use , -, *, or /. return fThe result of {a} {operator} {b} is {result} except Exception as e: return fCalculation error: {str(e)} async def _arun(self, a: float, b: float, operator: str) - str: 异步执行计算如果支持 return self._run(a, b, operator)这个工具定义了一个计算器智能体可以调用它来执行四则运算。BaseTool是LangChain定义工具的标准方式。4.3 编写运行脚本现在我们创建一个示例脚本将智能体、工具和空间组合起来。参考examples/01_langchain_chat.py。# examples/my_langchain_demo.py import asyncio import os from dotenv import load_dotenv from langchain_community.tools import DuckDuckGoSearchRun # 示例添加一个网络搜索工具 # 加载环境变量 load_dotenv() # 导入自定义模块 from agents.langchain_agent import LangChainAgent from tools.calculator import CalculatorTool from spaces.main_space import MySpace # 假设存在一个定义好的Space async def main(): # 1. 创建工具列表 search_tool DuckDuckGoSearchRun() calculator_tool CalculatorTool() tools [search_tool, calculator_tool] # 2. 创建LangChain智能体实例 langchain_agent LangChainAgent( idResearchAssistant, toolstools, llmNone # 使用类内部默认的ChatOpenAI ) # 3. 创建Agency Space并添加智能体 space MySpace() # 或使用 agency.Space() space.add(langchain_agent) # 4. 与智能体对话 print(LangChain Agent Demo Started. Type quit to exit.) while True: try: user_input input(\nYou: ) if user_input.lower() quit: break # 构造Agency格式的消息 message { from: User, to: ResearchAssistant, action: say, content: user_input } # 将消息放入Space触发智能体处理 # 注意这里简化了调用实际可能需要使用space.dispatch或异步方法 response await space.process_message(message) print(fAssistant: {response.get(content)}) except KeyboardInterrupt: break except Exception as e: print(fError: {e}) if __name__ __main__: asyncio.run(main())4.4 运行与验证在终端中确保虚拟环境已激活并且.env文件中的OPENAI_API_KEY已正确设置。运行你的脚本python examples/my_langchain_demo.py预期交互LangChain Agent Demo Started. Type quit to exit. You: 今天的天气怎么样 Assistant Entering new AgentExecutor chain... Action: duckduckgo_search Action Input: 今天天气 Observation: [搜索返回的天气信息摘要...] Thought: 根据搜索结果今天... Action: Final Answer Final Answer: 根据网络信息今天北京晴转多云气温15-25摄氏度。 Assistant: 根据网络信息今天北京晴转多云气温15-25摄氏度。 You: 计算一下 125 乘以 48 等于多少 Assistant Entering new AgentExecutor chain... Action: calculator Action Input: {a: 125, b: 48, operator: *} Observation: The result of 125 * 48 is 6000 Thought: 计算完成。 Action: Final Answer Final Answer: 125 乘以 48 等于 6000。 Assistant: 125 乘以 48 等于 6000。 You: quit你会看到LangChain智能体在内部进行思考Thought、选择工具Action、执行工具、并最终给出答案的完整链式过程。verboseTrue参数让这个过程可视化非常适合调试和学习。5. 扩展集成AutoGen实现多智能体协作agency-agents的魅力在于可以轻松混合不同的框架。假设我们想让一个LangChain智能体和一个AutoGen智能体在一个空间里协作。5.1 查看AutoGen智能体适配器首先查看agents/autogen_agent.py是如何实现的。其核心思想类似将AutoGen的AssistantAgent包装成agency.Agent。5.2 创建多智能体协作空间我们修改之前的示例创建一个新的Space容纳两个智能体。# examples/multi_agent_collab.py import asyncio from agents.langchain_agent import LangChainAgent from agents.autogen_agent import AutoGenAgent # 假设已实现 from tools.calculator import CalculatorTool from agency import Space async def main(): # 1. 创建工具和智能体 calc_tool CalculatorTool() langchain_tools [calc_tool] researcher LangChainAgent( idResearcher, toolslangchain_tools, description擅长信息检索和数据分析 ) writer AutoGenAgent( idWriter, llm_config{model: gpt-4o-mini}, # AutoGen风格的配置 description擅长总结和撰写报告 ) # 2. 创建Space space Space() space.add(researcher) space.add(writer) # 3. 定义一个协作任务 # 假设Space支持广播或定向消息 task 请协作完成一份关于可再生能源最新进展的简短报告。Researcher先搜集关键信息Writer随后进行总结撰写。 # 模拟任务发起 initial_message { from: Manager, to: [Researcher, Writer], # 发送给多个智能体 action: start_task, content: task } print(fManager: {task}) responses await space.broadcast(initial_message) # 假设有广播方法 # 处理响应这里简化实际需要更复杂的消息循环 for agent_id, response in responses.items(): print(f{agent_id}: {response.get(content, No response)}) # 在实际项目中你需要设计更精细的消息路由逻辑让智能体之间可以对话。 # Agency Space 提供了基础的消息传递机制你可以在此基础上构建复杂的交互协议。 if __name__ __main__: asyncio.run(main())这个示例展示了将不同框架的智能体纳入统一空间管理的可能性。真正的多轮协作需要更复杂的编排逻辑但agency和agency-agents为你打下了坚实的基础。6. 常见问题与排查思路在集成和使用过程中你可能会遇到以下典型问题问题现象可能原因解决思路ModuleNotFoundError: No module named agency1. 未安装agency库。2. 虚拟环境未激活或不对。1. 运行pip install agency。2. 确认终端提示符前有(venv)或重新激活虚拟环境。openai.AuthenticationError1..env文件未创建或路径不对。2.OPENAI_API_KEY未设置或错误。3. 代码中未正确加载.env。1. 确认项目根目录存在.env文件。2. 检查密钥是否正确确保没有多余空格。3. 在代码最开头添加load_dotenv()。LangChainAgent执行无反应或报错1. LangChain版本不兼容。2. 工具Tool定义格式错误。3. 提示词模板拉取失败。1. 检查requirements.txt中LangChain版本尝试固定版本如langchain0.1.0。2. 确保工具类正确继承BaseTool并实现了_run方法。3. 网络问题可尝试将提示词模板hub.pull(...)替换为本地字符串。智能体不调用工具直接由LLM回答1. 工具描述description不清晰LLM无法理解何时使用。2. LLM温度temperature过高导致输出随机。3. 提示词模板不适合工具调用。1. 优化工具描述明确使用场景和输入格式。2. 创建Agent时设置temperature0。3. 尝试使用不同的Agent类型如create_react_agent或自定义提示词。运行示例时出现asyncio相关错误1. 脚本未使用异步方式运行。2. 事件循环在IDE或某些环境中冲突。1. 确保主函数是async并使用asyncio.run(main())。2. 在Jupyter Notebook中使用await直接调用或尝试nest_asyncio.apply()。消息在Space中无法路由1. 消息格式不符合Space预期。2. 智能体ID与消息中的to字段不匹配。3. 智能体未正确添加到Space。1. 查阅agency文档确认标准的消息格式。2. 打印检查space.agents()列表确认智能体ID。3. 确保在调用space.add(agent)后才发送消息。7. 最佳实践与工程建议将agency-agents用于实际项目时遵循以下建议可以提升代码质量和可维护性1. 依赖与版本管理使用requirements.txt或pyproject.toml明确记录所有依赖及其版本特别是agency、langchain、openai等核心库。考虑使用pip-tools或poetry进行更精细的管理。定期更新与测试AI库更新频繁定期在测试环境中更新依赖并运行核心用例避免累积重大变更风险。2. 配置外部化将所有配置如模型名称、API Base URL、温度参数提取到配置文件如config/settings.py或config.yaml或环境变量中。避免在代码中硬编码API密钥。使用.env文件并通过python-dotenv加载。3. 智能体设计单一职责每个智能体应专注于一类任务。例如一个负责搜索一个负责分析一个负责格式化输出。清晰的ID与描述为智能体设置语义化的ID如DataAnalyzer和详细的description这有助于其他智能体或调度器理解其能力。健壮的错误处理在智能体的__call__方法中用try-except块包裹核心逻辑返回友好的错误信息并记录日志避免整个系统因单个智能体崩溃而停滞。4. 日志与可观测性集成日志库如logging或structlog在关键步骤收到消息、调用工具、返回结果、发生错误记录信息。考虑在Space层面添加中间件Middleware来记录所有流经的消息便于调试复杂的多智能体交互。5. 测试策略单元测试为每个自定义Tool编写测试验证其输入输出。集成测试为每个Agent类编写测试模拟输入消息验证其响应格式和内容是否符合预期。场景测试编写端到端测试启动一个包含多个智能体的Space发送任务验证最终的协作产出。6. 性能与扩展异步化充分利用agency和底层框架如LangChain的异步支持ainvoke,arun提高智能体处理并发请求的能力。资源池对于重量级资源如某些LLM客户端考虑在智能体间共享或使用连接池。水平扩展如果单个进程成为瓶颈agency支持分布式部署。你可以将不同的智能体部署到不同的服务节点上通过消息队列如Redis进行通信。7. 安全边界工具权限仔细审查智能体可用的工具。特别是网络访问、文件读写、系统命令执行等高风险工具应进行严格的输入校验和权限控制。输入净化对来自外部的、传递给LLM或工具的输入进行必要的清理和校验防止提示词注入等攻击。输出过滤对智能体的输出进行审查或过滤避免其返回不当内容。
AI智能体开发实战：基于agency-agents集成LangChain与AutoGen框架
最近在探索AI智能体开发时发现一个非常有趣且实用的开源项目——agency-agents。它不是一个独立的框架而是一个精心设计的示例集合展示了如何将多个流行的AI智能体框架如LangChain、AutoGen、CrewAI等与agency这个轻量级库进行集成。对于想要快速上手、对比不同框架特性或者需要一个清晰、可运行的“样板间”来启动自己智能体项目的开发者来说这个仓库无疑是一个宝藏。本文将带你深入解析agency-agents项目从环境搭建到核心代码解读再到实战演练手把手教你如何利用这些示例构建你自己的多智能体系统。1. 项目背景与核心概念在深入代码之前我们首先要理解几个关键概念agency库和它所集成的各个智能体框架。agency是什么agency是一个Python库它的核心目标是简化AI智能体的构建与管理。你可以把它想象成一个智能体的“运行环境”或“编排层”。它不强制你使用特定的AI模型或通信协议而是提供了一套统一的抽象如Agent、Space让你能更专注于智能体的行为逻辑而不是底层的通信细节。agency支持智能体之间的消息传递、并发执行并能方便地暴露智能体能力为API或WebSocket服务。为什么需要agency-agents虽然agency提供了基础架构但实际开发中我们常常希望利用更上层的框架如LangChain的工具链、AutoGen的对话模式、CrewAI的角色分工来快速赋予智能体强大的能力。然而如何将这些框架“安装”到agency的架构中对于新手来说可能是个挑战。agency-agents项目正是为了解决这个问题而生。它提供了多个即插即用的示例每个示例都演示了如何将一个特定的框架封装成agency兼容的智能体让你可以开箱即用或者以此为蓝本进行定制。常见应用场景快速原型验证当你有一个新想法需要快速验证多个智能体协作的可行性时可以直接克隆并修改这些示例。框架对比学习通过运行不同的示例你能直观感受LangChain、AutoGen、CrewAI等框架在agency环境下的工作方式和差异。生产项目起点这些示例代码结构清晰包含了依赖管理、配置读取、日志记录等工程化要素可以作为正式项目的脚手架。2. 环境准备与版本说明为了顺利运行agency-agents项目我们需要准备一个干净的Python环境。以下步骤将确保所有依赖正确安装。操作系统: 本文示例在 Ubuntu 22.04 / macOS Ventura 和 Windows 11 WSL2 下测试通过其他主流Linux发行版和macOS也应兼容。Python版本: 要求 Python 3.10 或更高版本。这是许多AI框架如LangChain的推荐版本。第一步克隆项目仓库打开终端执行以下命令获取项目代码git clone https://github.com/msitarzewski/agency-agents.git cd agency-agents第二步创建并激活虚拟环境强烈建议使用虚拟环境来隔离项目依赖避免与系统或其他项目的Python包冲突。# 创建虚拟环境 python -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows venv\Scripts\activate激活后终端提示符前通常会显示(venv)字样。第三步安装项目依赖项目根目录下的requirements.txt文件列出了所有示例可能需要的依赖。由于示例较多一次性安装所有依赖可能会很庞大。你可以选择安装全部或按需安装特定示例的依赖。# 安装全部依赖推荐以备探索所有示例 pip install -r requirements.txt如果安装过程缓慢可以考虑使用国内镜像源例如清华源pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple关键依赖版本说明以写作时最新稳定版为例agency: 1.0.0 - 核心编排库。openai: 1.0.0 - 用于访问GPT等模型多数示例需要。langchain: 0.1.0 - LangChain框架。autogen: 0.2.0 - AutoGen框架。crewai: 0.1.0 - CrewAI框架。litellm: - 可选的模型调用代理支持多种后端。重要提示AI生态发展迅速依赖版本可能频繁更新。如果运行示例时遇到ImportError或API不兼容错误请首先检查requirements.txt中的版本或尝试使用pip install --upgrade package_name更新到最新版。本文重点在于架构和集成思路具体API调用细节需根据你使用的框架版本进行调整。第四步配置API密钥绝大多数示例都需要访问大语言模型如OpenAI GPT。你需要准备相应的API密钥。在项目根目录下复制环境变量示例文件cp .env.example .env使用文本编辑器打开.env文件填入你的API密钥OPENAI_API_KEYsk-your-openai-api-key-here # 其他可能的密钥如ANTHROPIC_API_KEY, GROQ_API_KEY等根据示例需要填写确保.env文件已被加载。示例代码通常使用python-dotenv库来自动加载该文件。3. 项目结构与核心原理拆解让我们先浏览一下agency-agents的项目结构这有助于理解其设计哲学。一个典型的目录结构可能如下所示agency-agents/ ├── README.md ├── requirements.txt ├── .env.example ├── agents/ # 核心各类智能体的实现 │ ├── __init__.py │ ├── base_agent.py # 可能存在的基类 │ ├── langchain_agent.py │ ├── autogen_agent.py │ └── crewai_agent.py ├── spaces/ # Agency Space 定义智能体活动的“空间” │ └── main_space.py ├── tools/ # 自定义工具供智能体使用 │ └── calculator.py ├── examples/ # 可运行的示例脚本 │ ├── 01_langchain_chat.py │ ├── 02_autogen_group_chat.py │ └── 03_crewai_task_execution.py └── config/ # 配置文件 └── settings.py核心集成原理agency定义了一个Agent基类。任何想要在agency空间中运行的智能体都需要继承这个类并实现其__call__或相应的方法。agency-agents项目的核心工作就是为每个第三方框架如LangChain编写一个“适配器”Wrapper。这个适配器的主要职责是初始化加载第三方框架的智能体或链Chain并配置好必要的工具、模型。消息桥接将agency空间收到的消息通常是字典或特定对象转换成第三方框架能理解的输入格式。调用执行调用第三方框架的运行时获取结果。结果返回将第三方框架的输出重新封装成agency空间能识别的消息格式并返回。通过这种方式LangChainAgent、AutoGenAgent等就变成了agency空间中的一等公民可以和其他原生agency智能体或其它框架的智能体无缝通信和协作。4. 实战案例构建一个LangChain智能体并与之对话我们以最常用的LangChain为例详细走一遍从零开始利用agency-agents的示例创建一个具备网络搜索能力的对话智能体。4.1 理解示例代码langchain_agent.py首先查看agents/langchain_agent.py假设路径如此请以实际项目为准。这个文件是集成的关键。# agents/langchain_agent.py import os from typing import Any, Dict from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.tools import Tool from langchain import hub from agency import Agent class LangChainAgent(Agent): 一个将LangChain Agent封装为Agency Agent的适配器。 def __init__(self, id: str, tools: list[Tool], llmNone, **kwargs): super().__init__(id, **kwargs) # 1. 初始化LLM self.llm llm or ChatOpenAI(modelgpt-4o-mini, temperature0, api_keyos.getenv(OPENAI_API_KEY)) # 2. 从LangChain Hub拉取提示词模板或使用本地模板 self.prompt hub.pull(hwchase17/openai-tools-agent) # 3. 使用LangChain创建Agent self.agent create_openai_tools_agent(llmself.llm, toolstools, promptself.prompt) # 4. 创建执行器 self.agent_executor AgentExecutor(agentself.agent, toolstools, verboseTrue) async def __call__(self, message: Dict[str, Any]) - Dict[str, Any]: 处理来自Agency Space的消息。 # 提取人类输入的问题 human_input message.get(content, ) if not human_input: return {content: Received an empty message.} # 调用LangChain Agent执行器 try: response await self.agent_executor.ainvoke({input: human_input}) output response.get(output, No output generated.) except Exception as e: output fAn error occurred: {str(e)} # 将结果封装成Agency消息格式返回 return {content: output}代码解读继承LangChainAgent继承自agency.Agent。初始化在__init__中它完成了LangChain智能体的全套初始化准备LLM、工具、提示词模板并最终创建出AgentExecutor。消息处理__call__方法是agency智能体的入口。它接收一个消息字典从中提取用户输入然后调用agent_executor.ainvoke异步方法来执行LangChain智能体。最后将结果包装后返回。关键点tools参数需要传入一个LangChainTool对象的列表。这是智能体能力的来源。4.2 创建自定义工具为了让智能体更强大我们给它添加一个自定义工具。查看tools/calculator.py。# tools/calculator.py from langchain.tools import BaseTool from pydantic import BaseModel, Field from typing import Type class CalculatorInput(BaseModel): a: float Field(descriptionThe first number) b: float Field(descriptionThe second number) operator: str Field(descriptionThe operator, one of [, -, *, /]) class CalculatorTool(BaseTool): name calculator description Useful for performing basic arithmetic calculations. Input must be a JSON string with keys a, b, and operator. args_schema: Type[BaseModel] CalculatorInput def _run(self, a: float, b: float, operator: str) - str: 执行计算 try: if operator : result a b elif operator -: result a - b elif operator *: result a * b elif operator /: if b 0: return Error: Division by zero. result a / b else: return fError: Unsupported operator {operator}. Use , -, *, or /. return fThe result of {a} {operator} {b} is {result} except Exception as e: return fCalculation error: {str(e)} async def _arun(self, a: float, b: float, operator: str) - str: 异步执行计算如果支持 return self._run(a, b, operator)这个工具定义了一个计算器智能体可以调用它来执行四则运算。BaseTool是LangChain定义工具的标准方式。4.3 编写运行脚本现在我们创建一个示例脚本将智能体、工具和空间组合起来。参考examples/01_langchain_chat.py。# examples/my_langchain_demo.py import asyncio import os from dotenv import load_dotenv from langchain_community.tools import DuckDuckGoSearchRun # 示例添加一个网络搜索工具 # 加载环境变量 load_dotenv() # 导入自定义模块 from agents.langchain_agent import LangChainAgent from tools.calculator import CalculatorTool from spaces.main_space import MySpace # 假设存在一个定义好的Space async def main(): # 1. 创建工具列表 search_tool DuckDuckGoSearchRun() calculator_tool CalculatorTool() tools [search_tool, calculator_tool] # 2. 创建LangChain智能体实例 langchain_agent LangChainAgent( idResearchAssistant, toolstools, llmNone # 使用类内部默认的ChatOpenAI ) # 3. 创建Agency Space并添加智能体 space MySpace() # 或使用 agency.Space() space.add(langchain_agent) # 4. 与智能体对话 print(LangChain Agent Demo Started. Type quit to exit.) while True: try: user_input input(\nYou: ) if user_input.lower() quit: break # 构造Agency格式的消息 message { from: User, to: ResearchAssistant, action: say, content: user_input } # 将消息放入Space触发智能体处理 # 注意这里简化了调用实际可能需要使用space.dispatch或异步方法 response await space.process_message(message) print(fAssistant: {response.get(content)}) except KeyboardInterrupt: break except Exception as e: print(fError: {e}) if __name__ __main__: asyncio.run(main())4.4 运行与验证在终端中确保虚拟环境已激活并且.env文件中的OPENAI_API_KEY已正确设置。运行你的脚本python examples/my_langchain_demo.py预期交互LangChain Agent Demo Started. Type quit to exit. You: 今天的天气怎么样 Assistant Entering new AgentExecutor chain... Action: duckduckgo_search Action Input: 今天天气 Observation: [搜索返回的天气信息摘要...] Thought: 根据搜索结果今天... Action: Final Answer Final Answer: 根据网络信息今天北京晴转多云气温15-25摄氏度。 Assistant: 根据网络信息今天北京晴转多云气温15-25摄氏度。 You: 计算一下 125 乘以 48 等于多少 Assistant Entering new AgentExecutor chain... Action: calculator Action Input: {a: 125, b: 48, operator: *} Observation: The result of 125 * 48 is 6000 Thought: 计算完成。 Action: Final Answer Final Answer: 125 乘以 48 等于 6000。 Assistant: 125 乘以 48 等于 6000。 You: quit你会看到LangChain智能体在内部进行思考Thought、选择工具Action、执行工具、并最终给出答案的完整链式过程。verboseTrue参数让这个过程可视化非常适合调试和学习。5. 扩展集成AutoGen实现多智能体协作agency-agents的魅力在于可以轻松混合不同的框架。假设我们想让一个LangChain智能体和一个AutoGen智能体在一个空间里协作。5.1 查看AutoGen智能体适配器首先查看agents/autogen_agent.py是如何实现的。其核心思想类似将AutoGen的AssistantAgent包装成agency.Agent。5.2 创建多智能体协作空间我们修改之前的示例创建一个新的Space容纳两个智能体。# examples/multi_agent_collab.py import asyncio from agents.langchain_agent import LangChainAgent from agents.autogen_agent import AutoGenAgent # 假设已实现 from tools.calculator import CalculatorTool from agency import Space async def main(): # 1. 创建工具和智能体 calc_tool CalculatorTool() langchain_tools [calc_tool] researcher LangChainAgent( idResearcher, toolslangchain_tools, description擅长信息检索和数据分析 ) writer AutoGenAgent( idWriter, llm_config{model: gpt-4o-mini}, # AutoGen风格的配置 description擅长总结和撰写报告 ) # 2. 创建Space space Space() space.add(researcher) space.add(writer) # 3. 定义一个协作任务 # 假设Space支持广播或定向消息 task 请协作完成一份关于可再生能源最新进展的简短报告。Researcher先搜集关键信息Writer随后进行总结撰写。 # 模拟任务发起 initial_message { from: Manager, to: [Researcher, Writer], # 发送给多个智能体 action: start_task, content: task } print(fManager: {task}) responses await space.broadcast(initial_message) # 假设有广播方法 # 处理响应这里简化实际需要更复杂的消息循环 for agent_id, response in responses.items(): print(f{agent_id}: {response.get(content, No response)}) # 在实际项目中你需要设计更精细的消息路由逻辑让智能体之间可以对话。 # Agency Space 提供了基础的消息传递机制你可以在此基础上构建复杂的交互协议。 if __name__ __main__: asyncio.run(main())这个示例展示了将不同框架的智能体纳入统一空间管理的可能性。真正的多轮协作需要更复杂的编排逻辑但agency和agency-agents为你打下了坚实的基础。6. 常见问题与排查思路在集成和使用过程中你可能会遇到以下典型问题问题现象可能原因解决思路ModuleNotFoundError: No module named agency1. 未安装agency库。2. 虚拟环境未激活或不对。1. 运行pip install agency。2. 确认终端提示符前有(venv)或重新激活虚拟环境。openai.AuthenticationError1..env文件未创建或路径不对。2.OPENAI_API_KEY未设置或错误。3. 代码中未正确加载.env。1. 确认项目根目录存在.env文件。2. 检查密钥是否正确确保没有多余空格。3. 在代码最开头添加load_dotenv()。LangChainAgent执行无反应或报错1. LangChain版本不兼容。2. 工具Tool定义格式错误。3. 提示词模板拉取失败。1. 检查requirements.txt中LangChain版本尝试固定版本如langchain0.1.0。2. 确保工具类正确继承BaseTool并实现了_run方法。3. 网络问题可尝试将提示词模板hub.pull(...)替换为本地字符串。智能体不调用工具直接由LLM回答1. 工具描述description不清晰LLM无法理解何时使用。2. LLM温度temperature过高导致输出随机。3. 提示词模板不适合工具调用。1. 优化工具描述明确使用场景和输入格式。2. 创建Agent时设置temperature0。3. 尝试使用不同的Agent类型如create_react_agent或自定义提示词。运行示例时出现asyncio相关错误1. 脚本未使用异步方式运行。2. 事件循环在IDE或某些环境中冲突。1. 确保主函数是async并使用asyncio.run(main())。2. 在Jupyter Notebook中使用await直接调用或尝试nest_asyncio.apply()。消息在Space中无法路由1. 消息格式不符合Space预期。2. 智能体ID与消息中的to字段不匹配。3. 智能体未正确添加到Space。1. 查阅agency文档确认标准的消息格式。2. 打印检查space.agents()列表确认智能体ID。3. 确保在调用space.add(agent)后才发送消息。7. 最佳实践与工程建议将agency-agents用于实际项目时遵循以下建议可以提升代码质量和可维护性1. 依赖与版本管理使用requirements.txt或pyproject.toml明确记录所有依赖及其版本特别是agency、langchain、openai等核心库。考虑使用pip-tools或poetry进行更精细的管理。定期更新与测试AI库更新频繁定期在测试环境中更新依赖并运行核心用例避免累积重大变更风险。2. 配置外部化将所有配置如模型名称、API Base URL、温度参数提取到配置文件如config/settings.py或config.yaml或环境变量中。避免在代码中硬编码API密钥。使用.env文件并通过python-dotenv加载。3. 智能体设计单一职责每个智能体应专注于一类任务。例如一个负责搜索一个负责分析一个负责格式化输出。清晰的ID与描述为智能体设置语义化的ID如DataAnalyzer和详细的description这有助于其他智能体或调度器理解其能力。健壮的错误处理在智能体的__call__方法中用try-except块包裹核心逻辑返回友好的错误信息并记录日志避免整个系统因单个智能体崩溃而停滞。4. 日志与可观测性集成日志库如logging或structlog在关键步骤收到消息、调用工具、返回结果、发生错误记录信息。考虑在Space层面添加中间件Middleware来记录所有流经的消息便于调试复杂的多智能体交互。5. 测试策略单元测试为每个自定义Tool编写测试验证其输入输出。集成测试为每个Agent类编写测试模拟输入消息验证其响应格式和内容是否符合预期。场景测试编写端到端测试启动一个包含多个智能体的Space发送任务验证最终的协作产出。6. 性能与扩展异步化充分利用agency和底层框架如LangChain的异步支持ainvoke,arun提高智能体处理并发请求的能力。资源池对于重量级资源如某些LLM客户端考虑在智能体间共享或使用连接池。水平扩展如果单个进程成为瓶颈agency支持分布式部署。你可以将不同的智能体部署到不同的服务节点上通过消息队列如Redis进行通信。7. 安全边界工具权限仔细审查智能体可用的工具。特别是网络访问、文件读写、系统命令执行等高风险工具应进行严格的输入校验和权限控制。输入净化对来自外部的、传递给LLM或工具的输入进行必要的清理和校验防止提示词注入等攻击。输出过滤对智能体的输出进行审查或过滤避免其返回不当内容。
