2.3 集合与 TypedDict
第一部分：集合（Set）
什么是集合？
集合是无序、不重复的元素集合。在 AI Agent 中用于：
- 去重（已处理的任务ID）
- 成员检查（是否使用过某工具）
- 集合运算（交集、并集、差集）
🎯 小白理解指南：什么是"集合"？
集合就像签到表或投票箱：
- 不重复：一个人只能签一次到，重复签也只算一个人
- 无顺序：不关心谁先来谁后来，只关心"来没来过"
- 查找超快：判断某人是否签过到，一瞬间就能知道
在 AI Agent 中，集合常用于：
- "这个任务处理过没？"（任务去重）
- "这个工具用过没？"（工具追踪）
- "这个网页爬过没？"（避免重复爬取）
python
# 创建集合
tools_used = {"search", "calculator", "search"} # 自动去重
print(tools_used) # {'search', 'calculator'}
# 空集合
empty_set = set() # 注意：{} 是字典！
# 从列表创建
ids = [1, 2, 2, 3, 3, 3]
unique_ids = set(ids) # {1, 2, 3}集合操作
python
# 添加元素
tools = set()
tools.add("search")
tools.add("calculator")
# 删除元素
tools.remove("search") # KeyError if not exists
tools.discard("weather") # No error if not exists
# 成员检查（非常快！O(1)）
if "search" in tools:
print("已使用搜索工具")
# 集合运算
a = {1, 2, 3}
b = {2, 3, 4}
print(a | b) # 并集: {1, 2, 3, 4}
print(a & b) # 交集: {2, 3}
print(a - b) # 差集: {1}
print(a ^ b) # 对称差: {1, 4}🎯 小白理解指南：集合运算是什么意思？
想象两个班级的学生名单：
- 并集
|：两个班所有学生（合并，去重）- 交集
&：两个班都有的学生（同时在两边的）- 差集
-：只在A班、不在B班的学生- 对称差
^：只在一个班的学生（不同时出现的）python班级A = {"小明", "小红", "小刚"} 班级B = {"小红", "小刚", "小美"} 班级A | 班级B # {"小明", "小红", "小刚", "小美"} 所有人 班级A & 班级B # {"小红", "小刚"} 两边都有的人 班级A - 班级B # {"小明"} 只在A班的人
AI 应用：任务去重系统
python
class TaskDeduplicator:
"""任务去重器"""
def __init__(self):
self.processed_tasks: set[str] = set()
def is_duplicate(self, task_id: str) -> bool:
"""检查是否重复"""
return task_id in self.processed_tasks
def mark_processed(self, task_id: str) -> None:
"""标记为已处理"""
self.processed_tasks.add(task_id)
def process_task(self, task_id: str, task_content: str) -> str:
"""处理任务"""
if self.is_duplicate(task_id):
return f"任务 {task_id} 已处理，跳过"
# 处理逻辑
result = f"处理任务: {task_content}"
self.mark_processed(task_id)
return result
# 使用
dedup = TaskDeduplicator()
print(dedup.process_task("task_1", "搜索Python"))
print(dedup.process_task("task_1", "搜索Python")) # 跳过第二部分：TypedDict
🎯 小白理解指南：什么是 TypedDict？
普通字典就像一个随便写的便签——你想写什么键都行，写错了也没人提醒。
TypedDict 就像一个表格模板——规定好了有哪些列、每列填什么类型的数据。
比如"用户信息表"必须有：姓名（文字）、年龄（数字）、邮箱（文字）。如果你漏填或者填错类型，系统会提醒你。
为什么在 LangGraph 中很重要？ LangGraph 的 State（状态）就是用 TypedDict 定义的！它确保每个节点传递的状态都有统一的结构，不会因为拼写错误或少传字段而出bug。
为什么需要 TypedDict？
普通字典没有类型检查：
python
# 问题：字典的键和值类型不明确
config = {
"model": "gpt-4",
"temperature": 0.7,
"max_tokens": 2000
}
# IDE 无法提供补全
# 拼写错误无法检测
temp = config["temprature"] # typo!TypedDict 解决了这个问题：
python
from typing import TypedDict
class AgentConfig(TypedDict):
model: str
temperature: float
max_tokens: int
# 现在有类型检查！
config: AgentConfig = {
"model": "gpt-4",
"temperature": 0.7,
"max_tokens": 2000
}
# IDE 会提供补全和检查
temp = config["temperature"] # ✅可选字段
python
from typing import TypedDict, NotRequired
class AgentState(TypedDict):
messages: list[str] # 必需
context: NotRequired[dict] # 可选
metadata: NotRequired[dict] # 可选
# 有效
state1: AgentState = {"messages": []}
state2: AgentState = {"messages": [], "context": {}}与 LangGraph 的联系
🎯 小白理解指南：Annotated 是什么？
Annotated就是给类型加注释/加说明。
messages: list只说"这是一个列表"messages: Annotated[list, "消息历史"]说"这是一个列表，而且它是消息历史"在 LangGraph 中，Annotated 还可以指定状态合并策略（比如消息是追加还是替换），这在后面的章节会详细讲。
python
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph
from langchain_core.messages import BaseMessage
class MyAgentState(TypedDict):
"""LangGraph Agent 状态"""
messages: Annotated[list[BaseMessage], "消息历史"]
iteration: Annotated[int, "迭代次数"]
next_action: Annotated[str, "下一步动作"]
# 在 LangGraph 中使用
workflow = StateGraph(MyAgentState)实战：完整的状态系统
python
from typing import TypedDict, NotRequired, Literal
from dataclasses import dataclass
from datetime import datetime
# TypedDict 定义状态结构
class ConversationState(TypedDict):
session_id: str
messages: list[dict]
current_step: Literal["greeting", "processing", "responding"]
confidence: float
tools_used: set[str]
metadata: NotRequired[dict]
@dataclass
class StateManager:
"""状态管理器"""
def create_initial_state(self, session_id: str) -> ConversationState:
"""创建初始状态"""
return {
"session_id": session_id,
"messages": [],
"current_step": "greeting",
"confidence": 0.0,
"tools_used": set()
}
def validate_state(self, state: ConversationState) -> bool:
"""验证状态完整性"""
required_keys = {"session_id", "messages", "current_step"}
return all(key in state for key in required_keys)
# 使用
manager = StateManager()
state = manager.create_initial_state("session_001")
print(f"状态有效: {manager.validate_state(state)}")