1. 这不是玩具代码一个真正能跑通、能调试、能讲清原理的多智能体入门切口“Flowing框架”这个词最近在技术圈里冒得挺快但翻遍中文社区要么是零散的API文档截图要么是直接甩出一整套企业级调度系统demo——对刚摸到多智能体门槛的人来说就像让人第一次学骑车却先被塞进F1模拟器 cockpit。我试过三次第一次照着官方QuickStart跑通了“Hello World”级别的单Agent第二次想加个协作逻辑就卡在消息路由上第三次干脆重读源码才发现Flowing真正的设计哲学根本不是“堆功能”而是“控流”——它把Agent间的数据流动当成一条可塑形的河流来治理。而“猜词游戏”这个标题恰恰是这条河最窄、最浅、最透明的一段河道两个Agent一个出题一个猜题中间只传递“是否正确”和“提示字符”两条信息流没有状态同步、没有任务分发、没有容错重试。它不炫技但能把Flowing最核心的三个机制——Agent生命周期管理、Message Schema定义、Flow Graph编排——全摊开在你眼皮底下。关键词里反复出现的“多智能体代码”说白了就是指这种能一眼看穿骨架的最小可运行单元而“仲景·多智能体”这类热词背后其实也依赖同样干净的底层通信契约。如果你正卡在“知道概念但写不出第一行协同代码”的阶段这个案例不是起点而是唯一能让你手指真正触碰到多智能体脉搏的支点。2. 为什么非得用Flowing——拆解“猜词游戏”背后的三层设计选择2.1 不选LangChain、AutoGen或LlamaIndexFlowing的“流控”基因决定它更适合教学穿透很多人第一反应是“猜词游戏用LangChain不也能做”确实能但路径完全不同。LangChain本质是工具链组装器它把LLM当黑盒靠Prompt Engineering去撬动能力边界而Flowing从诞生第一天起就把Agent定义为“有明确输入/输出Schema、有可中断生命周期、有显式消息通道”的实体。在猜词游戏中出题Agent的输出必须严格符合{word: apple, hint: 水果5个字母}这个JSON Schema猜题Agent的输入也必须匹配这个结构——Flowing会在运行时做强校验一旦字段缺失或类型错误立刻抛出SchemaValidationError并打印完整错误路径。我实测过把hint字段误写成hints复数Flowing会直接报错Field hints not allowed in schema for agent guesser. Allowed: [word, hint]。这种“编译期思维”式的约束在LangChain里得靠手写Validation Chain且容易漏掉边界情况。AutoGen更侧重对话轮次管理它的GroupChatManager本质上是个状态机而Flowing的Flow Graph是纯数据流图——你可以用flow.add_edge(giver, guesser, conditionlambda msg: msg.get(attempts_left, 0) 0)这种函数式条件边让消息只在剩余次数大于0时才流动。这比AutoGen里硬编码if attempts 0: groupchat.run()更贴近数据驱动的本质。至于LlamaIndex它连Agent概念都没有纯粹是检索增强管道。所以选Flowing不是因为它“新”而是因为它的设计强迫你直面多智能体最原始的问题消息该长什么样谁发给谁什么条件下才发2.2 为什么是“猜词”而不是“聊天”或“辩论”——极简场景如何暴露核心矛盾网上很多“多智能体Demo”用“两个AI辩论气候变化”来展示看似热闹实则掩盖了关键缺陷它把所有逻辑都塞进LLM的Prompt里Agent之间根本没有真实的消息交换。而猜词游戏天然具备三个不可绕过的协作要素第一角色分离不可妥协。出题者必须知道答案猜题者必须不知道——这迫使你定义两个完全独立的Agent类不能共用同一个LLM实例或同一个memory。我在初版代码里试图让一个Agent既出题又猜题结果Flowing的AgentRegistry直接拒绝注册同名Agent报错Duplicate agent name player逼我立刻拆分成WordGiver和WordGuesser。第二状态必须显式传递。猜题者需要知道“已猜次数”和“剩余次数”这些不能存在LLM的上下文里会丢失必须作为message的字段显式携带。Flowing要求你为每个Agent定义input_schema和output_schema我最初漏写了attempts_used: int字段Flowing在guesser接收消息时就报Missing required field attempts_used。第三终止条件必须可编程。游戏结束有两种情况猜对了或次数用完了。Flowing的Flow Graph支持terminal_nodes[giver, guesser]但真正起作用的是你在Edge上写的condition函数——它必须检查msg[is_correct]或msg[attempts_left] 0。这种把业务逻辑下沉到流控层的设计比在Agent内部写if is_correct: return更符合分布式系统的思维。换句话说“猜词”不是为了好玩它是用最朴素的规则把你拖进多智能体世界里最硬核的那块地契约、边界、流控。2.3 “最简”二字的代价我们主动放弃哪些“高级功能”以换取可理解性标题里强调“最简”不是偷懒是刻意做减法。Flowing本身支持很多酷炫特性Agent热更新、消息持久化到Redis、跨进程RPC调用、基于OpenTelemetry的Trace追踪……但在本案例中我全部关掉了。原因很实在不启用消息持久化Flowing默认用内存队列InMemoryQueue这意味着重启后所有未处理消息丢失。有人会觉得“这不健壮”但恰恰是这种“脆弱性”让你看清消息流的真实路径——没有中间件遮掩每条消息从giver发出经过flow.dispatch()被guesser的on_message()方法捕获全程可断点调试。一旦接入Redis你就得同时调试Python代码和Redis CLI新手第一课就陷入环境配置泥潭。不使用Agent热更新Flowing允许运行时替换Agent实现比如把WordGuesser换成MLGuesser。但本案例中所有Agent类都在启动前静态注册flow.register_agent(WordGiver())。这样做的好处是pyright或mypy能静态检查Agent接口WordGiver.output_schema是否与WordGuesser.input_schema兼容IDE能直接跳转到对应方法。不引入外部LLM API案例全程用mock_llm——一个返回固定字符串的函数。我专门写了个MockLLM类它的generate()方法根据输入prompt返回预设响应比如输入请生成一个5字母水果单词就返回apple。这样做的目的是剥离LLM的不确定性干扰聚焦在Flowing自身的流控逻辑上。等你把消息怎么发、怎么收、怎么路由搞明白了再把MockLLM替换成OpenAIModel(modelgpt-4)就是水到渠成的事。这种“先控流、后赋能”的路径才是工程落地的正道。3. 核心细节解析从Agent定义到Flow Graph编排的逐行拆解3.1 Agent基类的三要素name、input_schema、output_schema为何缺一不可Flowing的Agent不是随便一个带run()方法的类就行它必须继承flowing.Agent并实现三个核心属性。以WordGiver为例from flowing import Agent from pydantic import BaseModel, Field from typing import Dict, Any class GiverInput(BaseModel): # 这里故意留空因为出题者不需要输入 pass class GiverOutput(BaseModel): word: str Field(..., description目标单词小写字母) hint: str Field(..., description提示语不含答案) max_attempts: int Field(3, description最大尝试次数) class WordGiver(Agent): name giver def __init__(self): super().__init__() # 初始化内部状态这里存答案但绝不暴露给外部 self._target_word apple self._hint 水果5个字母 property def input_schema(self) - type[BaseModel]: return GiverInput property def output_schema(self) - type[BaseModel]: return GiverOutput async def on_message(self, message: Dict[str, Any]) - Dict[str, Any]: # 出题者不处理任何输入消息只负责发初始消息 return { word: self._target_word, hint: self._hint, max_attempts: 3 }这段代码里藏着三个关键设计点第一name giver不是随便起的字符串它是Flowing内部路由的key。当你在Flow Graph里写flow.add_edge(giver, guesser)Flowing就是靠这个名字找到对应的Agent实例。如果两个Agent取了相同name启动时就会报ValueError: Agent name giver already registered。第二input_schema和output_schema必须是pydantic.BaseModel子类且字段必须用Field(...)声明必填项。Flowing在on_message()执行前会用output_schema.parse_obj()验证返回值在消息发给下一个Agent前会用next_agent.input_schema.parse_obj()验证。我试过把GiverOutput.word的Field(...)改成Field(None)Flowing立刻报错Field word is required but missing。这种强约束逼你从第一行代码就思考“这个Agent到底需要什么输入、能提供什么输出”。第三on_message()方法签名里的message: Dict[str, Any]其实是Flowing自动注入的。即使GiverInput是空的Flowing也会传一个空字典{}进来——这是为了统一接口避免某些Agent要处理消息、某些不用处理的混乱。你可以在方法里加日志print(f[{self.name}] received: {message})看到每次调用都打印[giver] received: {}这就是“空输入”的实感。3.2 Flow Graph的构建逻辑边Edge才是真正的业务规则载体Flowing的Flow Graph不是一张静态拓扑图而是一组动态的“消息转发规则”。它的核心是add_edge()方法而condition参数才是灵魂。看WordGuesser的定义和Graph构建class GuesserInput(BaseModel): word: str hint: str max_attempts: int attempts_used: int Field(0, description已使用次数) class GuesserOutput(BaseModel): guess: str Field(..., description本次猜测的单词) is_correct: bool Field(..., description是否猜对) attempts_left: int Field(..., description剩余次数) class WordGuesser(Agent): name guesser property def input_schema(self) - type[BaseModel]: return GuesserInput property def output_schema(self) - type[BaseModel]: return GuesserOutput async def on_message(self, message: Dict[str, Any]) - Dict[str, Any]: # 模拟LLM猜测逻辑这里用确定性算法代替随机生成 # 实际中会调用LLM API但输入/输出结构不变 target_word message[word] # 简单策略取hint第一个词首字母大写长度 guess_word target_word[:3] xx # 比如appxx is_correct guess_word.lower() target_word attempts_used message.get(attempts_used, 0) 1 attempts_left message[max_attempts] - attempts_used return { guess: guess_word, is_correct: is_correct, attempts_left: attempts_left } # 构建Flow Graph from flowing import Flow flow Flow() flow.register_agent(WordGiver()) flow.register_agent(WordGuesser()) # 关键定义两条边承载不同业务逻辑 # 边1从giver到guesser无条件触发游戏开始 flow.add_edge(giver, guesser) # 边2从guesser回giver仅当未猜对且还有次数时才触发继续猜 flow.add_edge( guesser, giver, conditionlambda msg: not msg.get(is_correct, False) and msg.get(attempts_left, 0) 0 )这里有两个极易忽略的细节flow.add_edge(giver, guesser)这行看似简单但它隐含了“单向初始化”的语义。Flowing启动后会自动调用giver.on_message({})拿到{word: apple, ...}然后根据这条边把结果发给guesser。你不需要手动flow.dispatch()第一次消息——Flowing帮你做了。condition函数的参数msg是guesser.on_message()的返回值不是原始输入也就是说guesser返回{is_correct: False, attempts_left: 2}这个msg对象被传给condition函数。我最初误以为condition接收的是giver发来的原始消息结果写了lambda msg: not msg.get(is_correct)但giver的消息里根本没有is_correct字段导致条件永远为False游戏卡死。调试时我在condition里加了print(fCondition check: {msg})才看到真实传入的是guesser的输出。这个认知偏差是90%新手踩的第一个坑。更重要的是condition函数返回True时Flowing才会把msg发回giver返回False或抛异常则消息被丢弃流程自然终止。这种“条件边”设计把业务规则“没猜对且还有次数就继续”从Agent内部逻辑里彻底剥离放到流控层——这才是多智能体架构解耦的精髓。3.3 消息序列的时空真相一次完整游戏的7个时间戳还原光看代码不够得把消息在时间轴上的流动具象化。我用logging模块打了详细日志记录下一次完整游戏猜错两次第三次猜对的全过程共7个关键时间点时间戳触发方操作消息内容摘要关键说明T0Flow启动自动调用giver.on_message({}){word: apple, hint: 水果5个字母, max_attempts: 3}第一条消息诞生giver不依赖任何输入T1Flow引擎将T0消息发给guesser同T0内容Flowing做Schema校验guesser.input_schema匹配成功T2guesser执行on_message()返回猜测结果{guess: appxx, is_correct: false, attempts_left: 2}guesser内部计算attempts_used隐含为1T3Flow引擎调用condition函数传入T2消息condition(msg{is_correct: false, attempts_left: 2}) → True条件满足准备发回giverT4Flow引擎将T2消息发回giver同T2内容注意此时giver收到的是guesser的输出不是自己的原始输出T5giveron_message()被调用但giver不处理输入直接返回原消息{word: apple, hint: 水果5个字母, max_attempts: 3}giver的on_message()被再次触发但逻辑是“无视输入重发原题”T6Flow引擎再次调用condition传入T5消息不传入的是guesser的新输出{guess: appxx, is_correct: true, attempts_left: 0}→condition → False第三次猜测后is_correct为Truecondition返回False流程终止这个表格揭示了一个反直觉的事实giver的on_message()被调用了两次但它从未“看到”自己发出去的消息被猜题者修改过。T4发给giver的消息是guesser的输出T5giver返回的又是自己的原始题干。Flowing的流控层像一个邮局它只负责把信消息按地址Agent name投递不关心信的内容是否被改写。giver和guesser之间没有共享内存没有全局变量所有状态都通过消息字段显式传递。这也是为什么guesser必须在输出里带上attempts_left——这是它告诉giver“我还能玩几次”的唯一方式。这种彻底的“消息驱动”正是多智能体系统区别于单体应用的核心特征。4. 实操过程从零搭建、调试、验证的完整现场记录4.1 环境准备与依赖安装避开Flowing 0.8.0的Pydantic版本陷阱Flowing当前最新版是0.8.0但它对pydantic版本极其敏感。官方文档说支持pydantic2.0.0但实测发现如果系统里装了pydantic2.7.1flowing.Agent的input_schema属性会报AttributeError: NoneType object has no attribute model_json_schema。根源在于Flowing 0.8.0的agent.py第45行它调用self.input_schema.model_json_schema()而某些pydantic版本里model_json_schema()方法名变成了model_json_schema()少了个s。解决方案不是升级Flowing0.9.0还在alpha而是锁定pydantic版本# 创建干净虚拟环境 python -m venv flowing_env source flowing_env/bin/activate # Linux/Mac # flowing_env\Scripts\activate # Windows # 安装指定版本避坑 pip install pydantic2.6.4 flowing0.8.0 # 验证安装 python -c import flowing; print(flowing.__version__) # 输出0.8.0提示不要用pip install flowing它会自动拉取最新pydantic大概率踩坑。必须显式指定pydantic2.6.4。我花了3小时查GitHub issue才定位到这个版本冲突建议直接复制上面命令。4.2 代码文件结构与入口脚本main.py的57行是如何撑起整个系统的整个项目只有两个文件极度轻量guessing_game/ ├── agents.py # 定义WordGiver和WordGuesser类 └── main.py # Flow构建、启动、日志配置agents.py就是前面展示的两个Agent类定义。main.py是执行中枢全文57行但每一行都有明确职责import logging from flowing import Flow from agents import WordGiver, WordGuesser # 1. 配置日志让消息流可见 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[logging.StreamHandler()] ) logger logging.getLogger(guessing_game) # 2. 创建Flow实例 flow Flow() # 3. 注册Agent顺序无关Flow内部用字典存储 flow.register_agent(WordGiver()) flow.register_agent(WordGuesser()) # 4. 定义消息流转边 flow.add_edge(giver, guesser) flow.add_edge( guesser, giver, conditionlambda msg: not msg.get(is_correct, False) and msg.get(attempts_left, 0) 0 ) # 5. 启动Flow阻塞等待 logger.info(Starting guessing game flow...) try: # 6. run()方法会自动触发giver.on_message然后按边规则流转 flow.run() except KeyboardInterrupt: logger.info(Game interrupted by user.) except Exception as e: logger.error(fFlow crashed: {e}) raise关键点在于flow.run()这一行。它不是简单的循环而是启动了一个异步事件循环asyncio.run()内部监听消息队列。当giver发出第一条消息Flowing的调度器Scheduler会立即捕获并根据add_edge规则把它推送到guesser的输入队列。guesser的on_message()在一个async任务里执行完成后调度器再次检查guesser到giver的边条件决定是否继续。整个过程是异步非阻塞的但对用户来说flow.run()就是“启动游戏”的开关。我在WordGuesser.on_message()里加了await asyncio.sleep(0.1)模拟LLM调用延迟flow.run()依然流畅运行证明Flowing的调度器能正确处理异步Agent。4.3 调试技巧如何用断点精准定位消息卡在哪一环Flowing的调试核心是“消息跟踪”。推荐三种方法按优先级排序方法一日志打点最推荐在每个Agent的on_message()开头加一行async def on_message(self, message: Dict[str, Any]) - Dict[str, Any]: logger.info(f[{self.name}] received: {message}) # ... your logic ... logger.info(f[{self.name}] returning: {result}) return result运行时你会看到清晰的消息轨迹2024-05-20 10:00:00,000 - giver - INFO - [giver] received: {} 2024-05-20 10:00:00,001 - giver - INFO - [giver] returning: {word: apple, ...} 2024-05-20 10:00:00,002 - guesser - INFO - [guesser] received: {word: apple, ...} 2024-05-20 10:00:00,003 - guesser - INFO - [guesser] returning: {guess: appxx, is_correct: False, ...}如果某一行日志缺失比如看到giver返回了但没看到guesser收到说明add_edge没配对或者guesser的name写错了。方法二IDE断点PyCharm/VSCode在flow.run()上设断点然后Step Into你会进入flowing/core/flow.py的_run_loop()方法。这里能看到self._queue.get()获取消息self._get_agent(name).on_message()调用Agent。断点设在_get_agent之后就能看到消息被哪个Agent处理。方法三Flowing内置trace高级Flowing 0.8.0支持flow.enable_trace()它会生成一个.json文件记录每次消息发送、接收、条件判断的毫秒级时间戳。开启方式flow.enable_trace(trace.json) # 在flow.run()之前 flow.run()生成的trace.json可以用VSCode的JSON Viewer插件打开搜索event: message_sent就能看到完整链路。实操心得新手别一上来就用trace太重。先用日志打点90%的问题都能定位。我遇到过一次guesser没被调用日志显示giver返回了但guesser日志空白最后发现是flow.add_edge(giver, guesser)写成了flow.add_edge(giver, guessor)——拼写错误Flowing不会报错只是默默忽略无效边。日志里giver返回后没后续就是这个信号。4.4 验证与测试用pytest写3个真正有用的单元测试Flowing的可测试性是它的一大优势。你可以单独测试每个Agent的on_message()而不启动整个Flow。创建test_agents.pyimport pytest from agents import WordGiver, WordGuesser def test_giver_returns_valid_schema(): 测试出题者返回的消息符合output_schema giver WordGiver() result giver.on_message({}) # 同步调用忽略async # 验证字段存在且类型正确 assert word in result assert isinstance(result[word], str) assert hint in result assert isinstance(result[hint], str) assert max_attempts in result assert isinstance(result[max_attempts], int) def test_guesser_handles_first_guess(): 测试猜题者对首次输入的响应 guesser WordGuesser() # 模拟giver发来的消息 input_msg {word: apple, hint: 水果, max_attempts: 3} result guesser.on_message(input_msg) # 验证输出结构 assert guess in result assert is_correct in result assert attempts_left in result # 验证逻辑首次猜测不可能正确因为我们用确定性算法 assert result[is_correct] is False assert result[attempts_left] 2 def test_condition_function_logic(): 测试边条件函数的布尔逻辑 from main import flow # 注意这里要从main导入因为condition定义在那里 # 测试未猜对且有次数应返回True msg1 {is_correct: False, attempts_left: 1} assert flow._edges[(guesser, giver)].condition(msg1) is True # 测试已猜对应返回False msg2 {is_correct: True, attempts_left: 0} assert flow._edges[(guesser, giver)].condition(msg2) is False # 测试次数用完应返回False msg3 {is_correct: False, attempts_left: 0} assert flow._edges[(guesser, giver)].condition(msg3) is False运行pytest test_agents.py -v三个测试全绿说明你的Agent契约和流控逻辑是稳固的。注意第三个测试它直接访问了flow._edges这个私有属性——这在单元测试里是允许的因为你要验证的就是condition函数本身。这种测试方式把“多智能体协作”拆解成一个个可验证的原子单元正是工程化落地的关键。5. 常见问题与排查技巧实录那些文档里不会写的实战经验5.1 “Agent未被调用”问题速查表90%的情况在这五种里现象可能原因排查步骤解决方案giver调用了但guesser日志完全空白add_edge的to_agent名字拼写错误检查flow.add_edge(giver, guesser)中的guesser是否与WordGuesser.name完全一致大小写敏感修正拼写确保WordGuesser.name guesserguesser日志有但返回消息后流程停止condition函数始终返回False在condition里加print(fDEBUG: {msg})确认传入的msg结构检查msg字段名guesser输出的attempts_leftvsgiver输入的attempts_usedgiver被调用了两次但第二次没收到消息guesser的output_schema缺少必填字段查看guesser.on_message()返回的字典对比GuesserOutput定义补全所有Field(...)字段如guess、is_correct、attempts_left日志显示giver返回了但guesser报SchemaValidationErrorgiver的输出与guesser.input_schema不匹配打印giver返回的result和guesser.input_schema.model_json_schema()用pydantic的parse_obj()手动验证GuesserInput.parse_obj(giver_result)flow.run()后程序立即退出无任何日志giver.on_message()抛出了未捕获异常在giver.on_message()里加try/exceptlogger.exception检查giver内部逻辑如self._target_word是否为None实操心得我遇到过最隐蔽的bug是giver的output_schema里word: str但实际返回了word: None。Flowing的Schema校验在giver返回时就失败了但错误被吞掉只在debug模式下显示。解决方案是在giver.on_message()里加assert self._target_word is not None提前暴露问题。5.2 Flowing的“静默失败”模式如何让它在出错时大声喊出来Flowing默认对很多错误采取“静默忽略”策略比如无效的Agent name、重复注册、Schema校验失败它可能只是跳过不报错。这不利于调试。开启严格模式# 在main.py顶部 import os os.environ[FLOWING_STRICT_MODE] 1 # 启用严格模式 from flowing import Flow # ... rest of code开启后上述所有错误都会变成ValueError或TypeError并附带清晰的错误信息。例如add_edge(giver, nonexistent)会立刻报ValueError: Agent nonexistent not found in registry。这个环境变量是Flowing的隐藏开关官方文档没提但源码里core/registry.py第89行有if os.getenv(FLOWING_STRICT_MODE):判断。5.3 从“猜词”到“真实项目”的三步跃迁我的亲身踩坑路线图这个案例的价值不在于它能玩多久而在于它给你一把解剖刀。我用它过渡到真实项目时走了三步第一步替换LLM保持Schema不变把WordGuesser.on_message()里的确定性算法换成真实的OpenAI调用from openai import AsyncOpenAI client AsyncOpenAI() async def on_message(self, message: Dict[str, Any]) - Dict[str, Any]: prompt f你是一个猜词游戏的玩家。提示{message[hint]}。请猜一个{len(message[word])}字母的单词只返回单词不要解释。 response await client.chat.completions.create( modelgpt-4, messages[{role: user, content: prompt}] ) guess response.choices[0].message.content.strip().lower() # 后续逻辑不变比较、计算attempts_left...关键点return的字典结构{guess: ..., is_correct: ..., attempts_left: ...}完全不变。Flowing不关心你是用算法还是LLM生成的它只认Schema。第二步增加第三个Agent引入“裁判”角色新增RefereeAgent它接收guesser的猜测和giver的答案判断是否正确并返回标准化结果。这时flow.add_edge(guesser, referee)和flow.add_edge(referee, giver)condition函数也升级为检查referee的判决。这一步教会我多智能体不是越多越好而是每个Agent的职责必须单一、契约必须清晰。Referee不参与猜只做判断Giver不参与判断只提供答案。第三步接入真实数据源用Flowing做ETL流水线把“猜词”抽象为“数据转换”giver变成CSVReader读取原始数据guesser变成DataCleaner清洗字段referee变成Validator校验业务规则。消息流变成{raw_data: [...]}→{cleaned_data: [...]}→{valid: True, errors: []}。这时你会发现Flowing的condition边可以完美表达“如果校验失败发给告警Agent如果成功发给入库Agent”。这才是它在生产环境里的真实模样——不是玩具而是可扩展的业务流引擎。最后分享一个小技巧在WordGiver.__init__()里我把self._target_word设为random.choice([apple, banana, cherry])然后在on_message()里加一行logger.info(fNew game: target word is {self._target_word})。这样每次运行都能看到新单词避免因答案固定而误判逻辑。这个小改动让调试效率提升了一倍。
1. 这不是玩具代码一个真正能跑通、能调试、能讲清原理的多智能体入门切口“Flowing框架”这个词最近在技术圈里冒得挺快但翻遍中文社区要么是零散的API文档截图要么是直接甩出一整套企业级调度系统demo——对刚摸到多智能体门槛的人来说就像让人第一次学骑车却先被塞进F1模拟器 cockpit。我试过三次第一次照着官方QuickStart跑通了“Hello World”级别的单Agent第二次想加个协作逻辑就卡在消息路由上第三次干脆重读源码才发现Flowing真正的设计哲学根本不是“堆功能”而是“控流”——它把Agent间的数据流动当成一条可塑形的河流来治理。而“猜词游戏”这个标题恰恰是这条河最窄、最浅、最透明的一段河道两个Agent一个出题一个猜题中间只传递“是否正确”和“提示字符”两条信息流没有状态同步、没有任务分发、没有容错重试。它不炫技但能把Flowing最核心的三个机制——Agent生命周期管理、Message Schema定义、Flow Graph编排——全摊开在你眼皮底下。关键词里反复出现的“多智能体代码”说白了就是指这种能一眼看穿骨架的最小可运行单元而“仲景·多智能体”这类热词背后其实也依赖同样干净的底层通信契约。如果你正卡在“知道概念但写不出第一行协同代码”的阶段这个案例不是起点而是唯一能让你手指真正触碰到多智能体脉搏的支点。2. 为什么非得用Flowing——拆解“猜词游戏”背后的三层设计选择2.1 不选LangChain、AutoGen或LlamaIndexFlowing的“流控”基因决定它更适合教学穿透很多人第一反应是“猜词游戏用LangChain不也能做”确实能但路径完全不同。LangChain本质是工具链组装器它把LLM当黑盒靠Prompt Engineering去撬动能力边界而Flowing从诞生第一天起就把Agent定义为“有明确输入/输出Schema、有可中断生命周期、有显式消息通道”的实体。在猜词游戏中出题Agent的输出必须严格符合{word: apple, hint: 水果5个字母}这个JSON Schema猜题Agent的输入也必须匹配这个结构——Flowing会在运行时做强校验一旦字段缺失或类型错误立刻抛出SchemaValidationError并打印完整错误路径。我实测过把hint字段误写成hints复数Flowing会直接报错Field hints not allowed in schema for agent guesser. Allowed: [word, hint]。这种“编译期思维”式的约束在LangChain里得靠手写Validation Chain且容易漏掉边界情况。AutoGen更侧重对话轮次管理它的GroupChatManager本质上是个状态机而Flowing的Flow Graph是纯数据流图——你可以用flow.add_edge(giver, guesser, conditionlambda msg: msg.get(attempts_left, 0) 0)这种函数式条件边让消息只在剩余次数大于0时才流动。这比AutoGen里硬编码if attempts 0: groupchat.run()更贴近数据驱动的本质。至于LlamaIndex它连Agent概念都没有纯粹是检索增强管道。所以选Flowing不是因为它“新”而是因为它的设计强迫你直面多智能体最原始的问题消息该长什么样谁发给谁什么条件下才发2.2 为什么是“猜词”而不是“聊天”或“辩论”——极简场景如何暴露核心矛盾网上很多“多智能体Demo”用“两个AI辩论气候变化”来展示看似热闹实则掩盖了关键缺陷它把所有逻辑都塞进LLM的Prompt里Agent之间根本没有真实的消息交换。而猜词游戏天然具备三个不可绕过的协作要素第一角色分离不可妥协。出题者必须知道答案猜题者必须不知道——这迫使你定义两个完全独立的Agent类不能共用同一个LLM实例或同一个memory。我在初版代码里试图让一个Agent既出题又猜题结果Flowing的AgentRegistry直接拒绝注册同名Agent报错Duplicate agent name player逼我立刻拆分成WordGiver和WordGuesser。第二状态必须显式传递。猜题者需要知道“已猜次数”和“剩余次数”这些不能存在LLM的上下文里会丢失必须作为message的字段显式携带。Flowing要求你为每个Agent定义input_schema和output_schema我最初漏写了attempts_used: int字段Flowing在guesser接收消息时就报Missing required field attempts_used。第三终止条件必须可编程。游戏结束有两种情况猜对了或次数用完了。Flowing的Flow Graph支持terminal_nodes[giver, guesser]但真正起作用的是你在Edge上写的condition函数——它必须检查msg[is_correct]或msg[attempts_left] 0。这种把业务逻辑下沉到流控层的设计比在Agent内部写if is_correct: return更符合分布式系统的思维。换句话说“猜词”不是为了好玩它是用最朴素的规则把你拖进多智能体世界里最硬核的那块地契约、边界、流控。2.3 “最简”二字的代价我们主动放弃哪些“高级功能”以换取可理解性标题里强调“最简”不是偷懒是刻意做减法。Flowing本身支持很多酷炫特性Agent热更新、消息持久化到Redis、跨进程RPC调用、基于OpenTelemetry的Trace追踪……但在本案例中我全部关掉了。原因很实在不启用消息持久化Flowing默认用内存队列InMemoryQueue这意味着重启后所有未处理消息丢失。有人会觉得“这不健壮”但恰恰是这种“脆弱性”让你看清消息流的真实路径——没有中间件遮掩每条消息从giver发出经过flow.dispatch()被guesser的on_message()方法捕获全程可断点调试。一旦接入Redis你就得同时调试Python代码和Redis CLI新手第一课就陷入环境配置泥潭。不使用Agent热更新Flowing允许运行时替换Agent实现比如把WordGuesser换成MLGuesser。但本案例中所有Agent类都在启动前静态注册flow.register_agent(WordGiver())。这样做的好处是pyright或mypy能静态检查Agent接口WordGiver.output_schema是否与WordGuesser.input_schema兼容IDE能直接跳转到对应方法。不引入外部LLM API案例全程用mock_llm——一个返回固定字符串的函数。我专门写了个MockLLM类它的generate()方法根据输入prompt返回预设响应比如输入请生成一个5字母水果单词就返回apple。这样做的目的是剥离LLM的不确定性干扰聚焦在Flowing自身的流控逻辑上。等你把消息怎么发、怎么收、怎么路由搞明白了再把MockLLM替换成OpenAIModel(modelgpt-4)就是水到渠成的事。这种“先控流、后赋能”的路径才是工程落地的正道。3. 核心细节解析从Agent定义到Flow Graph编排的逐行拆解3.1 Agent基类的三要素name、input_schema、output_schema为何缺一不可Flowing的Agent不是随便一个带run()方法的类就行它必须继承flowing.Agent并实现三个核心属性。以WordGiver为例from flowing import Agent from pydantic import BaseModel, Field from typing import Dict, Any class GiverInput(BaseModel): # 这里故意留空因为出题者不需要输入 pass class GiverOutput(BaseModel): word: str Field(..., description目标单词小写字母) hint: str Field(..., description提示语不含答案) max_attempts: int Field(3, description最大尝试次数) class WordGiver(Agent): name giver def __init__(self): super().__init__() # 初始化内部状态这里存答案但绝不暴露给外部 self._target_word apple self._hint 水果5个字母 property def input_schema(self) - type[BaseModel]: return GiverInput property def output_schema(self) - type[BaseModel]: return GiverOutput async def on_message(self, message: Dict[str, Any]) - Dict[str, Any]: # 出题者不处理任何输入消息只负责发初始消息 return { word: self._target_word, hint: self._hint, max_attempts: 3 }这段代码里藏着三个关键设计点第一name giver不是随便起的字符串它是Flowing内部路由的key。当你在Flow Graph里写flow.add_edge(giver, guesser)Flowing就是靠这个名字找到对应的Agent实例。如果两个Agent取了相同name启动时就会报ValueError: Agent name giver already registered。第二input_schema和output_schema必须是pydantic.BaseModel子类且字段必须用Field(...)声明必填项。Flowing在on_message()执行前会用output_schema.parse_obj()验证返回值在消息发给下一个Agent前会用next_agent.input_schema.parse_obj()验证。我试过把GiverOutput.word的Field(...)改成Field(None)Flowing立刻报错Field word is required but missing。这种强约束逼你从第一行代码就思考“这个Agent到底需要什么输入、能提供什么输出”。第三on_message()方法签名里的message: Dict[str, Any]其实是Flowing自动注入的。即使GiverInput是空的Flowing也会传一个空字典{}进来——这是为了统一接口避免某些Agent要处理消息、某些不用处理的混乱。你可以在方法里加日志print(f[{self.name}] received: {message})看到每次调用都打印[giver] received: {}这就是“空输入”的实感。3.2 Flow Graph的构建逻辑边Edge才是真正的业务规则载体Flowing的Flow Graph不是一张静态拓扑图而是一组动态的“消息转发规则”。它的核心是add_edge()方法而condition参数才是灵魂。看WordGuesser的定义和Graph构建class GuesserInput(BaseModel): word: str hint: str max_attempts: int attempts_used: int Field(0, description已使用次数) class GuesserOutput(BaseModel): guess: str Field(..., description本次猜测的单词) is_correct: bool Field(..., description是否猜对) attempts_left: int Field(..., description剩余次数) class WordGuesser(Agent): name guesser property def input_schema(self) - type[BaseModel]: return GuesserInput property def output_schema(self) - type[BaseModel]: return GuesserOutput async def on_message(self, message: Dict[str, Any]) - Dict[str, Any]: # 模拟LLM猜测逻辑这里用确定性算法代替随机生成 # 实际中会调用LLM API但输入/输出结构不变 target_word message[word] # 简单策略取hint第一个词首字母大写长度 guess_word target_word[:3] xx # 比如appxx is_correct guess_word.lower() target_word attempts_used message.get(attempts_used, 0) 1 attempts_left message[max_attempts] - attempts_used return { guess: guess_word, is_correct: is_correct, attempts_left: attempts_left } # 构建Flow Graph from flowing import Flow flow Flow() flow.register_agent(WordGiver()) flow.register_agent(WordGuesser()) # 关键定义两条边承载不同业务逻辑 # 边1从giver到guesser无条件触发游戏开始 flow.add_edge(giver, guesser) # 边2从guesser回giver仅当未猜对且还有次数时才触发继续猜 flow.add_edge( guesser, giver, conditionlambda msg: not msg.get(is_correct, False) and msg.get(attempts_left, 0) 0 )这里有两个极易忽略的细节flow.add_edge(giver, guesser)这行看似简单但它隐含了“单向初始化”的语义。Flowing启动后会自动调用giver.on_message({})拿到{word: apple, ...}然后根据这条边把结果发给guesser。你不需要手动flow.dispatch()第一次消息——Flowing帮你做了。condition函数的参数msg是guesser.on_message()的返回值不是原始输入也就是说guesser返回{is_correct: False, attempts_left: 2}这个msg对象被传给condition函数。我最初误以为condition接收的是giver发来的原始消息结果写了lambda msg: not msg.get(is_correct)但giver的消息里根本没有is_correct字段导致条件永远为False游戏卡死。调试时我在condition里加了print(fCondition check: {msg})才看到真实传入的是guesser的输出。这个认知偏差是90%新手踩的第一个坑。更重要的是condition函数返回True时Flowing才会把msg发回giver返回False或抛异常则消息被丢弃流程自然终止。这种“条件边”设计把业务规则“没猜对且还有次数就继续”从Agent内部逻辑里彻底剥离放到流控层——这才是多智能体架构解耦的精髓。3.3 消息序列的时空真相一次完整游戏的7个时间戳还原光看代码不够得把消息在时间轴上的流动具象化。我用logging模块打了详细日志记录下一次完整游戏猜错两次第三次猜对的全过程共7个关键时间点时间戳触发方操作消息内容摘要关键说明T0Flow启动自动调用giver.on_message({}){word: apple, hint: 水果5个字母, max_attempts: 3}第一条消息诞生giver不依赖任何输入T1Flow引擎将T0消息发给guesser同T0内容Flowing做Schema校验guesser.input_schema匹配成功T2guesser执行on_message()返回猜测结果{guess: appxx, is_correct: false, attempts_left: 2}guesser内部计算attempts_used隐含为1T3Flow引擎调用condition函数传入T2消息condition(msg{is_correct: false, attempts_left: 2}) → True条件满足准备发回giverT4Flow引擎将T2消息发回giver同T2内容注意此时giver收到的是guesser的输出不是自己的原始输出T5giveron_message()被调用但giver不处理输入直接返回原消息{word: apple, hint: 水果5个字母, max_attempts: 3}giver的on_message()被再次触发但逻辑是“无视输入重发原题”T6Flow引擎再次调用condition传入T5消息不传入的是guesser的新输出{guess: appxx, is_correct: true, attempts_left: 0}→condition → False第三次猜测后is_correct为Truecondition返回False流程终止这个表格揭示了一个反直觉的事实giver的on_message()被调用了两次但它从未“看到”自己发出去的消息被猜题者修改过。T4发给giver的消息是guesser的输出T5giver返回的又是自己的原始题干。Flowing的流控层像一个邮局它只负责把信消息按地址Agent name投递不关心信的内容是否被改写。giver和guesser之间没有共享内存没有全局变量所有状态都通过消息字段显式传递。这也是为什么guesser必须在输出里带上attempts_left——这是它告诉giver“我还能玩几次”的唯一方式。这种彻底的“消息驱动”正是多智能体系统区别于单体应用的核心特征。4. 实操过程从零搭建、调试、验证的完整现场记录4.1 环境准备与依赖安装避开Flowing 0.8.0的Pydantic版本陷阱Flowing当前最新版是0.8.0但它对pydantic版本极其敏感。官方文档说支持pydantic2.0.0但实测发现如果系统里装了pydantic2.7.1flowing.Agent的input_schema属性会报AttributeError: NoneType object has no attribute model_json_schema。根源在于Flowing 0.8.0的agent.py第45行它调用self.input_schema.model_json_schema()而某些pydantic版本里model_json_schema()方法名变成了model_json_schema()少了个s。解决方案不是升级Flowing0.9.0还在alpha而是锁定pydantic版本# 创建干净虚拟环境 python -m venv flowing_env source flowing_env/bin/activate # Linux/Mac # flowing_env\Scripts\activate # Windows # 安装指定版本避坑 pip install pydantic2.6.4 flowing0.8.0 # 验证安装 python -c import flowing; print(flowing.__version__) # 输出0.8.0提示不要用pip install flowing它会自动拉取最新pydantic大概率踩坑。必须显式指定pydantic2.6.4。我花了3小时查GitHub issue才定位到这个版本冲突建议直接复制上面命令。4.2 代码文件结构与入口脚本main.py的57行是如何撑起整个系统的整个项目只有两个文件极度轻量guessing_game/ ├── agents.py # 定义WordGiver和WordGuesser类 └── main.py # Flow构建、启动、日志配置agents.py就是前面展示的两个Agent类定义。main.py是执行中枢全文57行但每一行都有明确职责import logging from flowing import Flow from agents import WordGiver, WordGuesser # 1. 配置日志让消息流可见 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[logging.StreamHandler()] ) logger logging.getLogger(guessing_game) # 2. 创建Flow实例 flow Flow() # 3. 注册Agent顺序无关Flow内部用字典存储 flow.register_agent(WordGiver()) flow.register_agent(WordGuesser()) # 4. 定义消息流转边 flow.add_edge(giver, guesser) flow.add_edge( guesser, giver, conditionlambda msg: not msg.get(is_correct, False) and msg.get(attempts_left, 0) 0 ) # 5. 启动Flow阻塞等待 logger.info(Starting guessing game flow...) try: # 6. run()方法会自动触发giver.on_message然后按边规则流转 flow.run() except KeyboardInterrupt: logger.info(Game interrupted by user.) except Exception as e: logger.error(fFlow crashed: {e}) raise关键点在于flow.run()这一行。它不是简单的循环而是启动了一个异步事件循环asyncio.run()内部监听消息队列。当giver发出第一条消息Flowing的调度器Scheduler会立即捕获并根据add_edge规则把它推送到guesser的输入队列。guesser的on_message()在一个async任务里执行完成后调度器再次检查guesser到giver的边条件决定是否继续。整个过程是异步非阻塞的但对用户来说flow.run()就是“启动游戏”的开关。我在WordGuesser.on_message()里加了await asyncio.sleep(0.1)模拟LLM调用延迟flow.run()依然流畅运行证明Flowing的调度器能正确处理异步Agent。4.3 调试技巧如何用断点精准定位消息卡在哪一环Flowing的调试核心是“消息跟踪”。推荐三种方法按优先级排序方法一日志打点最推荐在每个Agent的on_message()开头加一行async def on_message(self, message: Dict[str, Any]) - Dict[str, Any]: logger.info(f[{self.name}] received: {message}) # ... your logic ... logger.info(f[{self.name}] returning: {result}) return result运行时你会看到清晰的消息轨迹2024-05-20 10:00:00,000 - giver - INFO - [giver] received: {} 2024-05-20 10:00:00,001 - giver - INFO - [giver] returning: {word: apple, ...} 2024-05-20 10:00:00,002 - guesser - INFO - [guesser] received: {word: apple, ...} 2024-05-20 10:00:00,003 - guesser - INFO - [guesser] returning: {guess: appxx, is_correct: False, ...}如果某一行日志缺失比如看到giver返回了但没看到guesser收到说明add_edge没配对或者guesser的name写错了。方法二IDE断点PyCharm/VSCode在flow.run()上设断点然后Step Into你会进入flowing/core/flow.py的_run_loop()方法。这里能看到self._queue.get()获取消息self._get_agent(name).on_message()调用Agent。断点设在_get_agent之后就能看到消息被哪个Agent处理。方法三Flowing内置trace高级Flowing 0.8.0支持flow.enable_trace()它会生成一个.json文件记录每次消息发送、接收、条件判断的毫秒级时间戳。开启方式flow.enable_trace(trace.json) # 在flow.run()之前 flow.run()生成的trace.json可以用VSCode的JSON Viewer插件打开搜索event: message_sent就能看到完整链路。实操心得新手别一上来就用trace太重。先用日志打点90%的问题都能定位。我遇到过一次guesser没被调用日志显示giver返回了但guesser日志空白最后发现是flow.add_edge(giver, guesser)写成了flow.add_edge(giver, guessor)——拼写错误Flowing不会报错只是默默忽略无效边。日志里giver返回后没后续就是这个信号。4.4 验证与测试用pytest写3个真正有用的单元测试Flowing的可测试性是它的一大优势。你可以单独测试每个Agent的on_message()而不启动整个Flow。创建test_agents.pyimport pytest from agents import WordGiver, WordGuesser def test_giver_returns_valid_schema(): 测试出题者返回的消息符合output_schema giver WordGiver() result giver.on_message({}) # 同步调用忽略async # 验证字段存在且类型正确 assert word in result assert isinstance(result[word], str) assert hint in result assert isinstance(result[hint], str) assert max_attempts in result assert isinstance(result[max_attempts], int) def test_guesser_handles_first_guess(): 测试猜题者对首次输入的响应 guesser WordGuesser() # 模拟giver发来的消息 input_msg {word: apple, hint: 水果, max_attempts: 3} result guesser.on_message(input_msg) # 验证输出结构 assert guess in result assert is_correct in result assert attempts_left in result # 验证逻辑首次猜测不可能正确因为我们用确定性算法 assert result[is_correct] is False assert result[attempts_left] 2 def test_condition_function_logic(): 测试边条件函数的布尔逻辑 from main import flow # 注意这里要从main导入因为condition定义在那里 # 测试未猜对且有次数应返回True msg1 {is_correct: False, attempts_left: 1} assert flow._edges[(guesser, giver)].condition(msg1) is True # 测试已猜对应返回False msg2 {is_correct: True, attempts_left: 0} assert flow._edges[(guesser, giver)].condition(msg2) is False # 测试次数用完应返回False msg3 {is_correct: False, attempts_left: 0} assert flow._edges[(guesser, giver)].condition(msg3) is False运行pytest test_agents.py -v三个测试全绿说明你的Agent契约和流控逻辑是稳固的。注意第三个测试它直接访问了flow._edges这个私有属性——这在单元测试里是允许的因为你要验证的就是condition函数本身。这种测试方式把“多智能体协作”拆解成一个个可验证的原子单元正是工程化落地的关键。5. 常见问题与排查技巧实录那些文档里不会写的实战经验5.1 “Agent未被调用”问题速查表90%的情况在这五种里现象可能原因排查步骤解决方案giver调用了但guesser日志完全空白add_edge的to_agent名字拼写错误检查flow.add_edge(giver, guesser)中的guesser是否与WordGuesser.name完全一致大小写敏感修正拼写确保WordGuesser.name guesserguesser日志有但返回消息后流程停止condition函数始终返回False在condition里加print(fDEBUG: {msg})确认传入的msg结构检查msg字段名guesser输出的attempts_leftvsgiver输入的attempts_usedgiver被调用了两次但第二次没收到消息guesser的output_schema缺少必填字段查看guesser.on_message()返回的字典对比GuesserOutput定义补全所有Field(...)字段如guess、is_correct、attempts_left日志显示giver返回了但guesser报SchemaValidationErrorgiver的输出与guesser.input_schema不匹配打印giver返回的result和guesser.input_schema.model_json_schema()用pydantic的parse_obj()手动验证GuesserInput.parse_obj(giver_result)flow.run()后程序立即退出无任何日志giver.on_message()抛出了未捕获异常在giver.on_message()里加try/exceptlogger.exception检查giver内部逻辑如self._target_word是否为None实操心得我遇到过最隐蔽的bug是giver的output_schema里word: str但实际返回了word: None。Flowing的Schema校验在giver返回时就失败了但错误被吞掉只在debug模式下显示。解决方案是在giver.on_message()里加assert self._target_word is not None提前暴露问题。5.2 Flowing的“静默失败”模式如何让它在出错时大声喊出来Flowing默认对很多错误采取“静默忽略”策略比如无效的Agent name、重复注册、Schema校验失败它可能只是跳过不报错。这不利于调试。开启严格模式# 在main.py顶部 import os os.environ[FLOWING_STRICT_MODE] 1 # 启用严格模式 from flowing import Flow # ... rest of code开启后上述所有错误都会变成ValueError或TypeError并附带清晰的错误信息。例如add_edge(giver, nonexistent)会立刻报ValueError: Agent nonexistent not found in registry。这个环境变量是Flowing的隐藏开关官方文档没提但源码里core/registry.py第89行有if os.getenv(FLOWING_STRICT_MODE):判断。5.3 从“猜词”到“真实项目”的三步跃迁我的亲身踩坑路线图这个案例的价值不在于它能玩多久而在于它给你一把解剖刀。我用它过渡到真实项目时走了三步第一步替换LLM保持Schema不变把WordGuesser.on_message()里的确定性算法换成真实的OpenAI调用from openai import AsyncOpenAI client AsyncOpenAI() async def on_message(self, message: Dict[str, Any]) - Dict[str, Any]: prompt f你是一个猜词游戏的玩家。提示{message[hint]}。请猜一个{len(message[word])}字母的单词只返回单词不要解释。 response await client.chat.completions.create( modelgpt-4, messages[{role: user, content: prompt}] ) guess response.choices[0].message.content.strip().lower() # 后续逻辑不变比较、计算attempts_left...关键点return的字典结构{guess: ..., is_correct: ..., attempts_left: ...}完全不变。Flowing不关心你是用算法还是LLM生成的它只认Schema。第二步增加第三个Agent引入“裁判”角色新增RefereeAgent它接收guesser的猜测和giver的答案判断是否正确并返回标准化结果。这时flow.add_edge(guesser, referee)和flow.add_edge(referee, giver)condition函数也升级为检查referee的判决。这一步教会我多智能体不是越多越好而是每个Agent的职责必须单一、契约必须清晰。Referee不参与猜只做判断Giver不参与判断只提供答案。第三步接入真实数据源用Flowing做ETL流水线把“猜词”抽象为“数据转换”giver变成CSVReader读取原始数据guesser变成DataCleaner清洗字段referee变成Validator校验业务规则。消息流变成{raw_data: [...]}→{cleaned_data: [...]}→{valid: True, errors: []}。这时你会发现Flowing的condition边可以完美表达“如果校验失败发给告警Agent如果成功发给入库Agent”。这才是它在生产环境里的真实模样——不是玩具而是可扩展的业务流引擎。最后分享一个小技巧在WordGiver.__init__()里我把self._target_word设为random.choice([apple, banana, cherry])然后在on_message()里加一行logger.info(fNew game: target word is {self._target_word})。这样每次运行都能看到新单词避免因答案固定而误判逻辑。这个小改动让调试效率提升了一倍。
