Flowing框架实战：用猜词游戏快速入门多智能体协作开发

1. 为什么选猜词游戏作为多智能体入门案例

1.1 多智能体框架的学习曲线问题

接触过智能体开发的朋友大概都有这种体会：单智能体的demo跑通很快，一旦要接入第二个、第三个智能体，代码复杂度就指数级上升。消息怎么传、状态怎么同步、谁先说话谁后说话、出错怎么兜底，这些问题在单智能体场景下根本不存在，但一到多智能体就全冒出来了。

市面上主流的智能体框架，像LangChain、CrewAI、AutoGen这些，功能确实强大，但上手门槛也不低。LangChain的Chain和Agent抽象层叠了好几层，CrewAI的角色定义和任务编排概念也不少，新手往往花两三天才搞明白一个最简单的双智能体对话该怎么写。这就带来一个尴尬：想学多智能体，却被框架本身的复杂度劝退。

Flowing框架的设计思路不太一样。它把多智能体的核心抽象压到了极简——智能体就是能接收消息、处理消息、发出消息的单元，智能体之间的连接就是消息通道。没有复杂的角色继承体系，没有层层嵌套的Chain，你定义几个智能体，把它们连起来，就能跑。这种设计哲学特别适合用来做教学案例，因为学习者可以把精力放在理解多智能体协作本身，而不是跟框架的API较劲。

1.2 猜词游戏为什么是理想的教学场景

猜词游戏这个场景选得很妙。它的规则天然就是多智能体的：一个出题者（出词方），一个猜词者（猜词方），出题者知道答案但不能直接说，猜词者不知道答案但可以不断提问或猜测，出题者根据猜测给出反馈（对了、接近了、远了），猜词者根据反馈调整策略。这个循环就是典型的多智能体协作——两个智能体各有各的信息和目标，通过消息交互逐步逼近共同结果。

从技术实现角度看，猜词游戏覆盖了多智能体系统的几个核心要素：信息不对称（出题者知道答案，猜词者不知道）、消息传递（猜测和反馈的往返）、状态维护（已经猜过哪些词、当前猜测范围）、终止条件（猜对或达到最大轮次）。这些要素在更复杂的多智能体应用里都会出现，但猜词游戏把它们压缩到了一个极简的框架里，特别适合作为第一个练手项目。

另外，猜词游戏还有一个隐性优势：它的运行结果非常直观。你不需要看日志分析半天才知道智能体协作得好不好，直接看猜词者几轮猜中、猜词路径是否合理，就能判断整个系统的表现。这种即时反馈对学习阶段的调试和调优非常友好。

1.3 Flowing框架的核心抽象

在动手写代码之前，有必要先把Flowing框架的几个核心概念理清楚。根据我对这类轻量级智能体框架的理解，Flowing的核心抽象大概包括以下几层：

Agent（智能体）：这是最基本的单元。每个Agent有自己的名称、系统提示词（定义它的角色和行为）、以及一个处理消息的方法。当Agent收到消息时，它会调用底层的大语言模型，结合自己的角色设定和对话历史，生成回复。

Flow（流）：Flow定义了智能体之间的连接关系和数据流向。在猜词游戏里，Flow就是出题者和猜词者之间的消息通道——猜词者的猜测流向出题者，出题者的反馈流回猜词者。

Message（消息）：智能体之间传递的内容单元。消息通常包含发送者、接收者、内容、以及可能的元数据（比如轮次编号）。

Runner（运行器）：负责驱动整个Flow运转，控制消息的传递顺序、轮次上限、终止条件等。

这套抽象的好处是概念少、关系清晰。你不需要理解什么“工具调用链”、“记忆后端”、“规划器”这些高级概念，就能先把一个多智能体系统跑起来。等基础跑通了，再逐步往里面加复杂度，学习路径很平滑。

提示：不同版本的Flowing框架在API细节上可能有差异，本文的代码示例基于常见的Flowing设计模式编写，核心逻辑通用，具体方法名请以你使用的版本为准。

2. 环境准备与项目骨架搭建

2.1 依赖安装与版本选择

