第六章框架开发实践

一句话总结

第四章你徒手实现了智能体范式,这一章看看"专业轮子厂"是怎么做的:AutoGen 让多个智能体开会,AgentScope 用消息驱动大型多智能体应用,CAMEL 让两个 AI 角色扮演互相激发,LangGraph 把智能体流程画成严格的状态图。学框架的正确姿势不是背 API,而是看懂每个框架"替你管了什么"。

本章你能学到什么

为什么徒手写完范式后还要学框架(以及框架到底帮你干了什么脏活)
四大框架的设计哲学、核心抽象、代表案例、适用场景,以及原书自己承认的代价
三个完全离线、不装包、不用密钥的手写实验:状态图引擎 / 消息中心 / 角色扮演引擎——把 LangGraph、AgentScope、CAMEL 的内核各自用几十行还原一遍
八条踩坑清单(终止词、model_info、归约器、容错、可观测……)与面试高频的框架对比题

一、从手动实现到框架:框架替你干了哪些脏活

回想第四章,你徒手处理了这些事:拼提示词、解析输出、管理历史、循环控制、异常兜底。一个玩具项目还好,真实项目里还有:多智能体的消息路由、并发执行、断点续跑、流式输出、可观测……框架的价值就是把这些"每个项目都要重写一遍的脏活"标准化。

但框架不是白拿的,代价是:学习成本(每个框架一套概念体系)、黑盒感(出问题不知道卡在哪层)、绑定风险(深度耦合后难迁移)。所以本书的路线才设计成:先徒手(第四章)→ 再看框架(本章)→ 最后自己造框架(第七章)——徒手过的人看框架源码如同看老朋友,没徒手过的人只能当"API 调用员"。

二、框架一:AutoGen —— 让智能体像团队一样开会

微软出品。核心抽象是对话(Conversation):一切协作都建模成多个智能体在群聊里轮流发言。

原书案例:软件开发团队。四个角色各配一段职责提示词,拉进一个"圆桌群聊":

ProductManager(需求分析)→ Engineer(写代码)→ CodeReviewer(审查)→ UserProxy(代表人类拍板);
用 RoundRobinGroupChat 让他们按顺序轮流发言,用 TextMentionTermination 约定"谁的发言里出现 TERMINATE 这个词,就散会";
任务:开发一个比特币价格显示应用。

四个角色的差别只在系统消息(system_message)这一段提示词里,原书给每个角色都写了"职责清单 + 工作步骤 + 交接口令":PM 结尾说"请工程师开始实现",工程师结尾说"请代码审查员检查",审查员结尾说"代码审查完成,请用户代理测试"——用一句口令把接力棒递给下一个人,这是多智能体提示词工程最实用的一招。

第四个角色 UserProxy 是特殊的:它用 UserProxyAgent 创建,不接 LLM,发言时会停下来等真人在终端里敲字(源码日志里那行 Enter your response: 就是它)。它的职责说明里写着"完成测试后请回复 TERMINATE"——终止权握在人手里。

核心代码形状(体会 API 风格即可,跑通需 pip install "autogen-agentchat" "autogen-ext[openai]"):

from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(       # 任何兼容 OpenAI 协议的服务都能接
    model=os.getenv("LLM_MODEL_ID", "gpt-4o"),
    api_key=os.getenv("LLM_API_KEY"),
    base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"),
)

engineer = AssistantAgent(name="Engineer", model_client=model_client,
                          system_message="你是一位资深软件工程师……完成后说'请代码审查员检查'。")
user_proxy = UserProxyAgent(name="UserProxy",
                            description="代表用户验证功能。完成测试后请回复 TERMINATE。")

team_chat = RoundRobinGroupChat(
    participants=[product_manager, engineer, code_reviewer, user_proxy],
    termination_condition=TextMentionTermination("TERMINATE"),
)
result = await team_chat.run(task="开发一个比特币价格显示应用……")

两个新手必踩的坑,先记下来: 1. 终止词写错=永远停不下来。 上面的 TextMentionTermination("TERMINATE") 必须和角色提示词里让它们说的那个词一模一样(大小写也算)。写成 APPROVE 而提示词里说的是 TERMINATE,群聊就会一直转到你手动 Ctrl+C。稳妥做法是再加一道保险丝:max_turns=N 或 MaxMessageTermination(N),和第四章的 max_steps 是同一个思想。 2. 接 AssistantAgent 用国产模型时可能报错,提示模型不支持 function calling / vision。原因是 AutoGen 需要知道模型的能力清单,而它只内置了 OpenAI 系列的。解法:给客户端补一个 model_info={"vision": False, "function_calling": True, "json_output": True, "family": "unknown"}。 3. 想看逐字输出,别用 run()(它憋到最后一次性返回),换 run_stream() 配 Console(...),发言过程就会实时滚动。

它替你管了:发言顺序、消息广播、终止判定、人类介入的插口。适合:角色分工明确的协作任务(软件团队、辩论、评审流水线)。这和 docs/2 第 12 篇手敲的多智能体旅行规划是同一思想,那边是你自己写的调度循环,这边框架代劳。

