人工智能大模型AI Agent工作流自动化RAG【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址https://gitcode.com/gh_mirrors/poc/PocketFlow点击查看免费下载多个 Agent 可以通过各自处理子任务、互相沟通进度来协同完成复杂工作。在 PocketFlow 中Agent 之间最常见的通信方式是共享存储 消息队列每个 Agent 是一个监听循环从队列中取消息、处理、再把结果投递给下一个 Agent。本文以官方设计模式文档 docs/design_pattern/multi_agent.md 为核心骨架结合AsyncNode/AsyncFlow的源码实现与仓库内可运行的完整示例 cookbook/pocketflow-multi-agent从零讲解如何用asyncio.Queue在多个异步 Agent 之间建立可靠的双向通信并给出一个可直接运行的Taboo 猜词游戏多 Agent 实战。什么时候才需要多 Agent多 Agent 的核心思想是分治与协作把一个大任务拆成子任务交给不同的 Agent每个 Agent 专注做好自己的部分并通过通信机制交换中间结果与进度。但在动手设计多 Agent 系统之前请先记住 PocketFlow 官方文档给出的一条最佳实践Most of time, you dont need Multi-Agents. Start with a simple solution first.大多数时候你并不需要多 Agent先从一个简单的方案开始。原因在于多 Agent 引入了额外的通信开销、协调复杂度和故障排查成本。如果单个 Flow 加上条件分支就能解决问题就不值得引入多 Agent。只有当任务确实需要多个职责独立、需要并行推进且彼此交换信息的角色时才适合采用这种架构。多 Agent 的通信基石共享存储与消息队列多个 Agent 协同的通信机制通常用共享存储中的消息队列来实现。在 PocketFlow 中所有节点共享一个shared字典shared store而asyncio.Queue对象可以被放进这个shared字典中成为 Agent 之间天然的异步消息通道共享存储sharedAgent 之间交换的上下文、历史记录等状态放在同一个字典里。消息队列asyncio.QueueAgent 之间点对点投递消息。发送方await queue.put(msg)接收方await queue.get()阻塞等待天然支持异步非阻塞通信。这一模式与 PocketFlow 的异步核心抽象一脉相承。从 pocketflow/init.py 的源码可以看到AsyncNode提供了prep_async/exec_async/exec_fallback_async/post_async四个可覆写的异步钩子而AsyncFlow通过_orch_async循环驱动节点执行async def _orch_async(self, shared, paramsNone): curr, p, last_action copy.copy(self.start_node), (params or {**self.params}), None while curr: curr.set_params(p) last_action await curr._run_async(shared) if isinstance(curr, AsyncNode) else curr._run(shared) curr copy.copy(self.get_next_node(curr, last_action)) return last_action这里的last_action由节点post_async返回的字符串决定了下一步走向哪个后继节点这正是实现监听 → 处理 → 回环继续监听无限循环 Agent 的底层机制。官方文档 docs/core_abstraction/async.md 也明确指出post_async()的典型用途之一就是跨多 Agent 的协调coordinating across multi-agents。示例一用 asyncio.Queue 实现心跳消息监听 Agent先看一个最简单的多 Agent 通信骨架一个 Agent 节点不断从队列中取消息、处理、然后连接到自己形成监听循环同时一个独立的协程作为系统消息发送者周期性向队列投递消息。class AgentNode(AsyncNode): async def prep_async(self, _): message_queue self.params[messages] message await message_queue.get() print(fAgent received: {message}) return message # Create node and flow agent AgentNode() agent agent # connect to self flow AsyncFlow(startagent) # Create heartbeat sender async def send_system_messages(message_queue): counter 0 messages [ System status: all systems operational, Memory usage: normal, Network connectivity: stable, Processing load: optimal ] while True: message f{messages[counter % len(messages)]} | timestamp_{counter} await message_queue.put(message) counter 1 await asyncio.sleep(1) async def main(): message_queue asyncio.Queue() shared {} flow.set_params({messages: message_queue}) # Run both coroutines await asyncio.gather( flow.run_async(shared), send_system_messages(message_queue) ) asyncio.run(main())运行输出Agent received: System status: all systems operational | timestamp_0 Agent received: Memory usage: normal | timestamp_1 Agent received: Network connectivity: stable | timestamp_2 Agent received: Processing load: optimal | timestamp_3这个例子展示了多 Agent 通信的三个关键要素通过 params 注入队列flow.set_params({messages: message_queue})把队列传给流节点在prep_async中通过self.params[messages]取出。从源码看AsyncFlow._orch_async会在每轮循环调用curr.set_params(p)把参数注入当前节点。自连接形成监听循环agent agent利用BaseNode.__rshift__把节点连接到自己post_async返回的 action 让流回到同一节点形成取一条 → 处理一条 → 再取一条的持续监听循环。并发驱动asyncio.gather(flow.run_async(shared), send_system_messages(queue))让 Agent 流与消息发送者协程并行运行队列在这两者之间完成数据搬运。示例二Taboo 猜词游戏 —— 双 Agent 交互式实战接下来是官方文档中更复杂的交互式多 Agent 示例两个 Agent 玩文字猜词游戏 Taboo。AsyncHinter提示者负责给出避开禁用词、不超过 5 个词的提示AsyncGuesser猜词者根据提示猜目标词。两者通过两条消息队列往返通信猜对即结束游戏。class AsyncHinter(AsyncNode): async def prep_async(self, shared): guess await shared[hinter_queue].get() if guess GAME_OVER: return None return shared[target_word], shared[forbidden_words], shared.get(past_guesses, []) async def exec_async(self, inputs): if inputs is None: return None target, forbidden, past_guesses inputs prompt fGenerate hint for {target}\nForbidden words: {forbidden} if past_guesses: prompt f\nPrevious wrong guesses: {past_guesses}\nMake hint more specific. prompt \nUse at most 5 words. hint call_llm(prompt) print(f\nHinter: Heres your hint - {hint}) return hint async def post_async(self, shared, prep_res, exec_res): if exec_res is None: return end await shared[guesser_queue].put(exec_res) return continue class AsyncGuesser(AsyncNode): async def prep_async(self, shared): hint await shared[guesser_queue].get() return hint, shared.get(past_guesses, []) async def exec_async(self, inputs): hint, past_guesses inputs prompt fGiven hint: {hint}, past wrong guesses: {past_guesses}, make a new guess. Directly reply a single word: guess call_llm(prompt) print(fGuesser: I guess its - {guess}) return guess async def post_async(self, shared, prep_res, exec_res): if exec_res.lower() shared[target_word].lower(): print(Game Over - Correct guess!) await shared[hinter_queue].put(GAME_OVER) return end if past_guesses not in shared: shared[past_guesses] [] shared[past_guesses].append(exec_res) await shared[hinter_queue].put(exec_res) return continue async def main(): # Set up game shared { target_word: nostalgia, forbidden_words: [memory, past, remember, feeling, longing], hinter_queue: asyncio.Queue(), guesser_queue: asyncio.Queue() } print(Game starting!) print(fTarget word: {shared[target_word]}) print(fForbidden words: {shared[forbidden_words]}) # Initialize by sending empty guess to hinter await shared[hinter_queue].put() # Create nodes and flows hinter AsyncHinter() guesser AsyncGuesser() # Set up flows hinter_flow AsyncFlow(starthinter) guesser_flow AsyncFlow(startguesser) # Connect nodes to themselves hinter - continue hinter guesser - continue guesser # Run both agents concurrently await asyncio.gather( hinter_flow.run_async(shared), guesser_flow.run_async(shared) ) asyncio.run(main())运行输出Game starting! Target word: nostalgia Forbidden words: [memory, past, remember, feeling, longing] Hinter: Heres your hint - Thinking of childhood summer days Guesser: I guess its - popsicle Hinter: Heres your hint - When childhood cartoons make you emotional Guesser: I guess its - nostalgic Hinter: Heres your hint - When old songs move you Guesser: I guess its - memories Hinter: Heres your hint - That warm emotion about childhood Guesser: I guess its - nostalgia Game Over - Correct guess!逐环节拆解通信流程这个游戏本质上是两条单向队列拼成的双向通信管道初始化main先把一个空字符串投进hinter_queue相当于开始信号否则AsyncHinter.prep_async会永久阻塞在await queue.get()。Hinter 出题取出 guess首轮为空串后把target_word、forbidden_words、past_guesses组装成提示词 prompt调用 LLM 生成提示语post_async把提示语put进guesser_queue返回continue触发自连接回环。Guesser 猜词从guesser_queue拿到提示语结合历史错误猜测生成单个词的猜测若猜中向hinter_queue投递GAME_OVER终止信号并返回end结束自身循环若未猜中把猜测追加进shared[past_guesses]作为上下文供下一轮提示使用再把猜测投回hinter_queue返回continue继续循环。终止机制AsyncHinter.prep_async读到GAME_OVER时返回Noneexec_async/post_async随之短路返回endHinter 循环终止。两个 Flow 都结束时asyncio.gather返回。这里用到了 PocketFlow 的条件转移语法hinter - continue hinter。从 pocketflow/init.py 源码可以看到BaseNode.__sub__返回一个_ConditionalTransition其__rshift__最终调用src.next(tgt, action)把continue这个动作注册为指向自身的后继边def __sub__(self, action): if isinstance(action, str): return _ConditionalTransition(self, action) raise TypeError(Action must be a string) class _ConditionalTransition: def __init__(self, src, action): self.src, self.action src, action def __rshift__(self, tgt): return self.src.next(tgt, self.action)这也印证了post_async返回值action 字符串与转移边- action 之间的对应关系——这是 PocketFlow 中实现回环监听与状态机流转的统一机制官方测试 tests/test_async_flow.py 中的test_async_flow_branching等用例也验证了异步节点基于 action 分流的正确性。仓库内可直接运行的完整实现官方文档中的两个示例是精简的教学版仓库的 cookbook 目录下提供了完整的、可直接运行的 Taboo 游戏实现 cookbook/pocketflow-multi-agent包含 4 个文件文件作用main.py入口AsyncHinter/AsyncGuesser节点、双队列、双 AsyncFlow 并发驱动utils.pycall_llm()封装基于 OpenAI SDK 调用gpt-4o-minirequirements.txt依赖pocketflow0.0.1、openai1.0.0、pyyaml6.0README.md架构说明AsyncHinter -- MessageQueue -- AsyncGuesser与运行输出示例运行步骤pip install -r requirements.txt export OPENAI_API_KEYyour_api_key_here python main.py相比文档示例cookbook 版在几处做了实战化调整值得对照学习目标词与终止判断cookbook 使用nostalgic作为目标词并统一用.lower()做大小写不敏感匹配exec_res.lower() shared[target_word].lower()避免 LLM 输出大小写差异导致死循环。循环终止hinter - continue hinter与guesser - continue guesser两条自连接边让两个 Agent 各自循环直到猜中后经GAME_OVER信号与end动作双双退出。共享状态演进past_guesses通过shared.get(past_guesses, [])惰性初始化并持续累积成为 Hint 的上下文记忆——这与 Agent 设计模式 中强调的上下文管理要点一致给 LLM 提供相关且精简的上下文最近几次错误猜测而不是堆砌全部历史。仓库示例的运行输出也展示了一次完整对局Hinter 给出提示、Guesser 连续猜测、最终猜中Nostalgic后游戏自动结束可以作为验证程序正确性的对照基准。从设计模式到生产实践的要点综合官方文档 docs/design_pattern/multi_agent.md 与上述实现落地多 Agent 通信时有几点经验值得沉淀先简单后复杂能用一个 Flow 条件分支解决的不要引入多 Agent多 Agent 的价值在于职责解耦与并行推进代价是通信与协调复杂度。队列与共享状态职责分离asyncio.Queue负责消息流谁在什么时候说话shared字典负责状态共同记忆、上下文、最终结果。示例中past_guesses放在 shared、而即时消息走队列正是这一分工的体现。显式定义终止条件无限循环的监听 Agent 必须有明确的结束信号。Taboo 示例用GAME_OVER哨兵消息 endaction 双保险终止避免两个 Agent 永久空转。善用 AsyncNode 的三阶段钩子prep_async做阻塞式取消息I/O 友好、exec_async放 LLM 调用、post_async做跨 Agent 协调写回队列、判定 next action。这一阶段划分在 docs/core_abstraction/async.md 中有明确说明能有效隔离读数据 / 算 / 写状态三类职责。失败重试与容错多 Agent 场景中任何一方 LLM 调用失败都会卡住通信循环可结合 Node 的max_retries与wait参数如AsyncHinter(max_retries3, wait10)为exec_async增加重试与退避提升整体鲁棒性。从源码层面看整个机制最终都收敛到 pocketflow/init.py 中不足百行的核心实现AsyncNode._run_async依次执行prep_async → _exec → post_asyncAsyncFlow._orch_async依据返回的 action 驱动节点跳转。多 Agent 通信并不需要框架层面的任何特殊支持——asyncio.Queue塞进 shared 字典配合自连接边PocketFlow 的既有抽象就足以支撑任意复杂的 Agent 协作拓扑。赞分享人工智能大模型AI Agent工作流自动化RAG【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址https://gitcode.com/gh_mirrors/poc/PocketFlow点击查看免费下载相关推荐lamp-cloud消息队列实战基于RabbitMQ的异步通信方案lamp cloud消息队列实战基于RabbitMQ的异步通信方案 lamp cloud是一个功能强大的微服务架构平台其消息队列功能基于RabbitMQ实现后端微服务认证鉴权API网关终极指南如何利用Hyperswitch实现高可靠支付消息异步通信终极指南如何利用Hyperswitch实现高可靠支付消息异步通信 Hyperswitch作为一款功能强大的支付编排平台其内部集成了高效的消息队列系统专门用后端金融科技agents24 Agent Teams 实战指南七套多 Agent 通信消息模板与协议全解agents24 Agent Teams 实战指南七套多 Agent 通信消息模板与协议全解 本篇围绕 agents24 仓库中 agent teams 插件AI 插件AI 技能开发工具上一篇告别PDF安全焦虑Stirling-PDF如何用AES守护你的文档隐私下一篇从源码到应用7 Taskbar Tweaker核心功能实现原理创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考