2.2 字典：Agent 状态的核心数据结构

引言：Agent 的"大脑"

在 LangGraph 中，State（状态） 是核心概念。每个 Agent 的状态本质上就是一个字典：

python

{
    "messages": [...],
    "current_step": "analyzing",
    "confidence": 0.85,
    "tools_used": ["search", "calculator"],
    "metadata": {"user_id": "123"}
}

字典是 Python 中最灵活、最强大的数据结构，理解字典就是理解如何管理 Agent 的状态。

🎯 小白理解指南：什么是"字典"？
Python 的字典就像真实世界的字典或通讯录：
通讯录里你用名字找电话号码
Python 字典里你用键（key）找值（value）
python
# 通讯录的例子
contacts = {
    "张三": "138-1234-5678",
    "李四": "139-8765-4321"
}
print(contacts["张三"])  # 输出：138-1234-5678
1
2
3
4
5
6
为什么 AI Agent 爱用字典？ 因为 Agent 需要同时记住很多不同类型的信息：当前在做什么、用户说了什么、用了哪些工具……字典可以把这些信息分门别类存好，需要时立刻取出来。

学习目标

✅ 掌握字典的创建和操作
✅ 理解字典的高级用法
✅ 掌握 defaultdict 和 Counter
✅ 实战：构建 Agent 状态管理系统

第一部分：字典基础

创建字典

python

# 空字典
config: dict = {}
config: dict[str, int] = {}  # 带类型注解

# 字面量创建
agent_config = {
    "model": "gpt-5",
    "temperature": 0.7,
    "max_tokens": 2000
}

# dict() 构造函数
config = dict(model="gpt-5", temperature=0.7)

# 从键值对列表创建
pairs = [("a", 1), ("b", 2)]
d = dict(pairs)  # {'a': 1, 'b': 2}

基本操作

python

config = {"model": "gpt-5-turbo", "temperature": 0.7}

# 访问值
print(config["model"])  # 'gpt-3.5-turbo'

# 安全访问（使用 get）
temp = config.get("temperature")  # 0.7
max_tokens = config.get("max_tokens", 2000)  # 不存在返回默认值

# 🎯 小白提示：get() 是"安全访问"
# config["xxx"] 如果键不存在会报错崩溃
# config.get("xxx") 键不存在会返回 None，程序不会崩
# config.get("xxx", 默认值) 键不存在返回你指定的默认值

# 添加/修改键值对
config["max_tokens"] = 2000
config["temperature"] = 0.5  # 修改

# 删除键
del config["temperature"]
removed = config.pop("model")  # 删除并返回值

# 检查键是否存在
if "model" in config:
    print("模型已配置")

# 获取所有键、值、键值对
keys = config.keys()      # dict_keys(['max_tokens'])
values = config.values()  # dict_values([2000])
items = config.items()    # dict_items([('max_tokens', 2000)])

第二部分：字典高级操作

🎯 小白理解指南：什么是"字典推导式"？
和列表推导式一样，字典推导式是快速创建字典的语法糖。
语法模板：{键: 值 for 变量 in 序列}
比如你想创建一个"数字→平方"的对照表，传统方式要写好几行，推导式一行搞定！

字典推导式

python

# 基本字典推导式
squares = {x: x**2 for x in range(5)}
# {0: 0, 1: 1, 2: 4, 3: 9, 4: 16}

# 带条件
even_squares = {x: x**2 for x in range(10) if x % 2 == 0}

# AI 应用：从列表创建索引映射
tools = ["search", "calculator", "weather"]
tool_map = {tool: i for i, tool in enumerate(tools)}
# {'search': 0, 'calculator': 1, 'weather': 2}

# 反转字典
original = {"a": 1, "b": 2, "c": 3}
reversed_dict = {v: k for k, v in original.items()}
# {1: 'a', 2: 'b', 3: 'c'}

合并字典

python

# Python 3.9+ 使用 | 运算符
dict1 = {"a": 1, "b": 2}
dict2 = {"c": 3, "d": 4}
merged = dict1 | dict2  # {'a': 1, 'b': 2, 'c': 3, 'd': 4}

# 更新（后者覆盖前者）
dict1 = {"a": 1, "b": 2}
dict2 = {"b": 3, "c": 4}
merged = dict1 | dict2  # {'a': 1, 'b': 3, 'c': 4}

# update() 方法
config = {"model": "gpt-3.5-turbo"}
config.update({"temperature": 0.7, "max_tokens": 2000})

# ** 解包（推荐）
defaults = {"temperature": 0.7, "max_tokens": 2000}
user_config = {"temperature": 0.5}
final_config = {**defaults, **user_config}
# {'temperature': 0.5, 'max_tokens': 2000}