别对多智能体产生幻想(重要):原书这个案例的真实运行日志里,PM、工程师、审查员各发言一轮就把活干完了,后面几轮基本是互相客套("感谢您的反馈""期待下次合作"),最后靠真人敲下 TERMINATE 才结束。它不会自动进入"审查员挑毛病→工程师返工→再审"的理想循环——想要返工,你得自己用条件路由(LangGraph)或更复杂的终止条件去逼它。这是本章最值钱的一条经验:轮询群聊只保证"每个人都说了话",不保证"活越干越好"。

三、框架二:AgentScope —— 消息驱动的多智能体操作系统

阿里巴巴达摩院出品。如果 AutoGen 像个灵活的"对话工作室",AgentScope 就是一整套"智能体操作系统"——它把开发、调试、部署全生命周期都管了。

3.1 核心抽象:一切都是消息(Msg)

别的框架里,智能体 A 找 B 干活是函数调用(A 直接调 B 的方法,得等 B 返回);AgentScope 里是发消息(A 把信投进邮筒就走,B 自己取)。就这一个改动带来四个好处:

好处	大白话
异步解耦	发信人不用站在原地等回信,天然支持高并发
位置透明	你不用关心收信人在本机还是在另一台服务器,邮局(MsgHub)负责送到
可观测	每封信都留了底,复盘和调试时能一封封翻
可靠	信可以存盘、可以重发,系统崩了也能从上次的信继续

消息的标准长相(四个字段:谁说的 / 说了什么 / 什么身份 / 附带信息):

from agentscope.message import Msg

message = Msg(
    name="Alice",              # 发送者
    content="Hello, Bob!",     # 内容
    role="user",               # 身份:user / assistant / system
    metadata={"timestamp": "2024-01-15T10:30:00Z", "priority": "normal"},
)

写自己的智能体只要继承 AgentBase、实现 reply(怎么回话)和可选的 observe(听到别人说话时怎么记):

class CustomAgent(AgentBase):
    def reply(self, x: Msg) -> Msg:                     # 核心:收一封信,回一封信
        return Msg(name=self.name, content=self.model(x.content), role="assistant")

    def observe(self, x: Msg) -> None:                  # 可选:只听不回,记进记忆
        self.memory.add(x)

3.2 四层架构:每层管什么

原书给了一张架构图,记住这个"从下往上"的顺序就够了:

基础组件层:Message(统一消息格式)、Memory(短期+长期记忆)、Model API(屏蔽不同大模型的差异)、Tool(工具);
智能体基础设施层:内置 ReAct 范式、并行工具调用、状态管理、钩子(hook),原生异步;
多智能体协作层:MsgHub(消息中心,负责路由)+ Pipeline(编排:顺序执行、并发执行);
开发部署层:AgentScope Runtime(生产运行时)+ AgentScope Studio(可视化调试台)。

名词补课——MsgHub(消息中心):就是邮局。它支持三种投递方式:点对点(只给一个人)、广播(给所有人)、组播(只给某个小圈子——狼人杀里"狼人频道"就靠这个)。它还能把消息自动存进 SQLite/MongoDB(持久化,断了能续),以及通过 RPC 把消息送到另一台服务器上的智能体(分布式,对你的代码完全透明)。

3.3 原书案例:三国狼人杀

6 名玩家(孙权、周瑜、曹操、张飞、司马懿、赵云),每人被塞进双重身份:游戏角色(狼人/预言家/女巫/村民)+ 三国人格。提示词模板长这样:

你是{character},在这场三国狼人杀游戏中扮演{role}。
重要规则:
1. 你只能通过对话和推理参与游戏
2. 不要尝试调用任何外部工具或函数
3. 严格按照要求的JSON格式回复

有意思的观察:同一个游戏角色,不同人格演出来完全不同——演狼人的"曹操"狡猾善伪装,演狼人的"张飞"直率冲动。

三个技术要点:

夜晚狼人阶段 = 临时组播频道。用 async with MsgHub(self.werewolves, enable_auto_broadcast=True, announcement=...) 开一个只有狼人收得到的频道,讨论完再 set_auto_broadcast(False) 关掉广播、进入投票;
投票阶段 = 并行扇出。fanout_pipeline(self.alive_players, msg=..., structured_model=...) 把同一个问题同时发给所有人、异步收齐——这才真的模拟了"同时投票",而不是排队一个个问;
规则用结构化输出硬约束。用 Pydantic 定义每种行动的格式,比如女巫的 WitchActionModelCN 有 use_antidote: bool / use_poison: bool / target_name,讨论用的 DiscussionModelCN 有 reach_agreement: bool / confidence_level: int = Field(ge=1, le=10) / key_evidence。游戏规则不再靠提示词"求"模型遵守,而是靠数据模型校验强制执行——这就是 docs/2 第 12 篇讲的结构化输出,框架内置了。

还有一个细节值得抄走:每个智能体调用都包 try/except,出错就塞一个默认响应让游戏继续,而不是整局崩掉。

try:
    response = await wolf("请分析当前局势", structured_model=DiscussionModelCN)
except Exception as e:
    print(f"⚠️ {wolf.name} 讨论时出错: {e}")
    response = DiscussionModelCN(reach_agreement=False, confidence_level=5,
                                 key_evidence="暂时无法分析")   # 兜底,不中断

它替你管了:消息路由、并行扇出、结构化输出校验、持久化、分布式。适合:游戏、模拟社会、大量智能体并发的场景。玩过 docs/2 第 14 篇"记忆小镇"的话,你徒手写的 NPC 对话循环,AgentScope 给了工业级版本。

