大家好，我是长期分享技术实战经验的博主。在探索AI智能体（Agent）开发时，你是否遇到过这样的困境：想快速搭建一个能自主协作、处理复杂任务的智能体系统，却苦于框架选择、环境配置和代码整合？网上资料要么过于零散，要么停留在概念层面，难以形成可落地的闭环方案。

本文将围绕 GitHub 上备受关注的 agency-agents 项目，为你带来一份从零到一的完整实战指南。无论你是对AI智能体感兴趣的新手，还是希望将智能体能力集成到现有项目中的开发者，都能通过本文掌握其核心概念、环境搭建、代码实战以及生产级的最佳实践。我们将手把手构建一个具备记忆、工具调用和协作能力的智能体系统，并深入探讨其背后的设计思想与工程化要点。

1. 背景与核心概念：什么是 Agency-Agents？

在深入代码之前，我们有必要厘清几个核心概念。 agency-agents 并非一个官方框架，而是一个在 GitHub 上开源的、旨在简化多智能体系统构建的示例项目或工具集。它的核心思想是“代理”（Agency），即创建能够感知环境、进行决策、执行动作并与其他智能体协作的软件实体。

智能体（Agent） 是什么？你可以将其理解为一个具备一定自主性的程序。它接收输入（如用户指令、环境状态），通过内部逻辑（可能包含大型语言模型LLM）进行处理，然后产生输出或执行动作（如调用API、修改数据）。单个智能体可以完成特定任务。

多智能体系统（Multi-Agent System, MAS） 则由多个这样的智能体组成。它们通过通信、协作或竞争，共同完成单个智能体难以处理的复杂任务。例如，一个智能体负责分析需求，一个负责编写代码，另一个负责执行测试。

那么， agency-agents 项目试图解决什么问题呢？它主要针对以下痛点：

1. 降低构建门槛 ：提供一套预先构建的智能体类型、通信机制和工具集成，让开发者无需从零开始设计消息路由、状态管理等底层架构。
2. 促进模块化 ：将不同的能力（如记忆、搜索、代码执行）封装成独立的智能体或工具，便于复用和组合。
3. 简化与LLM集成 ：通常内置了对 OpenAI GPT、Anthropic Claude 等主流大语言模型的便捷调用，让智能体拥有“思考”和“规划”的能力。

常见的应用场景包括：

• 自动化工作流 ：自动处理客服问答、内容审核、数据报告生成。
• 代码助手与审查 ：自动分析需求、生成代码片段、进行代码审查。
• 研究与分析 ：多个智能体分工协作，进行网络搜索、数据整理和报告撰写。
• 模拟与游戏 ：构建具有不同角色和目标的虚拟实体进行交互。

理解这些，我们就知道 agency-agents 的目标是提供一个构建此类系统的“脚手架”或“工具箱”。

2. 环境准备与版本说明

在开始实战之前，我们需要搭建一个稳定、可复现的开发环境。由于 agency-agents 项目可能快速迭代，以下配置以常见 Python 智能体开发环境为基础，重点演示核心思路。请根据项目实际 README 或 requirements.txt 进行调整。

操作系统 ：Windows 10/11， macOS 或 Linux (如 Ubuntu 20.04+) 均可。本文命令以 Linux/macOS 的 bash 为例，Windows 用户可在 Git Bash 或 WSL 中运行。

Python 版本 ：推荐使用 Python 3.9 或 3.10。版本过高或过低可能导致依赖包兼容性问题。使用 pyenv 或 conda 管理多个Python版本是非常好的实践。

# 检查Python版本
python3 --version
# 或
python --version

关键依赖与工具 ：

1. Git ：用于克隆项目代码。
2. Poetry 或 Pip ：Python 包管理工具。 agency-agents 项目可能使用 poetry 来管理依赖，因为它能更好地处理依赖隔离。我们将同时展示两种方式。
3. OpenAI API 密钥或其他LLM服务密钥 ：这是智能体“大脑”的燃料。你需要注册相应平台并获取密钥。