嵌套字典

🎯 小白理解指南：什么是"嵌套字典"？
就是字典里面套字典，像俄罗斯套娃一样。
想象一个公司的组织架构：
公司 → 部门 → 员工
每一层都可以用字典表示
在 AI Agent 中，状态信息往往是多层嵌套的：用户信息里有姓名、ID；对话信息里有消息列表、轮数……

python

agent_state = {
    "user": {
        "id": "123",
        "name": "Alice"
    },
    "conversation": {
        "messages": [],
        "turn_count": 0
    },
    "tools": {
        "available": ["search", "calculator"],
        "used": []
    }
}

# 访问嵌套值
user_name = agent_state["user"]["name"]

# 安全访问嵌套值
from typing import Any

def get_nested(d: dict, *keys: str, default: Any = None) -> Any:
    """安全获取嵌套字典的值"""
    for key in keys:
        if isinstance(d, dict):
            d = d.get(key, default)
        else:
            return default
    return d

# 使用
name = get_nested(agent_state, "user", "name")  # 'Alice'
missing = get_nested(agent_state, "user", "age", default=0)  # 0

第三部分：defaultdict 和 Counter

🎯 小白理解指南：defaultdict 解决什么问题？
普通字典访问不存在的键会报错。但有时候我们想：如果键不存在，就自动创建一个默认值。
比如统计单词出现次数：
普通字典：先检查键存不存在，不存在就设为 0，然后 +1
defaultdict：直接 +1，不存在会自动变成 0 再 +1
这就像自动售货机：普通字典是"没货就报错"，defaultdict 是"没货就自动补上再卖"。

defaultdict

python

from collections import defaultdict

# 问题：普通字典需要检查键是否存在
word_count = {}
for word in ["apple", "banana", "apple"]:
    if word in word_count:
        word_count[word] += 1
    else:
        word_count[word] = 1

# 解决方案：defaultdict
word_count = defaultdict(int)  # 默认值为 0
for word in ["apple", "banana", "apple"]:
    word_count[word] += 1

print(dict(word_count))  # {'apple': 2, 'banana': 1}

# 其他默认类型
# defaultdict(list) - 默认值为 []
# defaultdict(set) - 默认值为 set()
# defaultdict(dict) - 默认值为 {}

# AI 应用：按类别组织工具
from collections import defaultdict

tools_by_category = defaultdict(list)
tools_by_category["search"].append("google_search")
tools_by_category["search"].append("bing_search")
tools_by_category["math"].append("calculator")

Counter

🎯 小白理解指南：Counter 是什么？
Counter 是 Python 内置的计数神器——专门用来数"每个东西出现了几次"。
想象你在数一筐水果：苹果3个、香蕉2个、樱桃1个。 Counter 可以一行代码帮你数好！
它还能告诉你"出现最多的是什么"（most_common），在统计 AI 工具使用频率时特别有用。

python

from collections import Counter

# 计数器
words = ["apple", "banana", "apple", "cherry", "banana", "apple"]
counter = Counter(words)
print(counter)  # Counter({'apple': 3, 'banana': 2, 'cherry': 1})

# 最常见的元素
print(counter.most_common(2))  # [('apple', 3), ('banana', 2)]

# AI 应用：统计 Agent 使用的工具
tool_usage = Counter()
tool_usage["search"] += 1
tool_usage["calculator"] += 1
tool_usage["search"] += 1

print(f"搜索工具使用了 {tool_usage['search']} 次")

第四部分：实战 - Agent 状态管理系统

python

"""
Agent 状态管理系统
演示字典在 AI Agent 中的核心应用
"""

from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict, Counter
import json


@dataclass
class AgentState:
    """Agent 状态数据类"""
    
    # 必需字段
    session_id: str
    
    # 消息历史
    messages: List[Dict[str, str]] = field(default_factory=list)
    
    # 当前状态
    current_step: str = "init"
    confidence: float = 0.0
    
    # 工具使用
    tools_available: List[str] = field(default_factory=list)
    tools_used: List[str] = field(default_factory=list)
    
    # 元数据
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    # 统计信息
    stats: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
    
    # 时间戳
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)


