Agent 团队协作：如何实现“异步通信”

作为编程学习的视角，我们先把这两季的核心逻辑串起来：第九季解决了 “团队能沟通” 的问题（基础通信层），第十季解决了 “团队沟通有规则” 的问题（协议层），就像先给团队装了邮箱，再制定了 “请假审批” “方案评审” 的 公司制度 。

一、核心背景：为什么要做这两季的升级？

在第九季之前（第八季及更早），AI代理存在两个致命问题：

1. 一次性死亡：子 Agent 用完就销毁，没有 “身份” 和 “记忆” ，没法持续协作；
2. 无规则沟通：就算能通信，也只是零散消息，没有 “请求-响应” 的规范，比如关线程时容易数据丢失、改代码时没有审批流程。

第九季先搭建 “能持续沟通的团队”，第十季再给沟通加“规则和流程”，一步步接近真实的人类团队协作。

二、第九季：搭建 “有身份、能异步通信的持久 Agent 团队”

核心目标

让 AI Agent 从 “一次性工具” 变成 “持久存在的队友” ，支持 跨 Agent 的异步消息传递（比如A发消息给B，B忙完再看）。

关键组件拆解（带代码解释）

1. 团队管理核心：TeammateManager（队友生命周期管理）

作用： 创建、跟踪队友（Agent），给每个队友（Agent）分配独立线程运行（保证同时工作）。

class TeammateManager:
    def __init__(self, team_dir: Path):
        self.dir = team_dir  # 团队数据目录（存配置、邮箱）
        self.dir.mkdir(exist_ok=True)  # 目录不存在则创建
        self.config_path = self.dir / "config.json"  # 团队配置文件（存队友名单、状态）
        self.config = self._load_config()  # 加载配置（比如{"members": [{"name":"alice","role":"coder","status":"working"}]}）
        self.threads = {}  # 存储每个队友（Agent）的运行线程
    
    # 查找对应 Agent 的信息    
		def _find_member(self, name: str) -> dict:
        for m in self.config["members"]:
            if m["name"] == name:
                return m
        return None
        
		def _save_config(self, config=None):
        """保存团队配置到文件"""
        config = config or self.config
        self.config_path.write_text(json.dumps(config, indent=2))
        
    def spawn(self, name: str, role: str, prompt: str) -> str:
        # 1. 记录队友信息存储到配置文件（持久化，不会丢）
        member = {"name": name, "role": role, "status": "working"}
        self.config["members"].append(member)
        self._save_config()  # 写入config.json，就算程序重启也能找到队友
        
        # 2. 启动独立线程运行队友（持久存在，不是用完就扔）
        thread = threading.Thread(
            target=self._teammate_loop,  # 队友的核心工作循环
            args=(name, role, prompt), 
            daemon=True  # 守护线程：主程序退出时自动结束
        )
        thread.start()
        self.threads[name] = thread  # 跟踪线程状态
        return f"Spawned teammate '{name}' (role: {role})"

问题一：从 Agent 的创建上，是如何实现异步通信的？：
这和 Agent 的生命周期相关。具体可分为，（1）Agent 的持久化、（2）线程创建（跟管理 Agent 工作逻辑相关）、（3）线程的管理；

• 第一部分 - 持久化： 通过对每个 Agent 创建 json 文件的方式，持久化 Agent 的信息（包含 Agent 的名称、角色、状态）；
• 第二部分 - 创建线程： 给每个 Agent 开独立线程，让他们一直活着；为什么要开独立线程呢？，是为了让每个 Agent 的工作循环互不干扰（比如 A 队友工作时，B 队友可以摸鱼），实现异步通信功能；
• 第三部分 - 管理线程： 通过字典的形式存储所有的 Agent 线程，方便管理（比如，（1）查询线程是否存活；（2）线程状态是否空闲；（3）防止重复创建线程；

问题二：Agent 的工作逻辑是如何实现异步通信的呢？
如以下代码所示：