先把环境搭起来。Flowing框架本身通常是一个Python包，可以通过pip安装。除此之外，你还需要一个大语言模型的接入方式——可以是OpenAI的API，也可以是本地部署的模型（比如通过Ollama或vLLM），或者国内的通义千问、智谱等API。

pip install flowing pip install openai # 如果使用OpenAI兼容的API

版本选择上有个经验：Flowing这类快速迭代的框架，不同版本之间API变动可能比较大。建议先查一下你安装的版本号，然后对照该版本的文档来写代码。我踩过的坑是照着最新文档写代码，但本地装的是旧版本，结果方法名对不上，报了一堆AttributeError。

import flowing print(flowing.__version__)

另外，如果你用的是本地模型，需要确保模型服务已经启动并且可以通过HTTP访问。本地模型的好处是调试方便、不花钱，缺点是推理速度可能慢一些，猜词游戏这种需要多轮交互的场景，如果每轮等十几秒，调试起来会很痛苦。所以初期建议用API，等逻辑跑通了再考虑切本地模型。

2.2 项目目录结构

一个清晰的项目结构能让后续调试省很多事。我习惯这样组织：

guess-word-game/ ├── config.py # 模型配置、API密钥等 ├── agents.py # 智能体定义 ├── flow.py # 流程编排 ├── main.py # 入口 └── utils.py # 辅助函数（日志、计时等）

config.py里放模型相关的配置，比如API地址、密钥、模型名称、温度参数等。把这些集中管理，切换模型时只改一个文件。

agents.py里定义出题者和猜词者两个智能体。每个智能体的核心是系统提示词，这直接决定了它的行为质量。

flow.py里定义两个智能体怎么连接、消息怎么流转、什么时候终止。

main.py是入口，负责初始化、启动流程、打印结果。

utils.py放一些辅助函数，比如格式化输出、记录每轮耗时等。别小看这些辅助代码，调试多智能体系统时，清晰的日志能帮你省下大量时间。

2.3 模型接入配置

模型接入这块，核心是选一个支持多轮对话的API。猜词游戏需要智能体记住之前的猜测和反馈，所以模型必须能处理对话历史。大部分主流API都支持，关键是配置好参数。

# config.py MODEL_CONFIG = { "api_base": "https://api.example.com/v1", # 替换为你的API地址 "api_key": "your-api-key", # 替换为你的密钥 "model_name": "gpt-4o-mini", # 或你使用的模型名称 "temperature": 0.7, # 出题者用低温度，猜词者可以稍高 "max_tokens": 256, }

温度参数这里有个小技巧：出题者的温度可以设低一点（比如0.3），因为它需要稳定地给出反馈，不能太随机；猜词者的温度可以设高一点（比如0.8），鼓励它尝试不同的猜测策略，避免陷入固定模式。这个参数调整对游戏体验影响很大，后面还会细说。

注意：API密钥不要硬编码在代码里提交到版本控制。用环境变量或者单独的.env文件管理，.env记得加到.gitignore里。

3. 智能体角色设计与提示词工程

3.1 出题者智能体的设计要点

出题者智能体的核心职责是：心里有一个目标词，根据猜词者的猜测给出反馈，但不能直接泄露答案。它的系统提示词需要包含几个关键信息：目标词是什么、反馈规则是什么、不能做什么。

# agents.py from flowing import Agent def create_answer_agent(target_word: str): system_prompt = f"""你是一个猜词游戏的出题者。你心里有一个目标词："{target_word}"。 规则： 1. 当猜词者给出一个猜测时，你需要判断它与目标词的关系。 2. 如果猜对了，回复"正确"。 3. 如果猜错了，根据接近程度给出反馈： - "很接近"：猜测词与目标词语义高度相关或属于同一类别 - "有点接近"：猜测词与目标词有一定关联 - "不相关"：猜测词与目标词没有明显关联 4. 绝对不能直接说出目标词，也不能用同义词直接暗示。 5. 每次反馈只给一个等级，不要额外解释。 请严格按照规则回复，不要添加多余内容。""" return Agent( name="出题者", system_prompt=system_prompt, model_config=MODEL_CONFIG, )

