企业官网建设流程全解析

1. 为什么我要用猜词游戏来入门多智能体

多智能体系统这个词听起来很唬人，但真正动手写起来，最难的往往不是算法本身，而是怎么让多个智能体之间的消息流转起来、怎么定义它们各自的职责边界、怎么在本地快速验证逻辑而不被复杂的部署流程劝退。我见过太多人一上来就去啃那些动辄几千行的多智能体框架源码，结果连一个能跑起来的最小闭环都没搭出来，热情直接消耗殆尽。

Flowing 这个框架吸引我的地方就在于它的定位很克制——它不试图解决所有问题，而是把"智能体之间的消息传递"这件事做扎实。你不需要先理解什么复杂的调度算法，也不需要配置一堆外部服务，只要会写 Python 函数，就能把多个智能体串起来跑。而猜词游戏恰好是一个天然适合多智能体协作的场景：出题者知道答案但不能直接说，猜词者不知道答案但要根据反馈逐步逼近，裁判负责判断对错并给出提示。这三个角色之间的信息不对称和交互逻辑，几乎就是多智能体系统最核心的缩影。

这篇文章面向的是有 Python 基础、想快速上手多智能体开发但不想被复杂框架劝退的开发者。我会从零开始，把整个猜词游戏的实现过程拆开揉碎，包括为什么这样设计角色、消息怎么在智能体之间流转、Flowing 的核心 API 怎么用、以及我在实际调试中踩过的那些坑。你跟着走一遍，不仅能得到一个能跑的多智能体案例，更重要的是能理解多智能体协作的底层逻辑，之后换成其他场景也能举一反三。

提示：本文假设你已经装好了 Python 3.8 或更高版本，并且能在终端里正常执行 python 命令。如果你还没配好环境，建议先花十分钟把 Python 装好，后面的内容会顺畅很多。

2. Flowing 框架的核心机制与猜词游戏的适配逻辑

2.1 Flowing 到底解决了什么问题

在聊具体实现之前，得先说清楚 Flowing 的定位。市面上很多多智能体框架走的是"大而全"的路线，内置了记忆管理、工具调用、规划器、执行器一大堆模块，学起来成本很高。Flowing 走的是另一条路：它只关注智能体之间的消息传递和状态流转，把其他事情都交给你自己决定。

这个设计选择带来的直接好处是，你不需要理解框架的"世界观"就能开始写代码。在 Flowing 里，一个智能体本质上就是一个能接收消息、处理消息、发出消息的对象。你可以把它想象成一个邮箱：别人往你的邮箱里投信，你读完信之后决定要不要回信、回给谁、回什么内容。多个邮箱之间通过地址互相投递，就形成了消息网络。

Flowing 的核心抽象大概有这么几个：

• Agent（智能体）：消息的处理单元，每个智能体有自己的名字和消息处理逻辑。
• Message（消息）：智能体之间传递的数据包，包含发送者、接收者、内容等字段。
• Flow（流转）：定义消息如何在智能体之间传递的规则，决定了谁在什么条件下把消息发给谁。
• Runtime（运行时）：负责驱动整个流程运转，管理消息队列和智能体生命周期。

这套抽象的好处是足够简单，你不需要关心底层是怎么调度的，只需要定义好每个智能体收到消息后做什么、然后把消息发给谁就行。

2.2 猜词游戏为什么适合做多智能体入门案例

猜词游戏的规则大家都熟：一个人心里想一个词，其他人来猜，每次猜完告诉对方"对了"或者"不对"，有时候还会给点提示比如"接近了"或者"差远了"。这个看似简单的游戏，拆解成多智能体系统之后，恰好覆盖了多智能体协作的几个核心问题。

第一个是信息不对称。出题者知道答案，猜词者不知道，裁判知道答案但只负责判断。这种信息差是多智能体系统里最常见的场景——每个智能体掌握的信息不同，需要通过消息传递来逐步缩小差距。

第二个是角色分工。出题者负责生成答案和给出反馈，猜词者负责根据反馈调整策略，裁判负责仲裁。三个角色各司其职，不能互相替代。这对应到实际系统里就是不同智能体承担不同职责，通过消息协作完成整体目标。

第三个是闭环反馈。猜词者每猜一次，裁判给出判断，出题者根据判断决定是否继续，猜词者再根据反馈调整下一次猜测。这个循环一直持续到猜对或者达到最大次数。多智能体系统里大量的协作模式本质上都是这种"行动-反馈-调整"的闭环。