项目初始化与依赖安装 ：

假设我们已经从 GitHub 克隆了项目（如果项目名为 agency-agents ）。

# 1. 克隆项目（这里使用一个假设的仓库地址，请替换为实际地址）
git clone https://github.com/msitarzewski/agency-agents.git
cd agency-agents

# 2. 使用 Poetry 安装依赖（如果项目包含 pyproject.toml 且使用 poetry）
poetry install
poetry shell # 激活虚拟环境

# 或者，使用 Pip 和 virtualenv
python3 -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install -r requirements.txt  # 如果项目提供了该文件

版本兼容性说明 ：

• openai 库：版本 >= 0.27.0，但请注意，OpenAI SDK 的 API 在 v1.x 后有重大变化。如果项目代码使用的是较旧的 openai 版本（如 0.28 ），而你现在安装的是 1.x ，那么导入和调用方式将完全不同，会导致代码运行失败。 这是最常见的坑！
• 其他依赖如 langchain ， pydantic 等也需注意版本匹配。

一个实用的建议 ：在项目根目录查看 pyproject.toml 或 requirements.txt 文件，明确记录所有依赖及其版本。在无法确定时，可以尝试安装项目指定版本。

# 示例：如果 requirements.txt 中指定了 openai==0.28.1
pip install openai==0.28.1

环境变量配置 ： 智能体需要访问 LLM API，因此必须安全地配置密钥。强烈推荐使用 .env 文件配合 python-dotenv 管理，切勿将密钥硬编码在代码中。

1. 在项目根目录创建 .env 文件：

   touch .env
   

2. 在 .env 文件中填入你的密钥：

   # .env
   OPENAI_API_KEY=sk-your-actual-openai-api-key-here
   # 可能还有其他API密钥，如 ANTHROPIC_API_KEY, SERPER_API_KEY等
   

3. 在 Python 代码中加载环境变量：

   # config.py 或主程序开头
   from dotenv import load_dotenv
   import os
   
   load_dotenv()  # 加载 .env 文件中的变量到环境变量
   
   OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
   if not OPENAI_API_KEY:
       raise ValueError("请在 .env 文件中设置 OPENAI_API_KEY")
   

至此，我们的基础环境就准备好了。接下来，我们来剖析 agency-agents 的核心架构与组件。

3. 核心架构与组件拆解

理解一个多智能体系统的框架，关键在于理清其核心抽象： 智能体（Agent） 、 工具（Tool） 、 环境（Environment） 和 通信（Communication） 。我们结合 agency-agents 项目的常见设计模式来解读。

3.1 智能体基类与角色定义

通常，框架会定义一个 Agent 基类，所有具体的智能体都继承自它。这个基类规定了智能体的生命周期和基本能力。

# 假设的 agency_agents/core/agent.py 结构
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional

class Agent(ABC):
    """智能体抽象基类"""
    def __init__(self, name: str, role: str, memory: Optional['Memory'] = None):
        self.name = name
        self.role = role  # 如 “Coder”, “Researcher”, “Planner”
        self.memory = memory  # 记忆组件，用于存储对话历史或知识
        self.tools: List['Tool'] = []  # 该智能体可以使用的工具列表

    def register_tool(self, tool: 'Tool'):
        """为智能体注册一个可用的工具"""
        self.tools.append(tool)

    @abstractmethod
    async def perceive(self, observation: Any) -> None:
        """感知环境或接收消息"""
        pass

    @abstractmethod
    async def think(self) -> Any:
        """内部思考/规划过程，可能调用LLM"""
        pass

    @abstractmethod
    async def act(self) -> Any:
        """执行动作，可能是发送消息或调用工具"""
        pass

    async def run_cycle(self):
        """运行一个感知-思考-动作循环"""
        await self.perceive(...)
        thought = await self.think()
        result = await self.act()
        return result

关键点 ：