这里有几个设计决策值得说明。第一，反馈等级只设了三档（很接近、有点接近、不相关），而不是更细的评分。原因是档位太多，模型判断起来容易不一致，而且猜词者也不好利用太细的反馈。三档足够引导猜词方向，又不会让游戏变得太简单。

第二，明确禁止直接说出目标词或用同义词暗示。这个约束很重要，因为大语言模型有时候会“过于热心”，猜词者猜了几轮没猜中，它可能就直接把答案说出来了。加上这条约束后，模型会收敛很多。

第三，要求“只给一个等级，不要额外解释”。这是为了防止出题者给出过多信息，比如“很接近，它也是一种水果”——这等于把类别都说了，游戏就没意思了。当然，如果你想让游戏简单一点，可以允许出题者给出类别提示，这是难度调节的旋钮。

3.2 猜词者智能体的设计要点

猜词者智能体的职责是：根据出题者的反馈，逐步缩小范围，最终猜中目标词。它的提示词需要引导它采用合理的猜测策略，而不是随机乱猜。

def create_guesser_agent(): system_prompt = """你是一个猜词游戏的猜词者。你需要通过不断猜测，猜出出题者心里的目标词。 策略建议： 1. 第一轮可以猜一个常见词，试探出题者的反馈风格。 2. 根据反馈调整策略： - 如果反馈"很接近"，说明你猜的词和目标词属于同一类别，下一轮猜同类别的其他词。 - 如果反馈"有点接近"，说明方向大致对，但需要调整。 - 如果反馈"不相关"，说明方向错了，换一个完全不同的类别。 3. 每次只猜一个词，不要一次猜多个。 4. 不要重复猜已经猜过的词。 5. 如果连续两轮反馈相同，说明你的调整策略没起作用，需要换一个更大的方向。 请直接输出你猜测的词，不要添加解释。""" return Agent( name="猜词者", system_prompt=system_prompt, model_config={**MODEL_CONFIG, "temperature": 0.8}, )

猜词者的提示词里，策略建议是核心。没有这些建议，模型可能会随机猜词，效率很低。有了策略引导，它会更像一个“会玩”的玩家。特别是“连续两轮反馈相同就换大方向”这条，能有效防止猜词者陷入局部搜索出不来。

另外，要求“直接输出猜测的词，不要添加解释”也很重要。如果猜词者每次回复都带一堆分析，出题者解析起来会麻烦，而且浪费token。保持输出简洁，整个流程会更顺畅。

3.3 提示词中的常见坑与调优经验

提示词写得好不好，直接决定游戏能不能跑起来。我踩过的几个坑：

坑一：出题者反馈太模糊。早期版本我只写了“给出反馈”，没规定反馈格式，结果出题者有时候回复“不太对，再想想”，有时候回复“接近了”，格式完全不统一，猜词者没法稳定解析。后来强制规定三档反馈，问题就解决了。

坑二：猜词者重复猜同一个词。模型有时候会“忘记”自己猜过什么，尤其是对话历史长了之后。解决办法是在提示词里明确要求“不要重复”，同时在代码层面做一个去重检查，如果猜词者输出了重复的词，就自动追加一条系统消息提醒它。

坑三：出题者不小心泄露答案。这个前面提过，加上“绝对不能直接说出目标词”的约束后好很多。但偶尔还是会发生，尤其是猜词者猜了很多轮没猜中时。一个额外的保险是在代码层面检查出题者的回复，如果回复中包含了目标词，就强制替换成“不相关”。

坑四：猜词者第一轮就猜中。如果目标词太常见，猜词者第一轮猜“苹果”，而目标词恰好是“苹果”，游戏一轮就结束了，没什么意思。解决办法是选目标词时避开最高频的词，或者规定第一轮猜测不算数（作为热身）。我一般会选一个中等难度的词，比如“钢琴”、“沙漠”、“咖啡”这种，既不是最热门的，也不是特别冷门的。