第四个是终止条件。游戏不能无限进行下去，要么猜对了结束，要么猜太多次结束。多智能体系统同样需要明确的终止条件，否则消息会在智能体之间无限流转，把资源耗尽。

把这四点想清楚之后，你会发现猜词游戏虽然简单，但麻雀虽小五脏俱全。用它来练手，比一上来就搞什么复杂的工作流编排要实在得多。

2.3 三个智能体的职责边界怎么划

在动手写代码之前，我习惯先把每个智能体的职责边界画清楚。这一步看起来多余，但实际上能省掉后面大量的调试时间。职责边界模糊是多智能体系统出问题的最常见原因——两个智能体都觉得自己该处理某条消息，或者都觉得自己不该处理，结果要么重复处理要么消息丢失。

猜词游戏里三个智能体的职责我是这样划分的：

这个表格看起来简单，但它明确了每个智能体"什么时候被触发、收到什么、发出什么、什么时候停"。实际写代码的时候，只要严格按照这个边界来实现，基本不会出现消息乱飞的情况。

注意：职责边界一旦确定，就不要在实现过程中随意修改。我见过有人写着写着觉得"这个逻辑放在裁判里也行、放在出题者里也行"，结果两边都写了一遍，消息被处理了两次，排查了半天才发现是重复处理。

3. 从零搭建猜词游戏：环境、代码与运行

3.1 环境准备与 Flowing 安装

环境这块没什么好说的，Python 3.8 以上就行。我实测下来 3.8 到 3.12 都能跑，再高的版本理论上也没问题，但没必要为了跑一个入门案例去折腾最新版。安装 Flowing 直接用 pip 就行：

pip install flowing

如果你用的是虚拟环境（强烈建议），先激活环境再装：

python -m venv venv source venv/bin/activate # Linux/macOS # 或者 Windows 下用 venv\Scripts\activate pip install flowing

装完之后可以验证一下：

import flowing print(flowing.__version__)

能打印出版本号就说明装好了。如果报 ModuleNotFoundError，大概率是 pip 装到了别的 Python 环境里，检查一下 which python 和 which pip 是不是指向同一个环境。

3.2 定义消息结构：智能体之间到底传什么

消息结构的设计是多智能体系统的地基。传什么字段、字段什么类型、哪些字段是必须的，这些想不清楚后面会一直改。猜词游戏里我定义的消息结构大概是这样的：

from dataclasses import dataclass, field from typing import Optional, List @dataclass class GuessMessage: sender: str receiver: str content: str round_num: int = 0 history: List[str] = field(default_factory=list) is_correct: Optional[bool] = None is_finished: bool = False

这里有几个设计决策值得说一下。sender 和 receiver 是必须的，不然消息不知道从哪来到哪去。content 放具体的猜测词或者反馈内容。round_num 记录当前是第几轮，用来判断是否达到最大轮次。history 保存历史猜测记录，猜词者需要根据历史来调整策略。is_correct 是裁判给出的判断结果，is_finished 是全局的终止信号。

为什么用 dataclass 而不是普通 dict？因为 dataclass 有类型提示，写代码的时候 IDE 能给你补全，运行时也能少犯低级错误。而且字段有默认值的话，构造消息的时候不用每个字段都填，代码更干净。

提示：消息结构一旦确定，尽量不要再改。如果确实需要加字段，用带默认值的可选字段，不要动已有字段的类型和含义，否则所有智能体都得跟着改。

3.3 出题者智能体的实现细节

出题者的逻辑其实很简单：一开始生成一个目标词，然后等裁判的反馈。如果裁判说猜对了，就发结束信号；如果裁判说猜错了，就判断轮次是否用完，用完了也发结束信号，没用完就继续等。

import random class QuestionerAgent: def __init__(self, name: str, word_pool: list, max_rounds: int = 10): self.name = name self.word_pool = word_pool self.max_rounds = max_rounds self.target_word = None def start(self) -> GuessMessage: self.target_word = random.choice(self.word_pool) return GuessMessage( sender=self.name, receiver="judge", content=self.target_word, round_num=0 ) def handle_feedback(self, msg: GuessMessage) -> GuessMessage: if msg.is_correct: return GuessMessage( sender=self.name, receiver="guesser", content="游戏结束，猜对了", round_num=msg.round_num, is_finished=True ) if msg.round_num >= self.max_rounds: return GuessMessage( sender=self.name, receiver="guesser", content=f"游戏结束，答案是：{self.target_word}", round_num=msg.round_num, is_finished=True ) return GuessMessage( sender=self.name, receiver="guesser", content="继续猜", round_num=msg.round_num, is_finished=False )