• 异步（async/await） ：现代智能体框架普遍采用异步编程，以高效处理I/O操作（如网络请求）。
• 记忆（Memory） ：使智能体拥有上下文感知能力，可以是简单的列表，也可以是向量数据库。
• 工具（Tools） ：扩展智能体能力边界的关键。智能体本身不一定会写代码或搜索，但可以通过调用工具来完成。

3.2 工具系统：智能体的“手脚”

工具是将函数或API封装成智能体可调用的标准接口。一个典型的工具定义如下：

# 假设的 agency_agents/core/tool.py 结构
from pydantic import BaseModel, Field
from typing import Type, Callable, Any

class ToolInput(BaseModel):
    """工具输入参数的模型，使用Pydantic进行验证和文档化"""
    query: str = Field(..., description="搜索查询词")

class Tool:
    def __init__(self, name: str, description: str, func: Callable, args_schema: Type[BaseModel]):
        self.name = name
        self.description = description
        self.func = func
        self.args_schema = args_schema

    async def run(self, **kwargs) -> str:
        """执行工具，并返回字符串格式的结果"""
        # 1. 验证输入参数
        validated_args = self.args_schema(**kwargs)
        # 2. 执行实际函数
        result = await self.func(**validated_args.dict())
        # 3. 将结果格式化为字符串，便于LLM理解
        return str(result)

# 具体工具实现示例：一个简单的网络搜索工具（需要第三方API）
import aiohttp
async def web_search(query: str) -> str:
    """模拟网络搜索，实际应接入Serper、Google Search等API"""
    async with aiohttp.ClientSession() as session:
        # 这里是模拟，实际需要调用真实API
        # async with session.get(f'https://api.serper.dev/search?q={query}') as resp:
        #     data = await resp.json()
        #     return data.get('organic', [])[:3]
        return f"模拟搜索 '{query}' 的结果：相关文章1， 相关文章2。"

# 将函数包装成工具
search_tool = Tool(
    name="web_search",
    description="在互联网上搜索信息",
    func=web_search,
    args_schema=ToolInput
)

为什么需要 args_schema ？ LLM需要知道工具如何使用。通过Pydantic模型清晰地定义工具的名称、描述和参数格式，框架可以自动生成供LLM理解的“工具描述”，从而实现LLM对工具的自动调用（如 OpenAI 的 Function Calling）。

3.3 通信与协作：智能体如何“对话”

多智能体系统的魅力在于协作。通信机制通常有两种：

1. 消息总线/黑板（Message Bus/Blackboard） ：所有智能体向一个中央通道发送和监听消息。
2. 直接寻址（Direct Addressing） ：智能体明确指定消息接收者。

agency-agents 可能采用一种混合或简化的模式。例如，一个 Orchestrator 或 Coordinator 智能体负责接收用户请求，然后将其分解并分配给其他智能体。

# 假设的简单协调者智能体
class CoordinatorAgent(Agent):
    def __init__(self, name: str, agents: Dict[str, Agent]):
        super().__init__(name, role="Coordinator")
        self.agents = agents  # 它所管理的子智能体字典

    async def perceive(self, user_request: str):
        self.current_request = user_request

    async def think(self):
        # 使用LLM分析请求，决定需要哪些智能体协作
        # 例如，判断请求是“写代码”还是“做研究”
        analysis_prompt = f"""
        用户请求：{self.current_request}
        可用的智能体有：{list(self.agents.keys())}
        请分析这个任务需要哪个或哪几个智能体协作完成，并简要说明理由。
        """
        # 这里调用LLM进行分析，简化起见，我们直接写逻辑
        if "代码" in self.current_request or "编程" in self.current_request:
            self.plan = ["CoderAgent"]
        elif "研究" in self.current_request or "搜索" in self.current_request:
            self.plan = ["ResearcherAgent"]
        else:
            self.plan = ["GeneralAssistantAgent"]

    async def act(self):
        results = []
        for agent_name in self.plan:
            agent = self.agents.get(agent_name)
            if agent:
                # 将请求转发给对应的智能体，并获取结果
                result = await agent.handle_request(self.current_request)
                results.append(f"{agent_name}: {result}")
        return "\n".join(results)