实操心得：提示词调优是个迭代过程。建议先把两个智能体的提示词写出来，跑几轮看看效果，然后针对暴露的问题逐步加约束。不要一开始就想写一个完美的提示词，那几乎不可能。

4. 流程编排与消息传递机制

4.1 双智能体消息循环的实现

流程编排的核心是：猜词者发猜测，出题者给反馈，猜词者根据反馈再猜，如此循环，直到猜中或达到轮次上限。用Flowing的Flow来编排，大概是这样：

# flow.py from flowing import Flow, Runner def create_guess_flow(answer_agent, guesser_agent, max_rounds=10): flow = Flow() # 添加两个智能体 flow.add_agent(answer_agent) flow.add_agent(guesser_agent) # 定义消息流向：猜词者 -> 出题者 -> 猜词者 flow.add_edge(guesser_agent, answer_agent) flow.add_edge(answer_agent, guesser_agent) # 设置终止条件 flow.set_max_rounds(max_rounds) flow.set_termination_check(lambda msg: "正确" in msg.content) return flow

这段代码的关键是add_edge定义的消息流向。猜词者到出题者是一条边，出题者到猜词者是一条边，形成一个闭环。Runner会沿着这个闭环驱动消息传递，每轮猜词者发一条消息，出题者回一条消息，算一轮。

set_termination_check定义了终止条件：当出题者的回复中包含“正确”时，流程结束。这个检查函数可以根据需要扩展，比如同时检查轮次是否超限。

4.2 轮次控制与终止条件

轮次控制看起来简单，但有几个细节要注意。第一，最大轮次设多少合适？设太少，猜词者还没找到方向就结束了；设太多，如果猜词者策略不好，会一直绕圈子，浪费token和时间。根据我的经验，10轮是个比较合理的默认值。如果目标词比较偏，可以放宽到15轮；如果目标词很常见，5轮就够了。

第二，终止条件除了“猜中”和“轮次超限”，还应该加一个“异常终止”。比如出题者连续几轮给出无效反馈（格式不对），或者猜词者连续输出重复词，这时候应该主动终止，避免死循环。

def create_termination_check(max_rounds): def check(state): # 猜中 if state.last_message and "正确" in state.last_message.content: return True, "猜中" # 轮次超限 if state.round >= max_rounds: return True, "轮次超限" # 连续无效反馈 if state.consecutive_invalid >= 3: return True, "连续无效反馈" return False, "" return check

这个检查函数返回一个元组，第一个元素表示是否终止，第二个元素是终止原因。把终止原因记录下来，方便后续分析。

4.3 消息历史的维护与截断

多轮对话中，消息历史会越来越长。如果不加处理，到后面几轮，每次调用模型都要传很长的历史，token消耗大，而且模型可能被早期无关信息干扰。

处理策略有两种：全量保留和滑动窗口。全量保留适合轮次少（10轮以内）的场景，简单直接。滑动窗口只保留最近N轮的消息，适合轮次多的场景。

猜词游戏一般10轮左右，全量保留完全没问题。但如果你想把最大轮次设到20轮以上，建议用滑动窗口，保留最近5-6轮就够了。因为猜词者主要需要知道最近的反馈来调整策略，太早的猜测参考价值不大。

def truncate_history(messages, max_turns=6): """保留最近max_turns轮的消息""" if len(messages) <= max_turns * 2: return messages # 保留系统提示 + 最近max_turns轮 system_msgs = [m for m in messages if m.role == "system"] recent_msgs = messages[-(max_turns * 2):] return system_msgs + recent_msgs

注意：截断历史时一定要保留系统提示词，否则智能体会“忘记”自己的角色和规则，行为会变得很奇怪。

5. 完整代码实现与逐段解析

5.1 主流程代码

把前面的模块串起来，主流程大概长这样：