def _teammate_loop(self, name: str, role: str, prompt: str):
    """
    单个队友的核心工作循环（独立线程中运行）
    核心功能：
    1. 初始化队友的系统提示和对话上下文
    2. 循环读取消息队列（inbox）接收其他队友/管理器的消息
    3. 调用大模型（LLM）生成响应，支持工具调用
    4. 执行工具调用并返回结果给大模型
    5. 循环最多50次或直到停止（非工具调用）
    6. 循环结束后更新队友状态为idle并保存配置
    
    参数：
        name: str - 队友名称（唯一标识）
        role: str - 队友角色（如"后端开发"、"测试工程师"）
        prompt: str - 初始任务提示（队友的核心任务）
    """
    # 1. 构建系统提示词：定义队友的身份、工作目录、行为规则
    # WORKDIR是全局变量，代表队友的工作目录；send_message是预设的通信方式
    sys_prompt = (
        f"You are '{name}', role: {role}, at {WORKDIR}. "  # 身份与环境
        f"Use send_message to communicate. Complete your task."  # 行为规则：用send_message通信，完成任务
    )
    
    # 2. 初始化对话消息列表：初始只有用户（管理器）下发的任务提示
    # messages是大模型的对话上下文，格式遵循OpenAI API规范
    messages = [{"role": "user", "content": prompt}]
    
    # 3. 获取该队友可使用的工具列表（如文件操作、命令执行、消息发送等）
    # _teammate_tools()是管理器的私有方法，返回该队友支持的工具定义（符合OpenAI工具调用规范）
    tools = self._teammate_tools()
    
    # 4. 核心工作循环：最多执行50轮（避免无限循环）
    for _ in range(50):
        # 4.1 读取当前队友的收件箱消息（从全局消息总线BUS中读取）
        # BUS是全局消息总线，read_inbox(name)获取发给该队友的所有未读消息
        inbox = BUS.read_inbox(name)
        
        # 4.2 将收件箱的新消息添加到对话上下文（转JSON保证格式统一）
        for msg in inbox:
            messages.append({"role": "user", "content": json.dumps(msg)})
        
        try:
            # 4.3 调用大模型API生成响应
            # model=MODEL：全局变量，指定使用的大模型（如gpt-4o）
            # system=sys_prompt：设置大模型的系统角色
            # tools=tools：传入该队友可调用的工具列表
            # max_tokens=8000：限制单次响应的最大token数
            response = client.messages.create(
                model=MODEL,
                system=sys_prompt,
                messages=messages,
                tools=tools,
                max_tokens=8000,
            )
        except Exception:
            # 捕获API调用异常（如网络、额度、模型服务问题），直接终止循环
            break
        
        # 4.4 将大模型的响应添加到对话上下文（维护多轮对话状态）
        messages.append({"role": "assistant", "content": response.content})
        
        # 4.5 判断是否停止循环：如果响应不是因为工具调用停止，则结束工作
        # stop_reason="tool_use"表示大模型要求调用工具，需要继续循环；否则任务完成
        if response.stop_reason != "tool_use":
            break
        
        # 4.6 处理工具调用：遍历响应中的所有工具调用块
        results = []
        for block in response.content:
            # 筛选出类型为"tool_use"的响应块（大模型要求执行的工具）
            if block.type == "tool_use":
                # 执行工具调用：_exec是管理器的私有方法，封装工具执行逻辑
                # 参数：调用者名称、工具名称、工具入参
                output = self._exec(name, block.name, block.input)
                
                # 打印工具执行结果（截断到120字符，避免输出过长）
                print(f"  [{name}] {block.name}: {str(output)[:120]}")
                
                # 整理工具执行结果：符合OpenAI工具结果格式，用于返回给大模型
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,  # 关联对应的工具调用ID
                    "content": str(output),   # 工具执行的结果内容
                })
        
        # 4.7 将工具执行结果作为用户消息添加到上下文，供下一轮大模型处理
        messages.append({"role": "user", "content": results})
    
    # 5. 循环结束后：更新队友状态并保存配置
    # 5.1 查找该队友的信息（从管理器的成员列表中）
    member = self._find_member(name)
    
    # 5.2 如果队友存在且未被关闭（shutdown），将状态改为idle（闲置）
    if member and member["status"] != "shutdown":
        member["status"] = "idle"
        # 5.3 保存配置到文件（如config.json），持久化状态变更
        self._save_config()