3.4 记忆模块：让智能体拥有“过去”

没有记忆的智能体每次对话都是全新的开始。记忆模块用于存储和检索对话历史、任务上下文或知识片段。

# 一个基于列表的简单对话记忆
class SimpleMemory:
    def __init__(self, max_messages: int = 20):
        self.messages = []
        self.max_messages = max_messages

    def add(self, role: str, content: str):
        """添加一条消息"""
        self.messages.append({"role": role, "content": content})
        # 限制记忆长度，防止上下文过长
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]

    def get_context(self) -> List[Dict]:
        """获取最近的对话上下文"""
        return self.messages[-5:]  # 返回最近5条消息作为上下文

# 在智能体中使用记忆
class ChatAgent(Agent):
    def __init__(self, name: str, llm_client):
        super().__init__(name, role="ChatAssistant", memory=SimpleMemory())
        self.llm = llm_client

    async def respond_to(self, user_message: str) -> str:
        # 1. 将用户消息存入记忆
        self.memory.add("user", user_message)
        # 2. 从记忆中获取上下文
        context = self.memory.get_context()
        # 3. 结合上下文构造LLM提示词
        prompt = self._build_prompt(context)
        # 4. 调用LLM获取回复
        llm_response = await self.llm.chat(prompt)
        # 5. 将LLM回复存入记忆
        self.memory.add("assistant", llm_response)
        return llm_response

对于更复杂的场景，可能会使用向量数据库（如 Chroma, Pinecone）来实现基于语义的知识检索，这就是“长期记忆”或“知识库”。

理解了这些核心组件，我们就可以动手搭建一个具体的多智能体系统了。

4. 完整实战案例：构建一个代码生成与审查智能体系统

现在，我们将利用上述概念，构建一个由两个智能体协作的系统：一个 CoderAgent 负责根据需求生成代码，一个 ReviewerAgent 负责审查生成的代码并提出改进意见。用户只需提出需求，系统自动完成“生成-审查”的闭环。

4.1 项目结构设计

首先，创建清晰的项目目录。

agency-agents-demo/
├── .env                    # 环境变量文件（已添加到.gitignore）
├── .gitignore
├── pyproject.toml          # 或 requirements.txt
├── main.py                 # 主程序入口
└── src/
    ├── __init__.py
    ├── agents/             # 智能体定义
    │   ├── __init__.py
    │   ├── base.py         # Agent基类
    │   ├── coder.py        # CoderAgent
    │   └── reviewer.py     # ReviewerAgent
    ├── tools/              # 工具定义
    │   ├── __init__.py
    │   └── code_tools.py   # 代码相关工具
    ├── memory/             # 记忆模块
    │   ├── __init__.py
    │   └── simple_memory.py
    └── llm/                # LLM客户端封装
        ├── __init__.py
        └── openai_client.py

4.2 定义基础组件：LLM客户端与记忆

1. 封装 LLM 客户端 ( src/llm/openai_client.py ) ： 为了便于管理和未来切换模型，我们抽象一个LLM客户端。

import openai
from typing import List, Dict, Any
import os
from dotenv import load_dotenv

load_dotenv()

class OpenAIClient:
    def __init__(self, model: str = "gpt-3.5-turbo"):
        self.api_key = os.getenv("OPENAI_API_KEY")
        if not self.api_key:
            raise ValueError("OPENAI_API_KEY not set in environment variables")
        # 注意：openai 库版本兼容性。v1.x 后用法不同。
        # 假设使用 openai>=1.0.0
        self.client = openai.OpenAI(api_key=self.api_key)
        self.model = model

    async def chat_completion(self, messages: List[Dict[str, str]]) -> str:
        """异步调用ChatCompletion API"""
        try:
            # 注意：openai>=1.0.0 使用 client.chat.completions.create
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=0.7,
            )
            return response.choices[0].message.content
        except Exception as e:
            return f"LLM调用出错: {e}"

