🔄 状态与 Reducer

class State(TypedDict):
    messages: list[str]   # 没有指定 reducer

def node_a(state):
    return {"messages": ["来自A"]}   # 只返回新值

def node_b(state):
    return {"messages": ["来自B"]}

# node_a 和 node_b 都写了 messages，结果如何？
# 默认行为：后写覆盖先写！state["messages"] 最后只剩 ["来自B"]
# 这几乎从不是你想要的——你想要的是累积 ["来自A", "来自B"]

from typing import Annotated
from typing_extensions import TypedDict
import operator

# reducer 签名：(当前值, 新值) → 合并后的值
def concat_list(current: list, new: list) -> list:
    return current + new

class State(TypedDict):
    # Annotated[类型, reducer] —— 为字段指定合并函数
    messages: Annotated[list[str], concat_list]   # 用自定义 reducer 累积
    count: int                                     # 无 reducer，默认覆盖

# 现在节点只需返回"新增"的部分，不用手动拼接
def node_a(state):
    return {"messages": ["来自A"]}   # 只返回新增，reducer 自动累积！

def node_b(state):
    return {"messages": ["来自B"]}   # 同上

# 两个节点都写 messages，最终: ["来自A", "来自B"]  ✓

# 常用技巧：operator.add 就是列表拼接
class State2(TypedDict):
    items: Annotated[list, operator.add]   # 等价于上面的 concat_list

def add_messages(left, right):
    """消息列表的 reducer。

    比简单拼接更智能：
    1. 自动去重：同 id 的消息会被替换（而非重复）——支持"更新已有消息"
    2. 处理 RemoveMessage：可以删除指定 id 的消息（消息裁剪用）
    3. 自动给无 id 的消息生成 id
    """

# 标准用法：状态里用 Annotated[..., add_messages]
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]   # ← 关键！

# 现在节点只追加新消息，add_messages 自动处理累积/去重/删除
def chat_node(state: State) -> dict:
    # 只返回"新增"的 AIMessage，不用管历史
    return {"messages": [AIMessage(content="你好！")]}

from langgraph.graph import MessagesState

# MessagesState 内部就是：
class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

# 直接用，省得自己写
graph = StateGraph(MessagesState)
# 也可以扩展它，加自己的字段
class MyState(MessagesState):
    user_name: str   # 在 messages 基础上加字段

class BaseChannel(Generic[Value, Update, Checkpoint], ABC):
    """所有通道的基类。"""
    __slots__ = ("key", "typ")

    @property
    @abstractmethod
    def ValueType(self) -> Any:
        """存储的值类型"""

    @property
    @abstractmethod
    def UpdateType(self) -> Any:
        """接收的更新类型"""

    def checkpoint(self) -> Checkpoint | Any:
        """返回可序列化的当前状态快照（用于持久化）"""
    # update() 方法：调用 reducer 聚合新值 —— 这就是 reducer 的执行处
    # get() 方法：返回当前值

# pip install langgraph langchain-core
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.graph.message import MessagesState
from langchain_core.messages import HumanMessage, AIMessage

# ============ ① 无 reducer：覆盖（反面教材）============
class BadState(TypedDict):
    log: list[str]

def append_a(state):
    return {"log": ["A"]}   # 想追加，但...

g1 = StateGraph(BadState)
g1.add_node("a", append_a)
g1.add_node("b", lambda s: {"log": ["B"]})
g1.add_edge(START, "a"); g1.add_edge("a", "b"); g1.add_edge("b", END)
app1 = g1.compile()
print("无reducer:", app1.invoke({"log": []}))   # → {'log': ['B']}  被覆盖了！

# ============ ② 有 reducer：累积 ✓ ============
import operator
class GoodState(TypedDict):
    log: Annotated[list[str], operator.add]   # ← 加了 reducer

g2 = StateGraph(GoodState)
g2.add_node("a", lambda s: {"log": ["A"]})    # 只返回新增
g2.add_node("b", lambda s: {"log": ["B"]})
g2.add_edge(START, "a"); g2.add_edge("a", "b"); g2.add_edge("b", END)
app2 = g2.compile()
print("有reducer:", app2.invoke({"log": []}))  # → {'log': ['A', 'B']}  累积！

# ============ ③ add_messages：消息累积（Agent 标准做法）============
class ChatState(TypedDict):
    messages: Annotated[list, add_messages]

def chatbot(state: ChatState) -> dict:
    user_msg = state["messages"][-1].content
    return {"messages": [AIMessage(content=f"你说的是：{user_msg}")]}

g3 = StateGraph(ChatState)
g3.add_node("chatbot", chatbot)
g3.add_edge(START, "chatbot"); g3.add_edge("chatbot", END)
app3 = g3.compile()
result = app3.invoke({"messages": [HumanMessage(content="你好")]})
print("\n消息历史:")
for m in result["messages"]:
    print(f"  [{m.type}] {m.content}")
# → [human] 你好
#   [ai] 你说的是：你好   ← add_messages 自动累积了！

# ============ ④ 直接用 MessagesState（更省事）============
g4 = StateGraph(MessagesState)   # 内置 messages 字段
g4.add_node("bot", chatbot)
g4.add_edge(START, "bot"); g4.add_edge("bot", END)
app4 = g4.compile()
# 等价于上面的 ChatState 写法

from langchain_core.messages import RemoveMessage

def prune_old_messages(state):
    # 删除除了最后 N 条之外的所有消息
    keep = 6
    messages = state["messages"]
    delete_ids = [RemoveMessage(id=m.id) for m in messages[:-keep]]
    return {"messages": delete_ids}   # add_messages 会处理 RemoveMessage

# 这就是 Agent 控制上下文长度的标准技巧
# OpenCode 的 compaction 是更高级的版本（OC5）