class StateManager:
    """状态管理器"""
    
    def __init__(self):
        """初始化状态管理器"""
        self.states: Dict[str, AgentState] = {}
    
    def create_state(
        self,
        session_id: str,
        tools: Optional[List[str]] = None
    ) -> AgentState:
        """
        创建新状态
        
        Args:
            session_id: 会话ID
            tools: 可用工具列表
        
        Returns:
            新创建的状态
        """
        state = AgentState(
            session_id=session_id,
            tools_available=tools or []
        )
        self.states[session_id] = state
        return state
    
    def get_state(self, session_id: str) -> Optional[AgentState]:
        """获取状态"""
        return self.states.get(session_id)
    
    def update_state(
        self,
        session_id: str,
        updates: Dict[str, Any]
    ) -> None:
        """
        更新状态
        
        Args:
            session_id: 会话ID
            updates: 要更新的字段
        """
        state = self.states.get(session_id)
        if not state:
            raise ValueError(f"状态 {session_id} 不存在")
        
        # 更新字段
        for key, value in updates.items():
            if hasattr(state, key):
                setattr(state, key, value)
        
        state.updated_at = datetime.now()
    
    def add_message(
        self,
        session_id: str,
        role: str,
        content: str
    ) -> None:
        """添加消息到历史"""
        state = self.states.get(session_id)
        if state:
            state.messages.append({
                "role": role,
                "content": content,
                "timestamp": datetime.now().isoformat()
            })
            state.stats["message_count"] += 1
    
    def record_tool_use(self, session_id: str, tool_name: str) -> None:
        """记录工具使用"""
        state = self.states.get(session_id)
        if state:
            state.tools_used.append(tool_name)
            state.stats[f"tool_{tool_name}"] += 1
    
    def get_stats(self, session_id: str) -> Dict:
        """获取统计信息"""
        state = self.states.get(session_id)
        if not state:
            return {}
        
        # 工具使用统计
        tool_counter = Counter(state.tools_used)
        
        return {
            "session_id": session_id,
            "total_messages": len(state.messages),
            "current_step": state.current_step,
            "confidence": state.confidence,
            "tools_used_count": dict(tool_counter),
            "session_duration": (
                state.updated_at - state.created_at
            ).total_seconds(),
        }
    
    def export_state(self, session_id: str) -> str:
        """导出状态为 JSON"""
        state = self.states.get(session_id)
        if not state:
            return "{}"
        
        return json.dumps({
            "session_id": state.session_id,
            "messages": state.messages,
            "current_step": state.current_step,
            "confidence": state.confidence,
            "tools_used": state.tools_used,
            "metadata": state.metadata,
            "stats": dict(state.stats),
        }, indent=2, default=str)


# 使用示例
def main():
    """演示状态管理系统"""
    manager = StateManager()
    
    # 创建状态
    state = manager.create_state(
        session_id="session_001",
        tools=["search", "calculator", "weather"]
    )
    print(f"创建状态: {state.session_id}")
    
    # 添加消息
    manager.add_message("session_001", "user", "你好")
    manager.add_message("session_001", "assistant", "您好！")
    manager.add_message("session_001", "user", "天气怎么样？")
    
    # 记录工具使用
    manager.record_tool_use("session_001", "weather")
    manager.record_tool_use("session_001", "search")
    manager.record_tool_use("session_001", "weather")
    
    # 更新状态
    manager.update_state("session_001", {
        "current_step": "processing",
        "confidence": 0.85
    })
    
    # 获取统计
    stats = manager.get_stats("session_001")
    print("\n统计信息:")
    print(json.dumps(stats, indent=2))
    
    # 导出状态
    print("\n导出状态:")
    print(manager.export_state("session_001"))


if __name__ == "__main__":
    main()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197

本节总结

核心要点

字典：键值对，快速查找
get() 方法：安全访问，避免 KeyError
字典推导式：简洁创建字典
defaultdict：自动初始化默认值
Counter：计数统计

在 AI Agent 中的应用

操作	应用场景
字典	Agent 状态、配置、API 响应
get()	安全访问可选配置
**解包	合并默认配置和用户配置
defaultdict	分组数据、计数
Counter	统计工具使用频率

下一节：2.3 集合与 TypedDict

2.2 字典：Agent 状态的核心数据结构 ​

引言：Agent 的"大脑" ​

学习目标 ​

第一部分：字典基础 ​

创建字典 ​

基本操作 ​

第二部分：字典高级操作 ​

字典推导式 ​

合并字典 ​

嵌套字典 ​

第三部分：defaultdict 和 Counter ​

defaultdict ​

Counter ​

第四部分：实战 - Agent 状态管理系统 ​

本节总结 ​

核心要点 ​

在 AI Agent 中的应用 ​

2.2 字典：Agent 状态的核心数据结构

引言：Agent 的"大脑"

学习目标

第一部分：字典基础

创建字典

基本操作

第二部分：字典高级操作

字典推导式

合并字典

嵌套字典

第三部分：defaultdict 和 Counter

defaultdict

Counter

第四部分：实战 - Agent 状态管理系统

本节总结

核心要点

在 AI Agent 中的应用