代价(原书自己承认的):要懂异步编程和分布式概念,门槛偏高;简单的双人对话用它是过度工程化;框架相对新,生态和社区资源还在完善。

四、框架三:CAMEL —— 两个 AI 互相"套话"出成果

CAMEL 最初的目标就一句话:在最少人类干预下,让两个智能体通过角色扮演自主协作解决复杂任务。它靠两根支柱做到这件事。

4.1 支柱一:角色扮演(Role-Playing)

任务交给两个能力互补的角色:

AI 用户(AI User):提需求、下指令、规划步骤。原书例子里是"资深股票交易员"——懂市场但不会编程;
AI 助手(AI Assistant):执行、给方案。是"优秀的 Python 程序员"——会编程但不懂股票。

于是任务就变成一场"跨领域专家对话":交易员提专业需求,程序员翻译成代码。双方都无法独立完成的事,对话完成了。

注意这个反直觉的映射:CAMEL 里 user 是推动者/需求方,assistant 是执行者/方案方。所以"写电子书"案例里,负责规划结构的作家被分配给 user_role_name,提供专业内容的心理学家被分配给 assistant_role_name。搞反了对话就会互相推让。

4.2 支柱二:引导性提示(Inception Prompting)

光设角色不够——凭什么两个 AI 不跑题、不客套、不无限循环?靠开场前分别注入给它们的一段结构化系统提示词,四个部分:

明确自身角色:"你是一位资深的股票交易员……";
告知协作者角色:"你正在与一位优秀的 Python 程序员合作……";
定义共同目标:"你们的共同目标是开发一个股票交易策略分析工具";
设定行为约束和沟通协议(最关键):要求 AI 用户"一次只提出一个清晰、具体的步骤",要求 AI 助手"在完成上一步之前不要追问更多细节",并规定用特定标志(如 <SOLUTION>)标识完成。

"沟通协议"这一条是 CAMEL 的精髓,也是你能立刻抄进自己项目的技巧:两个 LLM 对话最容易死在"互相请对方先说"和"客套到超时",一条"一次只提一步 / 结尾必须写 Next request"就能治好。

4.3 原书案例:合写电子书

"AI 作家"(user)× "AI 心理学家"(assistant),协作创作《拖延症心理学》科普书,要求 8000–10000 字、实证为本、通俗易懂。

from camel.societies import RolePlaying
from camel.models import ModelFactory
from camel.types import ModelPlatformType

model = ModelFactory.create(                 # 任何平台都能接,这里以通义千问为例
    model_platform=ModelPlatformType.QWEN,
    model_type=os.getenv("LLM_MODEL"),
    url=os.getenv("LLM_BASE_URL"),
    api_key=os.getenv("LLM_API_KEY"),
)

role_play_session = RolePlaying(
    assistant_role_name="心理学家",          # 执行者:给专业内容
    user_role_name="作家",                   # 推动者:提结构和要求
    task_prompt=task_prompt,
    model=model,
    with_task_specify=False,                 # True 会先让 LLM 把任务描述细化一遍
)

chat_turn_limit, n = 30, 0
input_msg = role_play_session.init_chat()    # 开场白由 AI 自己生成,不用人写
while n < chat_turn_limit:
    n += 1
    assistant_response, user_response = role_play_session.step(input_msg)   # 一轮=各说一次
    if assistant_response.msg is None or user_response.msg is None:
        break                                # 防止提前终止时崩掉
    if "<CAMEL_TASK_DONE>" in user_response.msg.content:
        print("✅ 电子书创作完成!")
        break
    input_msg = assistant_response.msg       # 助手的输出 = 下一轮的输入

三个必须记住的点:

init_chat():开场白不用人类写,框架根据角色和任务自动生成;
step():驱动一轮完整交互(作家提需求 + 心理学家给内容),并把助手输出接成下一轮输入,形成环环相扣的创作链;
<CAMEL_TASK_DONE>:CAMEL 的约定终止标志。外层同时必须有 chat_turn_limit 兜底——和 AutoGen 的终止词、第四章的 max_steps 是同一个保险丝。

原书观察到的协作分了四个自然阶段(1–5 轮搭框架对齐目标 → 6–20 轮"心理学家给硬核知识、作家做通俗翻译" → 21–25 轮互换角色做打磨与事实核查 → 收尾总结)。注意那个"翻译官"动作:把"大脑中的现在偏见"译成"一个只顾眼前糖果、不顾长远健康的任性孩子"——这本教程里所有比喻,用的就是同一手法。

它替你管了:双角色的引导性提示工程、轮次推进、任务终止。适合:内容创作、头脑风暴、用"AI 教 AI"生成训练数据(docs/2 第 15 篇"蒸馏造数据"的一种玩法)。

代价(原书自己列的):极度依赖提示词质量——效果不好时你很难定位是角色定义、任务描述还是交互规则的问题;双人以上就吃力(缺少 AutoGen 那样的对话路由、AgentScope 那样的状态同步、以及意见分歧时的仲裁机制)。

五、框架四:LangGraph —— 把智能体画成状态图

LangChain 家族出品。它把智能体流程建模成状态机(State Machine),并画成有向图。核心抽象三件套:

State(状态):一个全局共享的数据结构(通常是 TypedDict),记录任务进行到哪了。所有节点都能读它、改它——它就是流程的"任务档案袋";
Node(节点):一个个普通 Python 函数,收状态、干活、返回"要更新哪些字段";
Edge(边):节点间的连线。条件边(Conditional Edges)通过一个判断函数看当前状态,动态决定下一步跳哪——分支和循环全靠它。

第一章说过 Workflow 和 Agent 的区别——LangGraph 的定位恰好是两者的桥梁:用 Workflow 的确定性骨架,装 Agent 的智能节点。出错可回溯、每步可检查,这是它在生产环境吃香的根本原因。

名词补课两个: - TypedDict:一个"规定了有哪些字段、每个字段是什么类型"的字典。写 class State(TypedDict): number: int 之后,编辑器就知道 state["number"] 是整数,拼错字段名会立刻标红。它只在写代码时帮你检查,运行时就是个普通字典。 - Annotated[list, add_messages] 归约器(reducer):LangGraph 默认的更新规则是"覆盖"——节点返回 {"messages": [新消息]} 会把整个列表换掉。给字段套上 Annotated[list, add_messages] 就改成"追加",新消息接在后面而不是顶掉旧的。对话历史字段必须这么写,否则每个节点跑完历史就只剩一条,这是新手最常踩的坑之一。

原书案例:智能搜索助手。三个节点串成一条链:

understand_query:LLM 干两件事——总结用户想知道什么,以及生成一个更适合搜索引擎的关键词。状态里因此有 user_query(用户原话的理解)和 search_query(优化后的搜索词)两个字段,这个拆分显著提升搜索质量;
tavily_search:调搜索 API,整个包在 try/except 里,失败就把 step 标记成 "search_failed";
generate_answer:检查 state["step"]——搜索成功就基于搜索结果作答,搜索失败就走降级分支:用模型自身知识回答并告知用户。

"降级兜底"做成一个显式分支,这是生产级思维,docs/2 第 16 篇坑 12 讲的就是这个。

组图时还有一个生产必备件:

from langgraph.checkpoint.memory import InMemorySaver

memory = InMemorySaver()
app = workflow.compile(checkpointer=memory)      # ← 关键

名词补课——checkpointer(检查点存储器):它在每个节点执行完之后自动把状态存一份。带来三个能力:①多轮对话——同一个 thread_id 再次调用时自动接上上次的状态,不用你手动拼历史;②断点续跑——程序崩了可以从最后一个检查点接着走,不用从头重来;③人工介入(human-in-the-loop)——可以在某个节点前暂停、等人批准再继续。InMemorySaver 存内存里(重启就没了,适合开发);生产要换成 SQLite/Postgres 版本。"能断点续跑"是很多公司选 LangGraph 的头号理由,面试时说得出这个词就赢了一半。

代价(原书自己列的):前期样板代码多——定义状态、节点、边一通操作,简单任务反而繁琐;你得多想"怎么控制流程",而不只是"做什么"。而且它不会涌现——流程是你预先定死的,别指望它像对话式框架那样冒出意外的协作花样。调试也有挑战:问题可能在节点内部逻辑、在节点间传递的状态、或在条件边的判断上,得对整张图有全局理解。

动手实战:三个离线实验,把三种"协作抽象"都手敲一遍

四个框架的抽象,三个能用几十行纯 Python 手写出来。 全部不联网、不需要密钥、不装任何包——目的不是替代框架,而是让你之后读框架源码时"看到的都是老朋友"。

framework_lab/
├── mini_graph.py       ① 手写状态图引擎    → 对应 LangGraph
├── msghub_demo.py      ② 手写消息中心      → 对应 AgentScope
└── roleplay_demo.py    ③ 手写角色扮演引擎  → 对应 CAMEL

实验一:手写一个状态图引擎(约 40 行)→ 看透 LangGraph

新建 mini_graph.py,逐段敲入:

"""实验一:用 40 行手写一个"状态图引擎",不装任何包。
理解 State / Node / Edge / 条件边,然后再看 LangGraph 就没有黑盒。
"""

END = "__END__"


class MiniGraph:
    def __init__(self):
        self.nodes = {}        # 名字 -> 处理函数
        self.edges = {}        # 名字 -> 固定下一站
        self.branches = {}     # 名字 -> (判断函数, {返回值: 下一站})
        self.entry = None

    def add_node(self, name, fn):
        self.nodes[name] = fn

    def add_edge(self, src, dst):
        self.edges[src] = dst

    def add_conditional_edges(self, src, router, mapping):
        self.branches[src] = (router, mapping)

    def set_entry_point(self, name):
        self.entry = name

    def invoke(self, state, max_steps=50):
        cur = self.entry
        for _ in range(max_steps):
            if cur == END:
                return state
            update = self.nodes[cur](state)       # 节点只返回"要改哪些字段"
            state = {**state, **(update or {})}   # 合并进全局状态
            if cur in self.branches:
                router, mapping = self.branches[cur]
                cur = mapping[router(state)]      # 条件边:看状态挑下一站
            else:
                cur = self.edges.get(cur, END)
        raise RuntimeError("超过最大步数,图里大概有死循环")


# ===== 用它跑一个"翻倍到过 100 就停"的流程 =====
def double_node(state):
    n = state["number"] * 2
    return {"number": n, "steps": state["steps"] + [f"翻倍 -> {n}"]}


def check_node(state):
    return {"done": state["number"] >= 100}


def route(state):
    return "finish" if state["done"] else "again"