这里有个细节：出题者把目标词发给裁判，而不是直接发给猜词者。为什么？因为猜词者不应该知道答案，它只能通过裁判的反馈来推断。如果出题者直接把答案发给猜词者，那游戏就没意义了。这个信息隔离是多智能体系统里很重要的一点——每个智能体只能看到自己该看到的信息。

3.4 猜词者智能体的策略设计

猜词者的逻辑稍微复杂一点，因为它需要根据历史反馈来调整策略。最简单的策略是随机猜，但那样体验太差。我实现了一个基于排除法的策略：维护一个候选词池，每次猜完之后根据反馈排除掉不可能的选项。

class GuesserAgent: def __init__(self, name: str, word_pool: list): self.name = name self.candidates = list(word_pool) self.guessed = [] def make_guess(self, msg: GuessMessage) -> GuessMessage: if msg.is_finished: return None if not self.candidates: self.candidates = list(self.guessed) guess = random.choice(self.candidates) self.candidates.remove(guess) self.guessed.append(guess) return GuessMessage( sender=self.name, receiver="judge", content=guess, round_num=msg.round_num + 1, history=list(self.guessed) )

这个策略的核心是"猜过的不再猜"。虽然简单，但比纯随机要靠谱得多。如果你想让猜词者更聪明，可以加入语义相似度判断——比如根据裁判的"接近了"反馈来优先猜语义相近的词。不过那需要额外的词向量模型，对于入门案例来说有点重了。

注意：猜词者的候选池不能为空。如果候选池被猜空了还没猜对，要么重置候选池，要么直接结束。我一开始忘了处理这个边界情况，结果程序在候选池为空的时候抛了异常，排查了半天。

3.5 裁判智能体的判断逻辑

裁判的职责最清晰：收到猜测，跟目标词比对，返回结果。但这里有个容易忽略的点——裁判需要维护轮次计数，因为轮次信息是判断是否继续的依据。

class JudgeAgent: def __init__(self, name: str, target_word: str): self.name = name self.target_word = target_word self.current_round = 0 def judge(self, msg: GuessMessage) -> GuessMessage: self.current_round = msg.round_num is_correct = (msg.content == self.target_word) return GuessMessage( sender=self.name, receiver="questioner", content=msg.content, round_num=self.current_round, is_correct=is_correct, history=msg.history )

裁判不生成新信息，只做判断和转发。这种"薄裁判"的设计在多智能体系统里很常见——裁判只负责仲裁，不参与策略决策，这样职责边界最清晰。

3.6 用 Flowing 把三个智能体串起来

前面三个智能体都是独立的类，现在需要用 Flowing 把它们串成一个流程。Flowing 的 Flow 抽象就是干这个的：

from flowing import Flow, Runtime def build_game_flow(word_pool, max_rounds=10): questioner = QuestionerAgent("questioner", word_pool, max_rounds) guesser = GuesserAgent("guesser", word_pool) init_msg = questioner.start() judge = JudgeAgent("judge", init_msg.content) flow = Flow() flow.add_agent(questioner) flow.add_agent(guesser) flow.add_agent(judge) current_msg = init_msg while not current_msg.is_finished: if current_msg.receiver == "judge": current_msg = judge.judge(current_msg) elif current_msg.receiver == "questioner": current_msg = questioner.handle_feedback(current_msg) elif current_msg.receiver == "guesser": next_msg = guesser.make_guess(current_msg) if next_msg is None: break current_msg = next_msg if current_msg.round_num > max_rounds: break return current_msg if __name__ == "__main__": words = ["苹果", "香蕉", "橘子", "葡萄", "西瓜", "草莓", "菠萝", "芒果"] result = build_game_flow(words, max_rounds=8) print(f"游戏结束：{result.content}")

这段代码的核心是一个 while 循环，不断根据当前消息的 receiver 字段决定谁来处理。这就是最简单的消息流转逻辑——谁收到消息谁处理，处理完把新消息放回循环。

实际用 Flowing 的 Runtime 会更优雅一些，它帮你管理消息队列和智能体调度，你只需要注册智能体和消息处理函数就行。但对于入门来说，先用显式的 while 循环把逻辑跑通，理解清楚消息是怎么流的，再去用框架的高级抽象，会顺畅很多。

4. 调试过程中暴露的三个典型问题

4.1 消息死循环：为什么加了终止条件还是停不下来