2. 实现简单记忆 ( src/memory/simple_memory.py ) ：

from typing import List, Dict

class SimpleMemory:
    def __init__(self, max_size: int = 10):
        self.memory: List[Dict] = []
        self.max_size = max_size

    def add(self, role: str, content: str):
        self.memory.append({"role": role, "content": content})
        if len(self.memory) > self.max_size:
            self.memory.pop(0)  # 移除最旧的记忆

    def get_recent(self, n: int = 5) -> List[Dict]:
        return self.memory[-n:] if self.memory else []

    def clear(self):
        self.memory.clear()

4.3 实现智能体基类与具体智能体

1. 智能体基类 ( src/agents/base.py ) ：

from abc import ABC, abstractmethod
from typing import Any, List
from src.memory.simple_memory import SimpleMemory

class BaseAgent(ABC):
    def __init__(self, name: str, role: str, llm_client):
        self.name = name
        self.role = role
        self.llm = llm_client
        self.memory = SimpleMemory()
        self.tools = []

    def register_tool(self, tool):
        self.tools.append(tool)

    @abstractmethod
    async def handle_request(self, request: str) -> str:
        """处理外部请求的核心方法"""
        pass

    def _get_context_prompt(self) -> str:
        """基于记忆构建上下文提示"""
        recent = self.memory.get_recent(3)
        context = "\n".join([f"{msg['role']}: {msg['content']}" for msg in recent])
        return context if context else "无先前对话。"

2. 代码生成智能体 ( src/agents/coder.py ) ：

from src.agents.base import BaseAgent
from typing import Dict

class CoderAgent(BaseAgent):
    def __init__(self, llm_client):
        super().__init__(name="Coder", role="Python代码生成专家", llm_client=llm_client)

    async def handle_request(self, request: str) -> str:
        # 将用户请求存入记忆
        self.memory.add("user", request)

        # 构建给LLM的提示词
        system_prompt = """你是一个专业的Python开发助手。根据用户需求，生成正确、高效、可读性好的Python代码。
        只返回代码块，并在代码块开始用```python标记，结束用```标记。不要包含任何解释性文字。
        """
        user_prompt = f"用户需求：{request}\n请生成满足上述需求的Python代码。"

        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ]

        # 调用LLM生成代码
        generated_code = await self.llm.chat_completion(messages)

        # 将生成的代码存入记忆
        self.memory.add("assistant", f"生成代码：\n{generated_code}")

        return generated_code

3. 代码审查智能体 ( src/agents/reviewer.py ) ：

from src.agents.base import BaseAgent

class ReviewerAgent(BaseAgent):
    def __init__(self, llm_client):
        super().__init__(name="Reviewer", role="代码审查专家", llm_client=llm_client)

    async def handle_request(self, code_to_review: str) -> str:
        # 注意：这里接收的是CoderAgent生成的代码
        self.memory.add("code_to_review", code_to_review)

        system_prompt = """你是一个资深的代码审查员。请仔细检查提供的Python代码，指出：
        1. 语法错误或潜在的运行时错误。
        2. 代码风格问题（PEP 8）。
        3. 可能的性能瓶颈或安全隐患。
        4. 给出具体的修改建议。
        请用清晰、有条理的方式列出问题和建议。
        """
        user_prompt = f"请审查以下Python代码：\n\n{code_to_review}"

        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ]

        review_comments = await self.llm.chat_completion(messages)
        self.memory.add("assistant", f"审查意见：\n{review_comments}")
        return review_comments

4.4 实现协调者与主程序

协调者/主控程序 ( main.py ) ： 这个协调者负责接收用户请求，并串联起 CoderAgent 和 ReviewerAgent 的工作流。

import asyncio
from src.llm.openai_client import OpenAIClient
from src.agents.coder import CoderAgent
from src.agents.reviewer import ReviewerAgent