# main.py import time from config import MODEL_CONFIG from agents import create_answer_agent, create_guesser_agent from flow import create_guess_flow from flowing import Runner def main(): target_word = "钢琴" max_rounds = 10 # 创建智能体 answer_agent = create_answer_agent(target_word) guesser_agent = create_guesser_agent() # 创建流程 flow = create_guess_flow(answer_agent, guesser_agent, max_rounds) # 运行 runner = Runner(flow) start_time = time.time() result = runner.run(initial_message="请开始猜词。") elapsed = time.time() - start_time print(f"\n=== 游戏结束 ===") print(f"目标词：{target_word}") print(f"总轮次：{result.round}") print(f"耗时：{elapsed:.1f}秒") print(f"结果：{result.termination_reason}") # 打印每轮记录 for i, turn in enumerate(result.history, 1): print(f"第{i}轮 - 猜词者：{turn.guess} | 出题者：{turn.feedback}") if __name__ == "__main__": main()

这段代码的逻辑很直白：定义目标词，创建两个智能体，编排流程，运行，打印结果。result.history里记录了每一轮的猜测和反馈，方便复盘。

5.2 关键参数的计算与选择

几个关键参数的取值需要说明一下理由：

最大轮次10轮：这个数字是基于猜词游戏的搜索空间估算的。假设目标词是一个中等常见度的词，猜词者第一轮试探，之后每轮缩小范围，理想情况下5-7轮能猜中。留10轮是给猜词者犯错的空间。如果10轮还猜不中，说明策略有问题，继续下去也是浪费。

温度0.3（出题者）和0.8（猜词者）：出题者需要稳定输出三档反馈之一，温度低能保证一致性。猜词者需要探索不同方向，温度高能增加多样性。这个组合是我试了几次后确定的，效果比较平衡。

max_tokens 256：猜词者和出题者的回复都很短（一个词或一个短语），256个token绰绰有余。设太大反而可能让模型“话多”，输出多余内容。

历史保留6轮：10轮的游戏，保留最近6轮，加上系统提示，总token量可控。如果模型上下文窗口小，可以进一步压缩到4轮。

5.3 运行结果示例与分析

跑一次实际的结果大概是这样：

第1轮 - 猜词者：水果 | 出题者：不相关 第2轮 - 猜词者：汽车 | 出题者：不相关 第3轮 - 猜词者：音乐 | 出题者：很接近 第4轮 - 猜词者：吉他 | 出题者：有点接近 第5轮 - 猜词者：钢琴 | 出题者：正确

这个结果很典型。猜词者前两轮在试探大类别，第三轮猜“音乐”得到“很接近”，说明方向对了。第四轮猜“吉他”得到“有点接近”，说明是乐器但不对。第五轮猜“钢琴”，命中。整个过程5轮，符合预期。

如果猜词者表现不好，比如：

第1轮 - 猜词者：苹果 | 出题者：不相关 第2轮 - 猜词者：香蕉 | 出题者：不相关 第3轮 - 猜词者：橘子 | 出题者：不相关 ...

这说明猜词者陷入了“水果”这个类别出不来。可能的原因是提示词里“换一个完全不同的类别”这条约束不够强，或者模型本身倾向于在相近词之间打转。解决办法是加强提示词，明确要求“如果连续两轮不相关，必须换到与之前完全无关的类别”。

6. 常见问题排查与调优技巧

6.1 智能体“不听话”的排查思路

多智能体系统最常见的问题就是智能体不按预期行为。排查思路可以按这个顺序来：

第一步，检查提示词。大部分“不听话”都是提示词没写清楚。比如出题者直接说出了答案，检查提示词里有没有“绝对不能直接说出目标词”这条约束。如果没有，加上；如果有但还发生，加强语气，比如改成“违反此规则将导致游戏失败”。

第二步，检查消息格式。如果猜词者输出了一长段话而不是一个词，检查提示词里有没有“直接输出猜测的词，不要添加解释”。同时可以在代码层面做后处理，提取第一个词作为猜测。

第三步，检查历史消息。如果智能体“忘记”了之前的轮次，检查历史消息是否正确传递。有时候框架的默认行为是不传历史，需要手动配置。

第四步，检查模型能力。如果换了小模型（比如7B参数的本地模型），它可能确实理解不了复杂的提示词。这时候要么换大模型，要么简化提示词和游戏规则。