我第一次跑的时候遇到了一个很典型的问题：程序跑了几轮之后不结束了，一直在出题者和猜词者之间来回传消息。排查了半天发现是终止条件的判断位置不对。

当时的逻辑是这样的：出题者收到裁判的"猜错了"反馈后，判断轮次是否用完，没用完就发"继续猜"给猜词者。猜词者收到"继续猜"后生成新的猜测发给裁判。裁判判断后发给出题者。看起来没问题对吧？

问题出在轮次计数上。裁判的 current_round 是从消息里读的，但猜词者生成新猜测的时候，round_num 是 msg.round_num + 1。而出题者收到裁判反馈时，msg.round_num 是裁判传过来的值。如果某一轮里消息在几个智能体之间转了好几圈，round_num 就会被重复加，导致轮次计数比实际快，或者反过来比实际慢。

修复方法很简单：轮次计数只在一个地方维护。我把它统一放在裁判里，其他智能体只读不写。裁判每次判断的时候把 current_round 加一，然后把这个值传给下游。这样轮次计数就是单调递增的，不会乱。

提示：多智能体系统里，共享状态的维护一定要指定唯一的"权威源"。多个智能体同时修改同一个状态，几乎必然出问题。

4.2 猜词者候选池耗尽的边界情况

第二个坑是猜词者的候选池被猜空。我一开始的候选池就是词库本身，如果词库只有 8 个词，最多猜 8 次就空了。如果 8 次都没猜对（虽然概率很低），第 9 次调用 random.choice 就会抛 IndexError。

修复方案有两个：一是候选池空了就重置为完整词库，允许重复猜；二是候选池空了直接结束游戏。我选了第一个，因为重置之后猜词者至少还能继续猜，虽然可能重复，但不会崩。

if not self.candidates: self.candidates = list(self.word_pool) # 可选：排除已经猜过的词 self.candidates = [w for w in self.candidates if w not in self.guessed]

这个边界情况提醒我，多智能体系统里每个智能体的内部状态都可能出现"耗尽"的情况，写代码的时候要把这些边界都想清楚。

4.3 消息字段缺失导致的静默失败

第三个坑最隐蔽。有一次我改消息结构的时候，给 GuessMessage 加了一个新字段，但忘了给某个智能体的消息构造处加上这个字段。结果程序不报错，但那个字段是 None，下游智能体读到 None 之后逻辑走偏了，表现是"游戏莫名其妙就结束了"。

这种静默失败在多智能体系统里特别难排查，因为消息在多个智能体之间流转，你很难一眼看出是哪一步出的问题。我的解决办法是在每个智能体处理消息的入口加断言：

def handle_feedback(self, msg: GuessMessage) -> GuessMessage: assert msg.round_num is not None, "round_num 不能为空" assert msg.is_correct is not None, "is_correct 不能为空" # ... 后续逻辑

断言虽然简单，但能在问题发生的第一时间定位到源头，比事后翻日志高效得多。

5. 从猜词游戏延伸出去的多智能体设计经验

5.1 角色划分的粒度怎么把握

猜词游戏里我用了三个智能体，但实际项目里角色数量可能更多也可能更少。角色划分的粒度怎么把握？我的经验是：一个智能体只做一件事，但这件事要足够完整。

什么叫"足够完整"？就是把这个智能体的职责单独拿出来，它能形成一个闭环——有明确的输入、明确的处理逻辑、明确的输出。如果某个智能体的逻辑里出现了"如果 A 情况就这样，如果 B 情况就那样"，而且这两种情况的处理逻辑完全不相关，那大概率应该拆成两个智能体。

反过来，如果两个智能体之间的消息传递非常频繁，而且每次传递的内容都很简单（比如只是转发一下），那可能应该合并成一个。消息传递本身是有开销的，过度拆分会让系统变得碎片化。

猜词游戏里出题者、猜词者、裁判三个角色的划分刚好合适：每个角色都有独立的决策逻辑，角色之间的消息传递也不频繁（每轮就几次），拆成三个不多不少。

5.2 消息协议设计的三条原则

消息协议是多智能体系统的"接口"，设计得好不好直接决定了系统的可维护性。我总结了三条原则：

第一条：消息要自包含。每条消息应该包含处理它所需的全部信息，不要依赖智能体的内部状态。比如裁判判断猜测是否正确，需要知道目标词，那目标词就应该在消息里，而不是让裁判去问出题者。自包含的消息让智能体之间解耦，任何一个智能体都可以独立测试。