异步功能实现的逻辑如下：

• 读取消息队列中的内容：读取类似异步邮箱中的内容，并将其封装到 messages 列表中；
• 第一次回答：大模型回答，解决邮箱中的问题；（两种情况:（1）无需调用工具，说明一次 QA 就解决了；（2）需要调用工具，则和原来的 Agent Loop 一样的）
• 调用所需工具：调用工具，得到 tool response；然后封装到 messages 列表中；
• Agent Loop：循环直至，无需使用工具，说明任务解决；
• 更新线程状态：更新线程状态；

2. 通信核心：MessageBus（基于JSONL的异步邮箱）

作用： 实现队友间的异步消息传递，就像公司的邮箱系统——A发消息给B，B忙完再看，消息存在文件里不会丢。

class MessageBus:
    def send(self, sender, to, content, msg_type="message", extra=None):
        # 构造标准化消息（谁发的、发给谁、内容、时间、类型）
        msg = {
            "type": msg_type,  # 消息类型（普通消息/关机请求/计划审批）
            "from": sender,
            "content": content,
            "timestamp": time.time()  # 时间戳，保证顺序
        }
        if extra:
            msg.update(extra)  # 附加字段（比如request_id）
        
        # 追加写入收件人的JSONL文件（append-only，只加不删，保证不丢消息）
        with open(self.dir / f"{to}.jsonl", "a") as f:
            f.write(json.dumps(msg) + "\n")

    def read_inbox(self, name):
        path = self.dir / f"{name}.jsonl"
        if not path.exists(): return "[]"
        
        # 读取所有消息并解析
        msgs = [json.loads(l) for l in path.read_text().strip().splitlines() if l]
        path.write_text("")  # 读完清空（drain），避免重复处理
        return json.dumps(msgs, indent=2)

MessageBus 的目的：

• 持久化 Agent 的异步邮箱： 每个 Agent 有自己的专属邮箱文件（比如 alice.jsonl、bob.jsonl）；
• 异步操作： 实现每个 Agent 发消息、读消息的功能，消息存储到 json 文件中；

3. 队友的工作循环：_teammate_loop

作用：让队友持续运行，定期查邮箱，和LLM交互，实现“持久工作”。

def _teammate_loop(self, name, role, prompt):
    # 初始化对话上下文
    messages = [{"role": "user", "content": prompt}]
    # 循环运行（最多50轮，避免无限循环）
    for _ in range(50):
        # 第一步：查邮箱，有消息就注入上下文
        inbox = BUS.read_inbox(name)
        if inbox != "[]":
            messages.append({"role": "user", "content": f"<inbox>{inbox}</inbox>"})
            messages.append({"role": "assistant", "content": "Noted inbox messages."})
        
        # 第二步：调用LLM，获取响应
        response = client.messages.create(...)
        
        # 第三步：如果LLM不调用工具了，就停止本轮工作
        if response.stop_reason != "tool_use":
            break
        
        # 第四步：执行工具（比如发消息、写代码），把结果追加到上下文
        # ...（工具执行逻辑）
    
    # 第五步：标记队友状态为“空闲”
    self._find_member(name)["status"] = "idle"

关键理解：

• 队友不是“一次性的”，而是在循环里持续运行，直到完成任务或达到循环上限；
• 每次和LLM对话前先查邮箱，保证队友能及时收到其他成员的消息；
• 状态从“working”变为“idle”，实现生命周期管理。

第九季核心亮点

特性	解决的问题	实现方式
持久化队友	Agent 用完就丢，无身份	线程+config.json持久化队友信息
异步通信	Agent 间无法传递消息	JSONL邮箱（send/read_inbox）
生命周期管理	Agent 只有“运行/死亡”两种状态	working/idle状态切换