if __name__ == "__main__":
    g = MiniGraph()
    g.add_node("double", double_node)
    g.add_node("check", check_node)
    g.set_entry_point("double")
    g.add_edge("double", "check")
    g.add_conditional_edges("check", route, {"again": "double", "finish": END})

    result = g.invoke({"number": 3, "steps": [], "done": False})
    print("处理轨迹:")
    for s in result["steps"]:
        print(" ", s)
    print("最终结果:", result["number"])
    print("走了", len(result["steps"]), "轮循环")

✅ Checkpoint 1:python mini_graph.py,真实输出:

处理轨迹:
  翻倍 -> 6
  翻倍 -> 12
  翻倍 -> 24
  翻倍 -> 48
  翻倍 -> 96
  翻倍 -> 192
最终结果: 192
走了 6 轮循环

盯着这 40 行看三件事:

invoke 里那个 for 循环就是所有图编排框架的全部秘密。 LangGraph、Dify 的工作流、n8n……内核都是"当前节点 → 执行 → 看状态决定下一站",区别只在周边设施(持久化、并发、可视化)。
state = {**state, **update} 这一行就是"归约器"的默认行为——覆盖。 想让某个字段变成"追加"(比如对话历史),就得给它单独写规则,这正是 LangGraph add_messages 干的事。把 double_node 的 state["steps"] + [...] 改成直接 [...],轨迹就只剩最后一条——亲手试一次,你就永远不会踩那个坑。
max_steps=50 是保险丝。 把 check_node 的 >= 100 改成 >= 10**9,你会看到它抛 超过最大步数,而不是把电脑跑死。任何带循环的智能体框架都必须有这道闸(第四章的 max_steps、AutoGen 的 max_turns、CAMEL 的 chat_turn_limit,同一件事)。

现在回头看:把 double_node 换成"调 LLM 生成答案"、check_node 换成"调 LLM 评审答案",这个循环就是第四章 Reflection 范式的图版实现! 框架没有魔法,只是把你徒手写的 while 循环变成了显式的、可视化的图。

实验二:手写一个消息中心(约 50 行)→ 看透 AgentScope

新建 msghub_demo.py。这个实验要让你亲眼看到"组播频道"和"并行扇出"到底是什么:

"""实验二:用 50 行手写一个"消息中心(MsgHub)",不装任何包。
理解 AgentScope 的三招:Msg 统一格式、私密频道(组播)、并行扇出(fanout)。
"""
from dataclasses import dataclass, field


@dataclass
class Msg:
    """统一消息格式:谁说的、说了什么、什么身份、附带元数据"""
    name: str
    content: str
    role: str = "assistant"
    metadata: dict = field(default_factory=dict)

    def __str__(self):
        return f"{self.name}: {self.content}"


class Agent:
    """最小智能体:有名字、有性格、有收件箱。reply 里本该调 LLM,这里用规则冒充"""

    def __init__(self, name, persona):
        self.name = name
        self.persona = persona
        self.inbox = []

    def observe(self, msg: Msg):
        self.inbox.append(msg)

    def reply(self, prompt: str) -> Msg:
        heard = len(self.inbox)
        return Msg(self.name, f"({self.persona})我听到了{heard}条消息,对'{prompt}'的回应是:同意")


class MsgHub:
    """消息中心:只把消息发给"在这个频道里"的人"""

    def __init__(self, participants, announcement=None):
        self.participants = participants
        self.announcement = announcement
        self.log = []

    def __enter__(self):
        if self.announcement:
            self.broadcast(Msg("主持人", self.announcement, role="system"))
        return self

    def __exit__(self, *exc):
        return False

    def broadcast(self, msg: Msg):
        self.log.append(msg)
        for a in self.participants:
            if a.name != msg.name:          # 自己说的话不用再发给自己
                a.observe(msg)

    def speak(self, agent, prompt):
        msg = agent.reply(prompt)
        self.broadcast(msg)
        return msg


def fanout(agents, prompt):
    """并行扇出:同一个问题同时发给所有人,收齐所有回答。
    真框架用 async 并发,这里用列表推导演示"同时问、一起收"的语义。"""
    return [a.reply(prompt) for a in agents]


if __name__ == "__main__":
    孙权 = Agent("孙权", "沉稳")
    周瑜 = Agent("周瑜", "多谋")
    曹操 = Agent("曹操", "多疑")
    张飞 = Agent("张飞", "直率")
    all_players = [孙权, 周瑜, 曹操, 张飞]
    werewolves = [孙权, 周瑜]

    print("===== 夜晚:狼人专属频道(只有狼人收到) =====")
    with MsgHub(werewolves, announcement="狼人们,请讨论今晚的击杀目标") as hub:
        hub.speak(孙权, "杀谁")
        hub.speak(周瑜, "杀谁")

    for a in all_players:
        print(f"  {a.name} 的收件箱里有 {len(a.inbox)} 条消息")

    print("\n===== 白天:全员广播 =====")
    with MsgHub(all_players, announcement="天亮了,请自由讨论") as day:
        day.speak(曹操, "谁是狼")

    for a in all_players:
        print(f"  {a.name} 的收件箱里有 {len(a.inbox)} 条消息")

    print("\n===== 投票:并行扇出 =====")
    for v in fanout(all_players, "请投票"):
        print(" ", v)

✅ Checkpoint 2:python msghub_demo.py,真实输出:

===== 夜晚:狼人专属频道(只有狼人收到) =====
  孙权 的收件箱里有 2 条消息
  周瑜 的收件箱里有 2 条消息
  曹操 的收件箱里有 0 条消息
  张飞 的收件箱里有 0 条消息

===== 白天:全员广播 =====
  孙权 的收件箱里有 4 条消息
  周瑜 的收件箱里有 4 条消息
  曹操 的收件箱里有 1 条消息
  张飞 的收件箱里有 2 条消息

===== 投票:并行扇出 =====
  孙权: (沉稳)我听到了4条消息,对'请投票'的回应是:同意
  周瑜: (多谋)我听到了4条消息,对'请投票'的回应是:同意
  曹操: (多疑)我听到了1条消息,对'请投票'的回应是:同意
  张飞: (直率)我听到了2条消息,对'请投票'的回应是:同意

这几行数字就是狼人杀能玩起来的全部原理:

夜晚过后,曹操和张飞的收件箱是 0 条。 信息隔离不是靠提示词"请不要偷看",而是物理上没发给他们——组播频道决定了谁能知道什么。智能体的"信息权限"要用消息路由实现,不能指望模型自觉,这是多智能体设计的第一原则。
白天广播后,曹操只有 1 条、张飞有 2 条(公告+曹操的发言)、狼人有 4 条。 每个人的上下文都不一样——所以他们的推理结论也就自然不一样,"演技"是信息差自然产生的,不是编出来的。
fanout 里每个人报出的"听到 N 条"各不相同,证明它们各自带着自己的上下文独立回答。 真框架用 async 让这些调用同时发出(所以叫并行);这里用列表推导只是把"同时问、一起收"的语义演示出来。

改一改:把 broadcast 里的 if a.name != msg.name 删掉,再跑一次——每个人的计数都会变多,因为智能体开始"听到自己说的话"。这在真实项目里会导致模型把自己的话当成别人的观点然后自我强化,是很隐蔽的一类 bug。

实验三:手写角色扮演引擎(约 50 行)→ 看透 CAMEL

新建 roleplay_demo.py。这个实验的重点不是代码,而是那两段系统提示词——CAMEL 的全部魔法都在里面:

"""实验三:用 50 行手写 CAMEL 的"角色扮演"引擎,不装任何包、不用密钥。
理解三件事:引导性提示(Inception Prompting)、AI用户/AI助手轮流推进、终止标志。
"""

TASK = "创作一本关于'拖延症心理学'的短篇科普电子书"

# ① 引导性提示:开场前分别注入给两个 AI 的"行动纲领"
USER_SYS = f"""你是【作家】。你不懂心理学,但你懂怎么把话说得好懂。
你的搭档是【心理学家】。你们的共同目标:{TASK}。
沟通协议:
- 一次只提一个具体步骤,格式必须是 "Instruction: ..." 开头
- 不要自己动手写内容,只负责提要求
- 你认为整本书已经完成时,只回复 <CAMEL_TASK_DONE>"""

ASSISTANT_SYS = f"""你是【心理学家】。你懂实证研究,但不擅长通俗表达。
你的搭档是【作家】。你们的共同目标:{TASK}。
沟通协议:
- 只回答对方这一步的要求,格式必须是 "Solution: ..." 开头
- 不要反问、不要抢着规划下一步
- 每次回答结尾写一行 "Next request." 把球踢回去"""

# ② 假 LLM:用剧本冒充模型。换成真 LLM 只需替换这两个列表
WRITER_SCRIPT = [
    "Instruction: 请为'引言'章节写 400 字草稿,说明什么是拖延。\nInput: None",
    "Instruction: 请解释'时间折扣理论',并给一个生活比喻。\nInput: None",
    "Instruction: 请给出 3 条可立即执行的改善建议。\nInput: None",
]
PSYCHOLOGIST_SCRIPT = [
    "Solution: 拖延是明知该做却反复推迟的行为倾向,本质是情绪调节问题而非时间管理问题。\nNext request.",
    "Solution: 时间折扣指越远的奖励在大脑里越不值钱。好比一个只顾眼前糖果的孩子。\nNext request.",
    "Solution: 1)把任务切到 2 分钟能开始 2)先做最难的一步 3)记录启动时刻而非完成时刻。\nNext request.",
]


class RolePlaying:
    def __init__(self, user_sys, assistant_sys, task_prompt):
        self.user_sys, self.assistant_sys = user_sys, assistant_sys
        self.task_prompt = task_prompt
        self.turn = 0

    def init_chat(self):
        """开场白由 AI 自己生成,不用人类写"""
        return "Instruction: 请先确认你理解本书的目标读者和篇幅要求。\nInput: None"

    def step(self, input_msg):
        """驱动一轮:AI助手先答,AI用户再提下一步"""
        i = self.turn
        assistant = (PSYCHOLOGIST_SCRIPT[i] if i < len(PSYCHOLOGIST_SCRIPT)
                     else "Solution: 全书内容已齐备。\nNext request.")
        user = (WRITER_SCRIPT[i] if i < len(WRITER_SCRIPT)
                else "<CAMEL_TASK_DONE>")
        self.turn += 1
        return assistant, user