6.2 猜词效率低的优化方法

猜词者效率低，表现为轮次用完了还没猜中，或者猜词路径很乱。优化方法有几个：

优化提示词策略。在猜词者的提示词里加入更具体的策略，比如“第一轮猜一个上位词（如‘动物’、‘食物’），根据反馈再猜下位词”。上位词策略能快速定位大类别，比随机猜效率高很多。

加入猜测历史摘要。如果历史消息太长被截断了，猜词者可能忘记猜过什么。可以在每轮开始时，把“已猜过的词”和“对应的反馈”整理成一个简短摘要，作为系统消息注入。这样即使历史被截断，猜词者也能看到关键信息。

调整温度。如果猜词者总是猜同一类词，把温度调高一点，增加随机性。如果猜词者太随机，总是跳来跳去，把温度调低一点。

换一个更强的模型。如果以上方法都试了效果还不好，可能是模型能力不够。换一个推理能力更强的模型，猜词策略会明显改善。

6.3 多智能体协作的稳定性保障

多智能体系统比单智能体更容易出问题，因为多了一个交互维度。保障稳定性的几个经验：

加超时机制。每次调用模型设置超时时间（比如30秒），超时就重试或跳过。避免某个智能体卡住导致整个流程挂起。

加异常捕获。模型调用可能失败（网络问题、API限流等），用try-except包起来，失败时给一个默认回复，让流程继续跑。

加日志记录。每轮的消息、耗时、token消耗都记录下来。出问题时，日志是排查的第一手资料。

加结果校验。出题者的反馈必须是三档之一，如果不是，代码层面强制修正。猜词者的输出必须是一个词，如果是一段话，提取第一个词。

def validate_feedback(feedback: str) -> str: """校验并修正出题者的反馈""" valid = ["正确", "很接近", "有点接近", "不相关"] for v in valid: if v in feedback: return v return "不相关" # 默认值 def extract_guess(guess: str) -> str: """从猜词者的输出中提取猜测词""" # 取第一行，去掉标点 first_line = guess.strip().split("\n")[0] return first_line.strip("。！？.,!?")

这两个校验函数看起来简单，但能挡掉大部分格式问题，让流程更稳定。

6.4 常见问题速查表

这张表是我在实际调试中总结的，大部分问题都能在里面找到对应方案。遇到新问题时，先对照这张表排查，能省不少时间。

7. 从猜词游戏扩展到更复杂的多智能体场景

猜词游戏跑通之后，你会发现Flowing框架的核心模式——定义智能体、连接消息通道、设置终止条件——可以套用到很多场景。比如把出题者换成“面试官智能体”，猜词者换成“求职者智能体”，就变成了一个模拟面试系统。或者把两个智能体换成“买家”和“卖家”，就变成了一个议价模拟。

扩展的关键在于：角色定义要清晰，消息协议要明确，终止条件要可判定。猜词游戏里，角色是出题者和猜词者，消息协议是“猜测-反馈”，终止条件是“猜中或轮次超限”。换成其他场景，只要把这三样重新定义，流程编排的骨架基本不用大改。

另一个扩展方向是增加智能体数量。比如在猜词游戏里加一个“裁判智能体”，负责监督出题者是否泄露答案、猜词者是否重复猜测。裁判智能体不直接参与猜词，但它的存在能提升整个系统的可靠性。这种“监督者”模式在更复杂的多智能体系统里很常见，比如自动化测试里的“验证智能体”、内容审核里的“审核智能体”。

Flowing框架的轻量特性在这里体现出优势：加一个智能体就是多定义一个Agent、多加几条边，不需要重构整个系统。这种可扩展性让猜词游戏不仅是一个教学案例，也是一个可以持续演进的实验平台。

我在实际使用中发现，多智能体系统的调试成本主要花在“理解智能体为什么这么行为”上。猜词游戏的好处是行为直观——猜词路径一目了然，出题反馈也简单。从这个简单场景入手，把多智能体的消息流转、状态维护、异常处理这些基本功练扎实，再去碰更复杂的场景，会从容很多。