第四单元——第九课：状态合并

下一课学“状态合并”。上一课两个并行节点各写一个字段这次让它们都往同一个facts列表添加资料。新建unit4_lesson9_reducer.pyfrom operator import add from typing import Annotated, NotRequired, TypedDict from langgraph.graph import START, END, StateGraph class BookstoreState(TypedDict): question: str fact: Annotated[list[str], add] answer: NotRequired[str] def find_discount(state: BookstoreState) - dict: fact 会员买书享受八折优惠。 if 会员 in state[question] else 图书按原价出售。 return {fact: [fact]} def find_returns(state: BookstoreState) - dict: return {fact: [购书后七天内可凭小票申请退货。]} def combine(state: BookstoreState) - dict[str, str]: return {answer: \n.join(sorted(state[fact]))} builder StateGraph(BookstoreState) builder.add_node(find_discount, find_discount) builder.add_node(find_returns, find_returns) builder.add_node(combine, combine) builder.add_edge(START, find_discount) builder.add_edge(START, find_returns) builder.add_edge([find_discount, find_returns], combine) builder.add_edge(combine, END) graph builder.compile() from pathlib import Path image_path Path(__file__).with_name(bookstore_graph.png) image_path.write_bytes(graph.get_graph().draw_mermaid_png()) print(流程图已保存, image_path) result graph.invoke({ question: 会员买书的优惠和退货规则是什么, facts: [], }) print(收集到的资料, result[fact]) print(最终回答\n, result[answer])运行python unit4_lesson9_reducer.py。重点是facts: Annotated[list[str], add]它告诉 LangGraph收到新的facts时用列表相加来追加而不是覆盖。两个并行节点因此都能把资料放进同一个列表combine再统一读取。官方 Reducer 文档sorted()只用于让最终回答的顺序固定不要依赖并行节点谁先完成。执行结果langGraph图课后提问从语法上详细解释facts: Annotated[list[str], add]里的写法答这行可以从里到外读facts: Annotated[list[str], add]facts:是TypedDict中一个字段的类型标注不是赋值。它没有创建列表运行时要用state[facts]访问示例中用facts: []提供初始值。list[str]表示这个字段预期是“字符串列表”例如[会员八折, 七天内可退货]。Annotated[类型, 附加信息]在原有类型上附加信息。所以Annotated[list[str], add]对类型检查而言仍是字符串列表但还带着add这个信息。add来自from operator import add是一个函数本身所以这里写add不写add()。对列表来说add([优惠], [退货])得到[优惠, 退货]。关键区别是**Python 语法只负责标注LangGraph 读取这个标注后才把add当作合并规则。**于是一个节点返回{facts: [优惠]}、另一个返回{facts: [退货]}时两份内容会合并而不是互相覆盖。官方接着问类似add的标注还有什么答有。add不是 LangGraph 专用关键字而是 Python 的operator.add函数。LangGraph 的合并规则也不限于一份固定名单只要函数接收“旧值、新值”两个参数并返回合并后的值就可以用。常见写法from operator import add, or_ from typing import Annotated facts: Annotated[list[str], add] # 两个列表拼接 tags: Annotated[set[str], or_] # 两个集合取并集自动去重 best_score: Annotated[float, max] # 保留较大的分数还可以自己写规则例如“保留较长的回答”def keep_longer(old: str, new: str) - str: return new if len(new) len(old) else old answer: Annotated[str, keep_longer]聊天记录有个特别常用的 LangGraph 规则add_messages它比普通add更适合消息因为能按消息 ID 更新已有消息。当前书店例子的资料列表用add就够了。再问:add_messages怎么引入答这样引入from langgraph.graph.message import add_messages在状态里使用from typing import Annotated, TypedDict from langchain_core.messages import AnyMessage from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list[AnyMessage], add_messages]这里的意思是节点更新messages时使用add_messages合并消息。