if __name__ == "__main__":
    session = RolePlaying(USER_SYS, ASSISTANT_SYS, TASK)
    print(f"协作任务: {session.task_prompt}\n")

    input_msg, n, limit = session.init_chat(), 0, 30
    while n < limit:
        n += 1
        assistant_msg, user_msg = session.step(input_msg)
        print(f"--- 第 {n} 轮 ---")
        print(f"心理学家 (AI Assistant): {assistant_msg}")
        print(f"作家 (AI User): {user_msg}\n")
        if "<CAMEL_TASK_DONE>" in user_msg or "<CAMEL_TASK_DONE>" in assistant_msg:
            print("✅ 电子书创作完成!")
            break
        input_msg = assistant_msg          # 助手的输出 = 下一轮的输入

    print(f"总共进行了 {n} 轮协作对话")

✅ Checkpoint 3:python roleplay_demo.py,真实输出的头尾:

协作任务: 创作一本关于'拖延症心理学'的短篇科普电子书

--- 第 1 轮 ---
心理学家 (AI Assistant): Solution: 拖延是明知该做却反复推迟的行为倾向,本质是情绪调节问题而非时间管理问题。
Next request.
作家 (AI User): Instruction: 请为'引言'章节写 400 字草稿,说明什么是拖延。
Input: None
...
--- 第 4 轮 ---
心理学家 (AI Assistant): Solution: 全书内容已齐备。
Next request.
作家 (AI User): <CAMEL_TASK_DONE>

✅ 电子书创作完成!
总共进行了 4 轮协作对话

三个观察:

step() 一次推进"一轮完整交互",而不是"一句话"。 这是 CAMEL API 最容易看错的地方:它返回的是一对回复(助手的 + 用户的)。
终止靠标志 + 轮次上限双保险。 把 limit 改成 2,你会看到它在没说完的情况下停在 总共进行了 2 轮——没有上限的双 AI 对话可以一直互相客套到你的余额见底,这是真花钱的坑。
换成真 LLM 只需替换那两个剧本列表(改成两次 API 调用,把 USER_SYS / ASSISTANT_SYS 当 system 提示词传进去)。请注意引导性提示原封不动就能用——因为它本来就是纯提示词工程,和框架无关。这就是本实验最大的收获:CAMEL 的价值 90% 在那两段提示词里,10% 在代码里。

选做升级(需装包 / 需密钥)

跑真 LangGraph:pip install langgraph,把实验一的 MiniGraph 换成 StateGraph——你会发现 API 几乎一一对应(add_node / add_edge / add_conditional_edges / set_entry_point / compile / invoke),只多一个 compile()。这是验证"你真的看懂了"的最快方式。
跑原书的三节点搜索助手:understand → search → answer,搜索节点用 ddgs(免密钥,写法抄 docs/2 第 13 篇)替代书中的 Tavily,失败分支照抄书中的降级思路,再加上 InMemorySaver 体验多轮对话。

六、四框架对比与选型心法

	AutoGen	AgentScope	CAMEL	LangGraph
核心抽象	对话/群聊	消息	角色扮演	状态图
擅长	团队协作式任务	大规模并发多智能体	双角色内容共创	可控的复杂流程
可控性	中(发言顺序可控,内容不可控)	中	低(全靠对话自驱)	高(每步显式定义)
上手难度	低	中	低	中
原书自陈的代价	对话不可预测、Token 消耗高、调试困难	过度工程化、异步+分布式门槛、生态较新	强依赖提示词、调试不透明、>2 个角色弱、无仲裁	样板代码多、不会涌现、需全局理解整张图

表格里刻意没有"生产采用率"这一行。网上流传的框架排行榜大多来自 GitHub Star 数或某篇博客的印象,原书没有给出任何采用率数据,我也不编。选型该看你的任务形状,不看谁星星多。

选型心法三句话:

流程说得清楚 → LangGraph(可控、可回溯、能断点续跑);
角色说得清楚、流程说不清楚 → AutoGen/CAMEL(让对话自己涌现出流程);
智能体数量多、要并发要工程化 → AgentScope。

以及最重要的一句:别做框架收藏家。四个都见过之后,深入用好一个,其他的保持"看得懂文档"即可。框架年年换,第四章那套"提示词→解析→执行→回喂"的内功不会变。

踩坑与专家提示

坑 1:终止词写错一个字母,群聊永远不结束(真金白银的坑)。 TextMentionTermination("TERMINATE") 匹配的字符串必须和角色提示词里让它们说的词完全一致。写成 APPROVE 而提示词里说 TERMINATE,群聊就会一直转,每一轮都在烧 Token。解法:永远同时挂第二道闸——RoundRobinGroupChat(..., max_turns=12) 或 MaxMessageTermination(20)。这和第四章的 max_steps、CAMEL 的 chat_turn_limit、实验一里的 max_steps=50 是同一件事:任何带循环的智能体都必须有保险丝,没有例外。

坑 2:接非 OpenAI 模型时 AutoGen 报 "model not found"。 OpenAIChatCompletionClient 会去查一张内置的模型能力表,你的自定义模型名不在表里就直接抛错。补上 model_info 手动声明能力即可:

model_client = OpenAIChatCompletionClient(
    model="qwen-plus", api_key=..., base_url=...,
    model_info={"vision": False, "function_calling": True,
                "json_output": True, "family": "unknown"},
)

坑 3:run() 跑完才出结果,你会以为程序卡死了。 多智能体一轮几十秒很正常。开发期一律用 run_stream() 配 Console(...),边跑边打印:await Console(team_chat.run_stream(task=...))。