class Orchestrator:
    def __init__(self):
        # 初始化LLM客户端（多个智能体可共享一个客户端，注意速率限制）
        self.llm_client = OpenAIClient()
        # 初始化各个智能体
        self.coder = CoderAgent(self.llm_client)
        self.reviewer = ReviewerAgent(self.llm_client)

    async def process_request(self, user_request: str) -> Dict[str, str]:
        """处理用户请求的核心工作流"""
        print(f"[Orchestrator] 收到用户请求: {user_request}")
        print("-" * 50)

        # 阶段1: 代码生成
        print("[Orchestrator] 调用 CoderAgent 生成代码...")
        generated_code = await self.coder.handle_request(user_request)
        print(f"[CoderAgent] 生成代码完成。")
        # print(generated_code) # 可以打印出来看

        # 阶段2: 代码审查
        print("\n[Orchestrator] 调用 ReviewerAgent 审查代码...")
        review_feedback = await self.reviewer.handle_request(generated_code)
        print(f"[ReviewerAgent] 审查完成。")

        # 返回结果
        return {
            "original_request": user_request,
            "generated_code": generated_code,
            "review_feedback": review_feedback
        }

async def main():
    orchestrator = Orchestrator()
    # 模拟用户请求
    user_request = "写一个Python函数，接收一个整数列表，返回列表中所有偶数的平方和。"
    
    result = await orchestrator.process_request(user_request)
    
    print("\n" + "="*50)
    print("最终结果汇总:")
    print("="*50)
    print(f"用户需求: {result['original_request']}")
    print("\n--- 生成的代码 ---")
    print(result['generated_code'])
    print("\n--- 代码审查意见 ---")
    print(result['review_feedback'])

if __name__ == "__main__":
    asyncio.run(main())

4.5 运行与验证

1. 安装依赖 ：在项目根目录创建 requirements.txt 并安装。

   # requirements.txt
   openai>=1.6.0
   python-dotenv>=1.0.0
   

   pip install -r requirements.txt
   

2. 配置环境变量 ：在项目根目录创建 .env 文件并填入你的 OPENAI_API_KEY 。

3. 运行程序 ：

   python main.py
   