第二条：字段要有明确语义。不要用 data、info、payload 这种含糊的字段名，要用 target_word、guess_content、is_correct 这种一看就知道是什么的字段名。多智能体系统里消息字段的可读性比省那几个字符重要得多。

第三条：预留扩展字段。消息结构一旦定下来，后面加字段的成本很高。所以一开始就留一两个通用的扩展字段，比如 metadata 或者 extra，类型用 dict，需要的时候往里塞东西，不用改消息结构。

5.3 终止条件的三种常见模式

多智能体系统必须有明确的终止条件，否则消息会无限流转。猜词游戏里用的是"目标达成或轮次耗尽"这种模式，实际项目里常见的终止条件大概有三种：

目标达成型：某个智能体完成了目标，发一个终止信号，所有智能体收到后停止。猜词游戏里猜对了就是这种。

资源耗尽型：轮次用完、时间到了、token 用完了，强制终止。猜词游戏里的最大轮次就是这种。

外部干预型：人工介入或者外部系统发信号终止。这种在自动化系统里用得少，但在需要人工审核的场景里很常见。

实际系统里往往是多种模式组合使用，比如"目标达成或轮次耗尽，以先到者为准"。猜词游戏用的就是这种组合。

5.4 本地调试多智能体的实用技巧

多智能体系统调试起来比单智能体麻烦，因为消息在多个智能体之间流转，出问题的时候很难定位是哪一步。我常用的几个技巧：

第一个是加日志。每个智能体处理消息的入口和出口都打日志，记录消息的关键字段。跑一遍之后看日志，消息的流转路径一目了然。

import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(message)s') def handle_feedback(self, msg): logging.info(f"{self.name} 收到消息: round={msg.round_num}, correct={msg.is_correct}") # ... 处理逻辑 logging.info(f"{self.name} 发出消息: receiver={result.receiver}") return result

第二个是单步执行。把 while 循环改成手动一步步执行，每执行一步就打印当前状态，确认没问题再执行下一步。这样能精确定位到是哪一步出的问题。

第三个是单元测试。每个智能体单独写测试，构造各种输入消息，验证输出是否符合预期。智能体级别的测试比系统级别的测试好写也好维护。

第四个是可视化消息流。把消息的流转路径画出来（用文本也行），看看有没有意外的跳转或者循环。猜词游戏的消息流应该是"出题者→裁判→出题者→猜词者→裁判→..."这种交替模式，如果出现了"猜词者→猜词者"这种自环，那肯定有问题。

6. 这套最简案例还能怎么继续折腾

猜词游戏跑通之后，如果你想继续深入多智能体，有几个方向可以扩展。

第一个方向是给猜词者加策略。现在的猜词者是随机猜，你可以让它根据裁判的反馈来调整策略。比如裁判说"接近了"，那下次就猜语义相近的词。这需要引入词向量或者简单的相似度计算，能让猜词者的表现明显提升。

第二个方向是加一个提示者智能体。现在的反馈只有"对"和"错"，你可以加一个提示者，根据猜测和目标词的差距给出更细的反馈，比如"很接近""有点远""方向错了"。这样猜词者能更快收敛，也更能体现多智能体协作的价值。

第三个方向是把游戏改成对抗模式。现在是出题者和猜词者协作，你可以改成两个猜词者竞争，谁先猜对谁赢。这就引入了竞争关系，多智能体系统里的博弈逻辑就出来了。

第四个方向是接入大模型。现在的智能体都是规则驱动的，你可以把猜词者的决策逻辑换成调用大模型，让大模型根据历史反馈来生成猜测。这样能处理更复杂的猜词场景，比如猜一个概念而不是一个具体的词。

不管往哪个方向扩展，核心思路都是一样的：明确角色职责、定义消息协议、设计终止条件、本地调试验证。把这四步做好，多智能体系统就不会太复杂。

我在实际项目里用 Flowing 搭过比猜词游戏复杂得多的流程，但底层逻辑和这个案例是一模一样的。区别只在于智能体更多、消息协议更复杂、终止条件更多样。所以别小看这个最简案例，把它吃透了，后面遇到再复杂的场景，你也能拆解成类似的模式来处理。

最后分享一个我踩过的坑：不要一开始就追求完美设计。我刚开始做多智能体的时候，总想把消息协议设计得面面俱到，结果设计了两天还没开始写代码。后来想通了，先用最简的消息结构把流程跑通，跑通之后再根据实际需要加字段、改逻辑。迭代比一次性设计靠谱得多，尤其是在你还不熟悉框架的时候。