上次折腾完电商询价智能体,我觉得LangGraph这玩意儿挺有意思。最近经常刷到辩论赛的视频,我就寻思:能不能用LangGraph搭一个辩论系统,让正方、反方、裁判三个AI自己吵起来?
说干就干。于是就有了这个——智能体的辩论赛。
一、先定个小目标
需求其实不复杂:
正方Agent:立场“科技发展利大于弊”,能记住双方论点,每轮递进,不重复,会反驳 反方Agent:立场“科技发展弊大于利”,同样有记忆、有规划 裁判Agent:把控流程,记录论点,最终判定胜负
辩论流程:正方先发言 → 裁判小结 → 反方发言 → 裁判小结 → 正方发言……循环几轮后,裁判给出最终判决。
难点在于:
每个Agent都要有记忆,不能车轱辘话来回说 每个Agent要有规划,不能东一榔头西一棒子 流程控制要清晰,裁判得知道什么时候该结束
我寻思,这不就是LangGraph的典型应用场景吗?有状态、多步骤、条件分支,完美匹配。
二、技术选型:为什么又是LangGraph?
LangGraph的优势在这种多角色、多轮对话的场景里特别明显:
状态管理:用一个全局State对象,所有节点共享,谁都能读写 条件边:根据裁判的判决结果决定是继续辩论还是结束 可观测性:每个节点的输入输出都能打印出来,调试方便
我设计的状态图长这样:
正方发言 → 裁判小结 → 反方发言 → 裁判小结 → 正方发言 → ... ↓ 轮次够了? ↙ ↘ 结束 继续裁判节点是核心,它既负责记录每轮小结,也负责判断是否达到最大轮次,最后给出胜负判定。
三、状态设计:一个字典管所有
先定义辩论过程中需要记住的所有信息:
from typing import Annotated, List, TypedDictfrom langgraph.graph.message import add_messagesfrom langchain_core.messages import HumanMessage, AIMessage, SystemMessageclassDebateState(TypedDict): messages: Annotated[List, add_messages] # 完整辩论记录 pro_arguments: List[str] # 正方论点列表 con_arguments: List[str] # 反方论点列表 round: int # 当前轮次 max_rounds: int # 最大轮次 verdict: str # 裁判最终判定 ("pro"/"con")messages用了LangGraph自带的add_messages归约器,会自动追加新消息,不用手动拼接pro_arguments和con_arguments分别记录双方已提出的论点,用于避免重复round和max_rounds控制流程verdict在裁判最终判定后设置,用来触发图结束
四、逐个节点拆解
节点1:正方Agent
核心任务:生成新论点,同时反驳反方。提示词里明确要求“避免重复”、“针对性反驳”。
defpro_node(state: DebateState): prompt = f"""你是一个正方辩手,立场是“科技发展利大于弊”。当前辩论历史:{format_history(state.get('messages', []))}你已经提出的论点:{state.get('pro_arguments', [])}反方已经提出的论点:{state.get('con_arguments', [])}现在轮到你发言。请:1. 避免重复已提过的论点;2. 如果有反方论点,针对性地反驳;3. 提出新的论据支持己方立场;4. 发言要简洁有力,控制在100字以内。请直接输出你的发言内容,不要有任何前缀。""" response = llm.invoke([HumanMessage(content=prompt)]) new_argument = response.content.strip() pro_message = AIMessage(content=new_argument, name="正方") new_messages = state.get('messages', []) + [pro_message] new_pro_args = state.get('pro_arguments', []) + [new_argument]return {"messages": new_messages,"pro_arguments": new_pro_args,"round": state.get('round', 1), # 显式保留轮次 }注意:每个消息我都加上了name属性,这样裁判就能分清是哪一方的发言。format_history函数会读取这个name来生成带角色的历史记录。
节点2:反方Agent
结构几乎一样,只是立场互换,这里不再重复贴代码。
节点3:裁判Agent
裁判有两个职责:
每轮结束时:增加轮次,引导下一轮 最后一轮结束时:总结双方观点,判定胜负
defjudge_node(state: DebateState): current_round = state.get('round', 1) max_rounds = state.get('max_rounds', 3) messages = state.get('messages', [])if current_round >= max_rounds:# 生成最终判定 summary_prompt = f"""请根据以下辩论记录,总结双方核心观点,并判定哪一方获胜(弊大于利 或 利大于弊)。辩论记录:{format_history(messages)}请以JSON格式输出:{{"summary": "总结内容", "winner": "pro"}} 或 {{"summary": "总结内容", "winner": "con"}}""" response = llm.invoke([ SystemMessage(content="你是一位公正的辩论裁判,负责总结并判定胜负。"), HumanMessage(content=summary_prompt) ])# 解析JSON(略) winner = ... # 解析出 "pro" 或 "con" judge_msg = AIMessage(content=f"【裁判总结】...\n获胜方:{'正方'if winner=='pro'else'反方'}", name="裁判")return {"messages": messages + [judge_msg],"round": current_round + 1,"verdict": winner, }else: guide_msg = AIMessage(content=f"第{current_round}轮结束,请继续辩论。", name="裁判")return {"messages": messages + [guide_msg],"round": current_round + 1, }裁判的总结部分用了JSON输出,后面再解析,这样结构清晰,不容易出错。
五、图编排:把节点串起来
节点写好了,关键是让它们按正确的顺序执行,并且裁判能决定什么时候结束。
from langgraph.graph import StateGraph, ENDdefbuild_debate_graph(): workflow = StateGraph(DebateState)# 添加节点 workflow.add_node("pro", pro_node) workflow.add_node("con", con_node) workflow.add_node("judge", judge_node)# 设置入口:正方先发言 workflow.set_entry_point("pro")# 顺序边:正方 -> 裁判,反方 -> 裁判 workflow.add_edge("pro", "judge") workflow.add_edge("con", "judge")# 条件边:裁判之后,根据是否有verdict决定是结束还是继续defnext_speaker(state: DebateState):if state.get("verdict"):return END# 根据轮次奇偶决定下一个发言方# round为偶数时,裁判刚处理完正方,应该轮到反方if state["round"] % 2 == 0:return"con"else:return"pro" workflow.add_conditional_edges("judge", next_speaker, {"pro": "pro","con": "con", END: END })return workflow.compile()这段代码的核心是next_speaker函数。它根据round的奇偶性决定下一个发言的是正方还是反方,同时一旦verdict出现,就走向END结束辩论。
六、跑起来看看
初始化状态:
initial_state = {"messages": [HumanMessage(content="辩论开始:科技发展利大于弊还是弊大于利?")],"pro_arguments": [],"con_arguments": [],"round": 1,"max_rounds": 3,"verdict": "",}执行:
app = build_debate_graph()final_state = app.invoke(initial_state)print(final_state["verdict"])输出示例(经过3轮辩论后):
正方:科技发展极大提升了生产效率,这是利大于弊的铁证。裁判:第1轮结束,请继续辩论。反方:但环境污染和隐私问题日益严重,不能只看效率。裁判:第2轮结束,请继续辩论。正方:新能源技术正在解决环境问题,科技本身不是罪。裁判:第3轮结束,请继续辩论。【裁判总结】正方强调效率提升和技术自我修复能力,反方强调污染和隐私。综合来看,正方论据更充分。获胜方:正方正常跑起来了,倒是这个谁赢谁输我也不知道AI判断的对不对,就先默认正方会赢吧!
七、踩坑实录:这些问题我全遇到了
这个项目虽小,但坑一个没少踩。记下来,下次长记性。
坑1:状态字段丢失,报KeyError: 'round'
现象:执行到 pro_node时报错KeyError: 'round'原因: pro_node返回时没有包含round字段,LangGraph合并状态时以为这个字段被删了解决:每个节点返回的字典里,必须显式包含所有需要保留的字段,即使值没变也要写上去。我后来统一写了 "round": state.get('round', 1)。
坑2:消息列表键名写错,'message' vs 'messages'
现象:裁判节点报错 KeyError: 'message'原因:手滑把 messages写成了单数message解决:全局搜索替换,统一用 messages。Python不报错才怪。
坑3:变量名张冠李戴,new_pro_args未定义
现象: con_node里返回了"pro_arguments": new_pro_args,但new_pro_args根本没定义原因:复制粘贴正方代码时忘了改变量名 解决:写代码时多看一眼,或者用IDE的重构功能。这种低级错误浪费了我半小时。
坑4:裁判的JSON解析失败,忘记import json
现象: NameError: name 'json' is not defined原因:用了 json.loads但没导入解决:文件开头加 import json。小问题,但容易被忽略。
坑5:compute_metrics?不,这里是format_history里拿不到name
现象:历史记录里显示的都是 AIMessage,没有“正方”“反方”原因:创建消息时没有设置 name属性解决:每个节点创建 AIMessage时加上name="正方"或name="反方",裁判的name="裁判"。然后在format_history中用getattr(msg, "name", "AI")获取。
坑6:LLM输出带Markdown代码块,JSON解析失败
现象:裁判总结时,LLM输出的JSON外面包了 json ...原因:大模型喜欢给JSON加代码块 解决:解析前先去掉代码块:
if"```"in text: text = text.split("```")[1]if text.startswith("json"): text = text[4:]text = text.strip()result = json.loads(text)坑7:循环依赖?不,这里是无限递归的条件边
现象:图执行到裁判后就卡住了,不往下走 原因: next_speaker里判断条件写反了,导致一直返回pro,裁判后永远走正方,正方后永远走裁判,死循环解决:仔细检查条件逻辑,用 round的奇偶性控制,并在裁判返回后更新round。
写在最后
以前觉得多Agent系统得用复杂的框架,得写大量的状态机代码。但LangGraph把状态管理、流程编排、条件分支都封装好了,我只需要关心每个Agent的提示词设计和业务逻辑。
正方、反方、裁判三个角色,各自有记忆、有规划,还能循环辩论——这些用传统方式实现至少要写几百行if-else,而用LangGraph,几十行图定义就搞定了。
当然,坑也是真坑。尤其是状态字段的传递,稍不留神就会丢。但踩过一遍之后,反而觉得这种设计很合理:显式返回要更新的字段,避免了隐式修改的副作用。
如果你也想尝试做多角色对话或者多步骤任务,也可以试试LangGraph。从最简单的线性链开始,慢慢加条件分支、加循环,最后会发现,原来“智能体”也可以搭积木一样简单。
这里没有人生导师, 只有一个摸索前行的人。
跑起来,一定要跑起来。这样,风就来了。
夜雨聆风