坑 4:LangGraph 的对话历史每步都被清空。 默认归约器是"覆盖"。存历史的字段必须写成 messages: Annotated[list, add_messages],否则节点返回 {"messages": [新消息]} 会把之前所有历史顶掉。实验一里"把 state["steps"] + [...] 改成 [...]"就是这个坑的最小复现。

坑 5:多智能体框架里,一个智能体抛异常就全局崩盘。 狼人杀这类长流程里,任何一次 LLM 调用都可能超时或返回不合格式的内容。每个智能体的每次调用都要独立 try/except 并给一个默认值,让流程带伤前进而不是整局重开(§3.3 的兜底代码)。

坑 6:别对"多智能体"产生幻想。 轮询群聊只保证"每个人都说了话",不保证"活越干越好"。原书日志里 PM、工程师、审查员各说一次话就开始互相客套,最后要人类手打 TERMINATE 才停。想要真的返工迭代,得自己把评审结论做成显式的条件边(实验一那个 route 函数),靠对话自觉是靠不住的。

坑 7:装了框架反而更难 debug。 框架把提示词拼装藏进了内部,出问题时你看不到实际发出去的内容。解法:先徒手实现一遍(第四章),再用框架——这也是本章三个离线实验存在的意义。上生产前一定接一个可观测工具(LangSmith / 自建日志),把每一次实际请求和响应落盘。

坑 8:结构化输出别靠"请以 JSON 格式回复"。 用框架提供的 Pydantic 约束(AgentScope 的 structured_model=)让它在协议层保证字段和取值范围(confidence_level: int = Field(ge=1, le=10)),比在提示词里请求靠谱一个数量级。参见第五章/docs/2 第 16 篇关于 JSON 解析失败的坑。

安全提示:密钥永远只放 .env,.env 必须写进 .gitignore。 本章四个框架的示例都从 os.getenv(...) 读密钥,不要图省事写进代码——一旦提交到公开仓库,几分钟内就会被扫走。真泄露了,第一件事是去平台吊销并重发,改代码是第二件事。

面试视角

"LangGraph 和 AutoGen 的本质区别?"——控制反转的方向不同:LangGraph 是你定义流程,LLM 填空(图是骨架);AutoGen 是LLM 对话驱动流程(对话是骨架)。前者可控性强适合生产,后者灵活适合探索。
"为什么很多公司用 LangGraph 而不是裸写?"——答框架管的脏活:状态持久化(断点续跑)、条件路由、并行分支、人工介入节点(human-in-the-loop)、与 LangSmith 打通的可观测。再补一句"简单场景裸写更好,避免过度工程",体现判断力。
"多智能体一定比单智能体好吗?"——不。多智能体引入通信成本、错误传播、调试地狱(docs/2 第 16 篇坑 10)。原则:单智能体+好工具能解决的,绝不上多智能体。
"LangGraph 的 checkpointer 解决什么问题?"——三件事:多轮对话(同一 thread_id 自动接上上次状态)、断点续跑(崩了从最后一个检查点继续)、人工介入(节点前暂停等审批)。能说出这三条,面试官会认为你真的上过生产。
"什么是归约器(reducer)?为什么对话历史字段要特殊处理?"——LangGraph 默认的状态更新是覆盖;对话历史需要追加,所以要写成 Annotated[list, add_messages]。不写就会每步只剩一条消息——这是最高频的实战 bug。
"怎么防止多智能体对话停不下来?"——双保险:语义终止条件(TextMentionTermination,且终止词必须和角色提示词里说的完全一致)+ 硬性轮次上限(max_turns / MaxMessageTermination)。只有前者就是在赌模型听话。
"CAMEL 里 AI User 和 AI Assistant 谁是需求方?"——AI User 是推动者/需求方,AI Assistant 是执行者。这个映射反直觉,写代码时角色填反是最常见错误:提结构和要求的那个填 user_role_name。

划重点

框架的本质是把每个项目都要重写的脏活标准化:消息路由、循环控制、结构化输出、状态管理。代价是学习成本、黑盒感、绑定风险。
四框架四个核心抽象:AutoGen=群聊,AgentScope=消息,CAMEL=角色扮演,LangGraph=状态图。抓住抽象就抓住了一切 API。
三个离线实验各自证明一件事:invoke 里那个 for 循环就是所有图编排框架的内核;信息隔离靠消息路由而不是靠提示词请求(曹操夜里收到 0 条);CAMEL 的价值 90% 在那两段引导性提示里。
任何带循环的智能体都必须有保险丝——max_steps / max_turns / chat_turn_limit,叫法不同,是同一道闸。终止词写错会烧到你余额见底。
别对多智能体产生幻想:轮询群聊只保证每个人都说了话,不保证活越干越好。要真的迭代返工,得自己写显式的条件边。
选型心法:流程清楚用图,角色清楚用对话,规模大用消息;深入一个,了解其余。不要相信没有出处的"采用率排行榜"。
先徒手、再框架、后造轮子——下一章(第七章)就轮到你自己造框架了。

下一章预告:第七章是全书的"内功大考"——从0到1构建你自己的智能体框架。完整的 15 文件手敲教程在 docs/2 第 18 篇,第七章文档会带你先看懂架构设计,再动手。

第六章 框架开发实践