4. 预期输出 ： 程序将依次打印出协调者调度、代码生成、代码审查的过程，并最终输出生成的Python代码和审查意见。你可能会看到类似下面的输出（具体代码和审查意见因LLM输出而异）：

   [Orchestrator] 收到用户请求: 写一个Python函数，接收一个整数列表，返回列表中所有偶数的平方和。
   --------------------------------------------------
   [Orchestrator] 调用 CoderAgent 生成代码...
   [CoderAgent] 生成代码完成。
   
   [Orchestrator] 调用 ReviewerAgent 审查代码...
   [ReviewerAgent] 审查完成。
   
   ==================================================
   最终结果汇总:
   ==================================================
   用户需求: 写一个Python函数，接收一个整数列表，返回列表中所有偶数的平方和。
   
   --- 生成的代码 ---
   ```python
   def sum_of_squares_of_evens(numbers):
       """
       计算整数列表中所有偶数的平方和。
       """
       return sum(x**2 for x in numbers if x % 2 == 0)
   

   --- 代码审查意见 ---

   1. 功能正确性 ：代码逻辑正确，使用了生成器表达式，内存友好。
   2. 代码风格 ：符合PEP 8，函数名清晰，有文档字符串。
   3. 潜在改进 ：
      • 可考虑增加输入类型检查（如使用 isinstance(numbers, list) 和 all(isinstance(i, int) for i in numbers) ），但会降低简洁性。
      • 对于空列表或非常大的列表，当前实现是高效的。
   4. 安全性 ：无安全隐患。 总体评价：代码简洁、高效、可读性好，是高质量的解决方案。

至此，一个简易但完整的双智能体协作系统就搭建完成了。 CoderAgent 和 ReviewerAgent 各司其职，通过 Orchestrator 串联，完成了从需求到代码再到质量反馈的完整流程。

5. 常见问题与排查思路

在实际开发和运行中，你可能会遇到以下典型问题：

问题现象	可能原因	排查步骤与解决方案
ModuleNotFoundError: No module named 'openai'	1. 未安装 openai 包。 2. 虚拟环境未激活或包未安装在当前环境。	1. 运行 pip install openai 。 2. 确认终端处于正确的虚拟环境中（ which python 或 pip list 查看）。
openai.AuthenticationError	1. API 密钥未设置或错误。 2. 密钥所在环境变量名不对。 3. 密钥已过期或被禁用。	1. 检查 .env 文件是否存在且内容正确，确保 load_dotenv() 已调用。 2. 在代码中打印 os.getenv(“OPENAI_API_KEY”) 的前几位（切勿打印完整密钥）确认是否加载成功。 3. 登录 OpenAI 平台检查密钥状态和余额。
AttributeError: module 'openai' has no attribute 'ChatCompletion'	OpenAI SDK 版本不兼容 。这是最常见、最关键的坑。代码基于旧版 (v0.28) 编写，但安装了新版 (v1.x) SDK。	1. 检查 openai 版本： pip show openai 。 2. 方案A（推荐） ：将代码升级到新版 API。将 openai.ChatCompletion.create 改为 client.chat.completions.create ，如本文示例。 3. 方案B ：降级SDK： pip install openai==0.28.1 。但这不是长久之计。
智能体不调用工具或调用错误	1. 工具未正确注册到智能体。 2. 工具的函数签名或 args_schema 与LLM生成的调用参数不匹配。 3. LLM未收到正确的工具描述。	1. 检查 agent.register_tool(tool) 是否执行。 2. 打印LLM收到的消息和生成的函数调用参数，对比工具定义。 3. 确保工具的描述清晰，参数格式符合 OpenAI Function Calling 规范。
程序报错 RuntimeError: Event loop is closed	异步事件循环在Windows或某些环境下处理不当。	使用 asyncio.run(main()) 作为入口点，并确保所有异步调用都正确 await 。避免在同步函数中创建新的事件循环。
LLM响应慢或超时	1. 网络问题。 2. OpenAI API 服务繁忙。 3. 提示词（Prompt）过长或过于复杂。	1. 检查网络连接。 2. 重试请求，或添加指数退避重试逻辑。 3. 优化提示词，减少不必要的内容。对于长上下文，考虑使用流式响应或摘要记忆。
多智能体协作陷入循环或无效沟通	1. 智能体角色定义不清，任务分配逻辑有误。 2. 缺乏终止条件或协调机制。	1. 在协调者（Orchestrator）中实现更明确的任务分解和路由逻辑。 2. 为智能体对话设置最大轮次限制。 3. 引入一个“仲裁者”智能体来判定任务是否完成或需要继续。

6. 最佳实践与工程建议

将智能体系统从Demo推向生产，需要考虑更多工程化因素。

1. 配置管理与安全

• 密钥管理 ：绝对不要将API密钥提交到代码仓库。使用 .env 文件，并通过 .gitignore 忽略它。在生产环境中，使用云服务商提供的密钥管理服务（如 AWS Secrets Manager, GCP Secret Manager, Azure Key Vault）。
• 配置外部化 ：将模型类型、温度、最大令牌数等参数放在配置文件（如 config.yaml ）中，便于不同环境（开发、测试、生产）切换。

2. 健壮性与错误处理

• 重试与退避 ：LLM API 调用可能因网络或服务限流失败。实现带指数退避的自动重试机制。

  import asyncio
  import random
  from openai import RateLimitError, APIError
  
  async def call_llm_with_retry(client, messages, max_retries=3):
      for attempt in range(max_retries):
          try:
              return await client.chat_completion(messages)
          except (RateLimitError, APIError) as e:
              if attempt == max_retries - 1:
                  raise
              wait_time = (2 ** attempt) + random.random()
              print(f"API调用失败，{wait_time:.2f}秒后重试... 错误: {e}")
              await asyncio.sleep(wait_time)
  

• 超时控制 ：为每个LLM调用或工具调用设置超时，防止单个环节卡死整个系统。

  import asyncio
  async def safe_agent_call(agent, request, timeout=30):
      try:
          return await asyncio.wait_for(agent.handle_request(request), timeout=timeout)
      except asyncio.TimeoutError:
          return f"{agent.name} 处理超时，请检查网络或服务状态。"
  

• 验证与过滤 ：对LLM生成的内容（特别是代码、命令）进行严格验证和沙箱执行，避免执行恶意或危险操作。

3. 性能与可观测性

• 异步并发 ：充分利用 asyncio ，让多个智能体或工具调用可以并发执行，显著提升系统吞吐量。
• 日志记录 ：详细记录每个智能体的输入、输出、LLM调用详情和工具调用结果。这不仅是调试的需要，也是理解系统行为和优化提示词的关键。
• 监控与指标 ：记录API调用次数、令牌消耗、响应时间、错误率等指标，便于成本控制和性能分析。

4. 架构设计进阶

• 状态持久化 ：将智能体的记忆、会话状态等保存到数据库（如Redis, PostgreSQL），实现跨会话的持久化智能体。
• 可插拔工具 ：设计良好的工具接口，使得新增或替换工具（如将Google搜索换成Bing搜索）变得非常容易，无需修改智能体核心逻辑。
• 混合编排模式 ：结合基于规则的编排（如本文的 Orchestrator ）和基于LLM的自主协作。例如，让一个“规划师”智能体动态决定任务分解和智能体调度。

5. 提示词工程

• 系统提示词（System Prompt） ：精心设计每个智能体的系统提示词，明确其角色、职责和行为边界。这是塑造智能体行为最有效的手段。
• 少样本学习（Few-shot Learning） ：在提示词中提供几个高质量的输入输出示例，能显著提升智能体在特定任务上的表现。
• 思维链（Chain-of-Thought） ：鼓励智能体在输出最终答案前，先输出推理步骤。这不仅能提高答案质量，也便于调试和审查。

通过遵循这些最佳实践，你可以构建出更加稳定、高效、可维护的多智能体系统，从而应对真实的业务场景挑战。

7. 总结与扩展方向

本文我们深入探讨了基于 agency-agents 理念构建多智能体系统的完整流程。我们从核心概念入手，逐步实现了环境搭建、架构设计、智能体与工具开发，并最终完成了一个代码生成与审查的协作案例。关键点在于理解智能体作为自主单元、工具作为能力扩展、以及协调者作为流程控制的核心思想。

下一步，你可以从以下几个方向深化学习与实践：

1. 探索更复杂的智能体类型 ：实现具有长期记忆（向量数据库）、网络搜索、代码执行（安全沙箱）、文件读写等能力的智能体。
2. 集成更强大的框架 ：本文是“从零实现”的教学。在实际项目中，可以考虑基于更成熟的框架开发，如 LangChain 、 AutoGen 、 CrewAI 等。它们提供了更丰富的智能体模板、工具集成和编排能力。
3. 构建图形化界面 ：为你的智能体系统开发一个Web界面（使用Streamlit、Gradio或FastAPI），让非技术用户也能方便地使用。
4. 深入提示词与模型微调 ：研究如何通过更精细的提示词工程，甚至对开源模型进行微调（Fine-tuning），来让智能体在特定领域（如法律、医疗、金融）的表现更加专业。
5. 关注智能体安全与伦理 ：随着智能体能力增强，必须考虑其生成内容的准确性、偏见、以及被滥用的风险。在设计系统时，要加入内容过滤、事实核查和人工审核环节。

多智能体系统是当前AI应用的前沿领域，它代表了让AI从“被动应答”走向“主动协作”的重要一步。希望这份教程能成为你探索这一领域的坚实起点。动手修改文中的代码，添加新的智能体，尝试解决你实际工作中的问题，是学习的最佳途径。如果在实践中遇到新的问题，欢迎在评论区交流探讨。