0.1 Python 核心基础
学习目标 对于有一定 Python 基础的朋友,本章可以跳过,从第一章开始你的智能体搭建之旅。
本章专为 LangGraph 和 AI Agent 开发设计,通过实战代码快速掌握 Python 核心语法。我们采用"先看代码,后学原理"的方式,每个概念都关联实际的 Agent 开发场景。
本章内容
- ✅ 掌握 Python 变量与数据类型
- ✅ 理解数据结构(List、Dict、Tuple、Set)在 Agent 中的应用
- ✅ 学会控制流(条件判断、循环)
- ✅ 掌握函数定义与 Lambda 表达式
- ✅ 理解类型提示(为 LangGraph 做准备)
📚 术语表
| 术语名称 | LangGraph 定义和解读 | Python 定义和说明 | 重要程度 |
|---|---|---|---|
| State | Agent的状态字典,存储消息历史、意图、上下文等信息。LangGraph中每个节点接收state并返回更新后的state | 字典(dict)类型,键值对集合,可变数据结构 | ⭐⭐⭐⭐⭐ |
| TypedDict | 定义严格的State结构,类似TypeScript的Interface。确保状态字段类型安全,便于IDE智能提示 | typing模块提供的类型,为字典添加类型注解,支持静态类型检查 | ⭐⭐⭐⭐⭐ |
| Lambda | 用于条件边(Conditional Edge)的简洁路由函数,如lambda s: "next_node",快速定义节点转换逻辑 | 匿名函数,格式lambda 参数: 表达式,适合简单单行逻辑 | ⭐⭐⭐⭐ |
| List | 存储消息历史(messages)、工具调用记录等有序数据。Agent核心数据结构 | 有序可变序列,支持索引、切片、append等操作 | ⭐⭐⭐⭐⭐ |
| Dict | Agent状态的基础结构,键值对存储配置、上下文、元数据 | 无序可变映射,通过键访问值,支持get/items/keys等方法 | ⭐⭐⭐⭐⭐ |
| f-string | 构建LLM提示词、格式化日志输出的现代方式,如f"用户{name}的意图是{intent}" | 格式化字符串字面量,f"...{变量}..."支持表达式求值 | ⭐⭐⭐⭐ |
| Type Hints | LangGraph要求的类型注解,定义节点函数签名def node(state: AgentState) -> AgentState | Python 3.5+特性,为变量和函数添加类型信息,增强代码可读性 | ⭐⭐⭐⭐⭐ |
| Literal | 限定字段只能是特定值,如role: Literal["user", "assistant"]避免拼写错误 | typing.Literal类型,定义字面量类型,编译时检查值合法性 | ⭐⭐⭐⭐ |
| Optional | 表示函数可能返回None,如Optional[str]等价于Union[str, None] | typing.Optional类型,明确告知调用方需处理空值情况 | ⭐⭐⭐⭐ |
| List Comprehension | 简洁处理消息过滤、数据转换,如[msg for msg in messages if msg["role"]=="user"] | 列表推导式,一行代码完成循环+过滤+映射,性能优于传统循环 | ⭐⭐⭐⭐ |
1. 环境准备:开始你的第一行代码
推荐开发环境
选项 1:在线运行(最简单)
- 本网站内置 Python 运行环境,无需安装
- 需要运行 LangGraph 代码时,请先配置 API Key
选项 2:本地 Jupyter Lab(推荐学习。可搭配 Anaconda 使用)
# 安装 Jupyter Lab
pip install jupyterlab
# 启动
jupyter lab选项 3:Cursor / VS Code(推荐开发)
# 安装扩展:
# 1. Python (by Microsoft)
# 2. Jupyter (by Microsoft)Hello, AI World!
运行你的第一个 Python 程序:
# 验证环境
import sys
print(f"Python 版本: {sys.version}")
print(f"🤖 你好,AI Agent 世界!")预期结果:
Python 版本: 3.13.0 (main, Oct 7 2024, 05:02:14) [Clang 15.0.0]
🤖 你好,AI Agent 世界!2. 变量与数据类型
30秒快速上手:最简变量示例
如果你是完全的 Python 小白,先看这个最简单的例子:
# 最简单的变量例子 - 就像给东西贴标签
name = "小明" # 文字用引号括起来
age = 18 # 数字直接写
height = 1.75 # 小数点的数字
is_student = True # True 或 False(注意大写)
# 打印出来看看
print(name) # 输出:小明
print(age) # 输出:18
print(height) # 输出:1.75
print(is_student) # 输出:True
# 组合打印
print(name, "今年", age, "岁") # 输出:小明 今年 18 岁💡 新手提示:
- 变量名就像给数据起个名字,方便后面使用
=是赋值,把右边的值给左边的变量print()用来显示结果
2.1 变量:存储 Agent 状态
在 AI Agent 中,变量用于存储用户消息、系统状态、对话历史等关键信息。
# 代码示例:Agent 基本状态
user_message = "帮我查一下今天的天气"
user_id = "user_12345"
message_count = 3
is_authenticated = True
confidence_score = 0.85
print(f"用户消息: {user_message}")
print(f"用户ID: {user_id}")
print(f"消息数量: {message_count}")
print(f"已认证: {is_authenticated}")
print(f"置信度: {confidence_score}")
# 查看类型
print(f"\n数据类型:")
print(f" user_message: {type(user_message)}")
print(f" message_count: {type(message_count)}")
print(f" confidence_score: {type(confidence_score)}")
print(f" is_authenticated: {type(is_authenticated)}")预期结果:
用户消息: 帮我查一下今天的天气
用户ID: user_12345
消息数量: 3
已认证: True
置信度: 0.85
数据类型:
user_message: <class 'str'>
message_count: <class 'int'>
confidence_score: <class 'float'>
is_authenticated: <class 'bool'>核心数据类型
| 类型 | 说明 | AI 应用场景 |
|---|---|---|
str | 字符串(文本) | 用户消息、AI 回复、提示词 |
int | 整数 | 消息计数、步骤编号、重试次数 |
float | 浮点数 | 置信度分数、温度参数、相似度 |
bool | 布尔值 | 条件判断、状态标志、工具调用成功与否 |
None | 空值 | 可选参数、未初始化的状态 |
2.2 字符串操作:处理文本
# 代码示例:字符串在 Agent 中的常见操作
user_input = " 帮我预订明天去北京的机票 "
# 清理输入
cleaned = user_input.strip() # 去除首尾空格
print(f"清理后: '{cleaned}'")
# 大小写转换
lowercase = cleaned.lower()
uppercase = cleaned.upper()
print(f"小写: {lowercase}")
print(f"大写: {uppercase}")
# 字符串拼接
prompt = f"用户请求:{cleaned}\n请分析意图并给出建议。"
print(f"\n构建的提示词:\n{prompt}")
# 字符串分割
keywords = "天气,航班,酒店,餐厅"
keyword_list = keywords.split(",")
print(f"\n关键词列表: {keyword_list}")
# 检查子串
if "预订" in cleaned:
print("\n✅ 检测到预订意图")
# 字符串替换
sanitized = cleaned.replace("明天", "2024-10-28")
print(f"日期规范化: {sanitized}")预期结果:
清理后: '帮我预订明天去北京的机票'
小写: 帮我预订明天去北京的机票
大写: 帮我预订明天去北京的机票
构建的提示词:
用户请求:帮我预订明天去北京的机票
请分析意图并给出建议。
关键词列表: ['天气', '航班', '酒店', '餐厅']
✅ 检测到预订意图
日期规范化: 帮我预订2024-10-28去北京的机票2.3 f-string:现代字符串格式化
# 代码示例:构建 AI 提示词
user_name = "Alice"
city = "北京"
temperature = 0.7
max_tokens = 1000
# f-string 基础
greeting = f"你好,{user_name}!"
print(greeting)
# 表达式求值
tokens_used = 850
remaining = f"已使用 {tokens_used}/{max_tokens} tokens ({tokens_used/max_tokens:.1%})"
print(remaining)
# 多行 f-string(构建复杂提示词)
system_prompt = f"""你是一个智能助手。
用户信息:
- 姓名:{user_name}
- 位置:{city}
参数设置:
- Temperature: {temperature}
- Max Tokens: {max_tokens}
请根据用户需求提供帮助。"""
print(f"\n{system_prompt}")
# 格式化数字
pi = 3.14159265359
print(f"\nπ ≈ {pi:.2f}") # 保留2位小数
print(f"π ≈ {pi:.5f}") # 保留5位小数预期结果:
你好,Alice!
已使用 850/1000 tokens (85.0%)
你是一个智能助手。
用户信息:
- 姓名:Alice
- 位置:北京
参数设置:
- Temperature: 0.7
- Max Tokens: 1000
请根据用户需求提供帮助。
π ≈ 3.14
π ≈ 3.141593. 数据结构:组织 Agent 数据
🚀 30秒快速上手:最简列表和字典
如果你是 Python 小白,先看这两个最常用的数据结构:
# 列表 - 就像一个有序的购物清单
fruits = ["苹果", "香蕉", "橙子"]
print(fruits[0]) # 输出:苹果(第一个,从0开始数)
print(fruits[-1]) # 输出:橙子(最后一个)
fruits.append("葡萄") # 添加新水果
print(fruits) # 输出:['苹果', '香蕉', '橙子', '葡萄']
# 字典 - 就像通讯录,有名字就能找到对应信息
student = {"姓名": "小明", "年龄": 18, "成绩": 95}
print(student["姓名"]) # 输出:小明
student["年龄"] = 19 # 修改年龄
print(student) # 输出:{'姓名': '小明', '年龄': 19, '成绩': 95}💡 新手提示:
- 列表用方括号
[],按顺序存放多个东西 - 字典用花括号
{},用"名字:值"的方式存放,方便查找
3.1 列表(List):对话历史管理
列表是 AI Agent 中最常用的数据结构,用于存储消息序列、工具调用记录等。
# 代码示例:管理对话历史
messages = []
# 添加用户消息
messages.append({"role": "user", "content": "你好"})
messages.append({"role": "assistant", "content": "你好!有什么可以帮你的吗?"})
messages.append({"role": "user", "content": "今天天气怎么样?"})
# 访问消息
print("=== 对话历史 ===")
print(f"总消息数: {len(messages)}")
print(f"第一条消息: {messages[0]}")
print(f"最后一条消息: {messages[-1]}")
print(f"最近两条消息: {messages[-2:]}")
# 遍历消息
print("\n=== 完整对话 ===")
for i, msg in enumerate(messages, 1):
role_icon = "👤" if msg["role"] == "user" else "🤖"
print(f"{i}. {role_icon} {msg['content']}")
# 列表操作
messages.insert(0, {"role": "system", "content": "你是一个友好的助手"})
print(f"\n添加系统消息后,总消息数: {len(messages)}")
# 删除消息
removed = messages.pop(1) # 删除索引1的消息
print(f"删除的消息: {removed}")
# 列表推导式(过滤用户消息)
user_messages = [msg for msg in messages if msg["role"] == "user"]
print(f"\n用户消息: {user_messages}")预期结果:
=== 对话历史 ===
总消息数: 3
第一条消息: {'role': 'user', 'content': '你好'}
最后一条消息: {'role': 'user', 'content': '今天天气怎么样?'}
最近两条消息: [{'role': 'assistant', 'content': '你好!有什么可以帮你的吗?'}, {'role': 'user', 'content': '今天天气怎么样?'}]
=== 完整对话 ===
1. 👤 你好
2. 🤖 你好!有什么可以帮你的吗?
3. 👤 今天天气怎么样?
添加系统消息后,总消息数: 4
删除的消息: {'role': 'user', 'content': '你好'}
用户消息: [{'role': 'user', 'content': '今天天气怎么样?'}]3.2 字典(Dict):Agent 状态管理
字典是 LangGraph 的核心数据结构,用于存储 Agent 的状态信息。
# 代码示例:LangGraph 状态对象
agent_state = {
"user_id": "user_12345",
"session_id": "sess_abc",
"messages": [
{"role": "user", "content": "帮我查天气"},
{"role": "assistant", "content": "好的,请问是哪个城市?"}
],
"context": {
"intent": "weather_query",
"location": None,
"confidence": 0.92
},
"tool_calls": [],
"metadata": {
"start_time": "2024-10-27 10:00:00",
"turn_count": 1
}
}
# 访问数据
print(f"用户ID: {agent_state['user_id']}")
print(f"意图: {agent_state['context']['intent']}")
print(f"置信度: {agent_state['context']['confidence']:.1%}")
# 更新状态
agent_state['context']['location'] = "北京"
agent_state['metadata']['turn_count'] += 1
print(f"\n更新后的位置: {agent_state['context']['location']}")
print(f"对话轮次: {agent_state['metadata']['turn_count']}")
# 添加新字段
agent_state['context']['weather_data'] = {
"temperature": 22,
"condition": "晴朗"
}
# 获取所有键
print(f"\n状态顶层字段: {list(agent_state.keys())}")
print(f"上下文字段: {list(agent_state['context'].keys())}")
# 安全访问(使用 get)
api_key = agent_state.get('api_key', 'NOT_SET')
print(f"\nAPI Key: {api_key}")
# 检查键是否存在
if 'tool_calls' in agent_state:
print("✅ 状态包含 tool_calls 字段")
# 字典遍历
print("\n=== 元数据详情 ===")
for key, value in agent_state['metadata'].items():
print(f" {key}: {value}")预期结果:
用户ID: user_12345
意图: weather_query
置信度: 92.0%
更新后的位置: 北京
对话轮次: 2
状态顶层字段: ['user_id', 'session_id', 'messages', 'context', 'tool_calls', 'metadata']
上下文字段: ['intent', 'location', 'confidence', 'weather_data']
API Key: NOT_SET
✅ 状态包含 tool_calls 字段
=== 元数据详情 ===
start_time: 2024-10-27 10:00:00
turn_count: 23.3 元组(Tuple):不可变数据
# 代码示例:消息格式、配置常量
# 元组用于固定格式的数据
message_format = ("role", "content", "timestamp")
print(f"消息格式: {message_format}")
# 解包元组
role, content, timestamp = message_format
print(f"字段1: {role}, 字段2: {content}, 字段3: {timestamp}")
# 多返回值(实际是元组)
def parse_message(text):
"""解析消息,返回意图和实体"""
intent = "weather_query"
entity = "北京"
confidence = 0.95
return intent, entity, confidence # 返回元组
intent, entity, conf = parse_message("北京天气")
print(f"\n解析结果: 意图={intent}, 实体={entity}, 置信度={conf}")
# 元组不可修改
try:
message_format[0] = "new_role"
except TypeError as e:
print(f"\n❌ 元组不可修改: {e}")
# 元组作为字典键(列表不行)
cache = {
("user_123", "weather"): "晴天 22°C",
("user_456", "weather"): "多云 18°C"
}
print(f"\n缓存查询: {cache[('user_123', 'weather')]}")预期结果:
消息格式: ('role', 'content', 'timestamp')
字段1: role, 字段2: content, 字段3: timestamp
解析结果: 意图=weather_query, 实体=北京, 置信度=0.95
❌ 元组不可修改: 'tuple' object does not support item assignment
缓存查询: 晴天 22°C3.4 集合(Set):去重与快速查找
# 代码示例:管理用户兴趣标签
user_interests = {"科技", "旅游", "美食", "科技", "运动"} # 自动去重
print(f"用户兴趣(已去重): {user_interests}")
# 添加兴趣
user_interests.add("阅读")
user_interests.add("科技") # 重复添加无效
print(f"添加后: {user_interests}")
# 集合运算
available_topics = {"科技", "美食", "娱乐", "财经"}
# 交集:用户感兴趣且可用的话题
common = user_interests & available_topics
print(f"\n推荐话题(交集): {common}")
# 并集:所有相关话题
all_topics = user_interests | available_topics
print(f"所有话题(并集): {all_topics}")
# 差集:用户兴趣但不可用的
missing = user_interests - available_topics
print(f"缺失话题(差集): {missing}")
# 快速检查
if "科技" in user_interests:
print("\n✅ 用户对科技感兴趣")
# 转换:列表 → 集合(去重)
message_keywords = ["天气", "查询", "天气", "北京", "查询"]
unique_keywords = set(message_keywords)
print(f"\n关键词去重: {list(unique_keywords)}")预期结果:
用户兴趣(已去重): {'运动', '旅游', '科技', '美食'}
添加后: {'运动', '旅游', '阅读', '科技', '美食'}
推荐话题(交集): {'科技', '美食'}
所有话题(并集): {'娱乐', '运动', '旅游', '阅读', '财经', '科技', '美食'}
缺失话题(差集): {'运动', '旅游', '阅读'}
✅ 用户对科技感兴趣
关键词去重: ['查询', '天气', '北京']📊 数据结构选择指南
| 数据结构 | 有序 | 可变 | 去重 | 使用场景 |
|---|---|---|---|---|
| List | ✅ | ✅ | ❌ | 对话历史、消息队列、工具调用记录 |
| Dict | ❌ | ✅ | Key去重 | Agent 状态、配置参数、API 响应 |
| Tuple | ✅ | ❌ | ❌ | 固定格式、函数返回、常量定义 |
| Set | ❌ | ✅ | ✅ | 标签去重、权限检查、集合运算 |
4. 控制流:Agent 决策逻辑
🚀 30秒快速上手:最简 if 和 for
如果你是 Python 小白,先看这两个最基础的控制语句:
# if - 条件判断,就像"如果...就..."
score = 85
if score >= 90:
print("优秀")
elif score >= 60:
print("及格") # 这个会被执行
else:
print("不及格")
# for - 循环,重复做事情
fruits = ["苹果", "香蕉", "橙子"]
for fruit in fruits:
print("我喜欢", fruit)
# 输出:
# 我喜欢 苹果
# 我喜欢 香蕉
# 我喜欢 橙子💡 新手提示:
if后面的代码要缩进(按 Tab 键或 4 个空格)for会依次取出列表里的每一项
4.1 条件判断:路由与决策
在 LangGraph 中,条件判断用于实现条件边(Conditional Edge),决定 Agent 的下一步行为。
# 代码示例:根据用户意图路由
def route_by_intent(user_message):
"""根据用户消息路由到不同节点"""
# 【设计思想】先统一转小写,避免大小写不匹配问题
# 这是文本预处理的标准实践
message_lower = user_message.lower()
# 【核心逻辑】基于关键词匹配的意图识别
# 类似决策树:从最明确的意图开始判断
if "天气" in message_lower:
# 高置信度(0.9):天气是单一明确的意图
intent = "weather_query"
next_node = "weather_tool"
confidence = 0.9
elif "预订" in message_lower or "订票" in message_lower:
# 稍低置信度(0.85):预订可能需要更多澄清
intent = "booking"
next_node = "booking_tool"
confidence = 0.85
elif "计算" in message_lower or "=" in message_lower:
# 最高置信度(0.95):计算特征非常明显
intent = "calculator"
next_node = "calculator_tool"
confidence = 0.95
else:
# 【回退策略】未匹配时交给通用LLM处理
# 低置信度(0.5)提示需要人工审核
intent = "general_chat"
next_node = "llm_response"
confidence = 0.5
# 【返回结构】包含决策结果和置信度,便于后续审计和调试
return {
"intent": intent,
"next_node": next_node,
"confidence": confidence
}
# 测试路由
test_messages = [
"今天北京天气怎么样?",
"帮我预订明天去上海的机票",
"计算 123 + 456",
"你好,介绍一下你自己"
]
print("=== Intent Routing ===")
for msg in test_messages:
result = route_by_intent(msg)
print(f"\n消息: {msg}")
print(f" → 意图: {result['intent']}")
print(f" → 下一节点: {result['next_node']}")
print(f" → 置信度: {result['confidence']:.0%}")预期结果:
=== Intent Routing ===
消息: 今天北京天气怎么样?
→ 意图: weather_query
→ 下一节点: weather_tool
→ 置信度: 90%
消息: 帮我预订明天去上海的机票
→ 意图: booking
→ 下一节点: booking_tool
→ 置信度: 85%
消息: 计算 123 + 456
→ 意图: calculator
→ 下一节点: calculator_tool
→ 置信度: 95%
消息: 你好,介绍一下你自己
→ 意图: general_chat
→ 下一节点: llm_response
→ 置信度: 50%4.2 三元运算符:简洁的条件表达式
# 代码示例:快速条件赋值
def evaluate_response_quality(confidence_score):
"""评估响应质量"""
# 【传统写法】标准 if-else 分支
if confidence_score > 0.8:
quality = "高质量"
else:
quality = "需要确认"
# 【三元运算符】一行代码实现条件赋值
# 格式: 值1 if 条件 else 值2
# 适用于简单的二选一场景,代码更简洁
quality_concise = "高质量" if confidence_score > 0.8 else "需要确认"
# 【嵌套三元运算符】实现多级分类
# 逻辑链: >0.9→优秀, >0.7→良好, 其他→一般
# 注意:嵌套超过2层会降低可读性,应改用if-elif
level = "优秀" if confidence_score > 0.9 else "良好" if confidence_score > 0.7 else "一般"
return {
"quality": quality_concise,
"level": level,
"should_confirm": confidence_score < 0.8 # 布尔表达式直接作为值
}
# 测试
scores = [0.95, 0.75, 0.5]
for score in scores:
result = evaluate_response_quality(score)
print(f"置信度 {score}: {result}")预期结果:
置信度 0.95: {'quality': '高质量', 'level': '优秀', 'should_confirm': False}
置信度 0.75: {'quality': '需要确认', 'level': '良好', 'should_confirm': True}
置信度 0.5: {'quality': '需要确认', 'level': '一般', 'should_confirm': True}4.3 for 循环:批量处理
# 代码示例:批量处理消息
messages = [
{"id": 1, "content": "天气查询", "processed": False},
{"id": 2, "content": "预订机票", "processed": False},
{"id": 3, "content": "闲聊", "processed": False}
]
print("=== 处理消息队列 ===")
# 【enumerate技巧】同时获取索引i和元素msg
# enumerate(list)返回(索引,元素)元组,避免手动维护计数器
for i, msg in enumerate(messages):
print(f"\n处理消息 #{msg['id']}: {msg['content']}")
# 【原地修改】直接修改字典,无需重新赋值
# Python中字典是可变对象,循环中的修改会影响原列表
msg['processed'] = True
msg['timestamp'] = f"2024-10-27 10:0{i}:00"
print(f" ✅ 已处理,时间戳: {msg['timestamp']}")
# 【列表推导式】一行代码完成过滤+提取
# 等价于: for循环 + if判断 + append操作
# 比传统循环更简洁,性能也略优
processed_ids = [msg['id'] for msg in messages if msg['processed']]
print(f"\n已处理的消息ID: {processed_ids}")
# 【字典推导式】快速构建映射关系
# 格式: {key表达式: value表达式 for item in iterable}
id_to_content = {msg['id']: msg['content'] for msg in messages}
print(f"ID映射: {id_to_content}")
# 【range循环】生成数字序列
# range(1, 4)生成[1,2,3],注意右边界不包含
print("\n=== 重试机制 ===")
max_retries = 3
for attempt in range(1, max_retries + 1):
print(f"尝试 #{attempt}/{max_retries}")
success = (attempt == 2) # 模拟第2次成功
if success:
print(" ✅ 成功!")
break # 成功后立即退出循环,避免浪费重试次数
print(" ❌ 失败,重试中...")预期结果:
=== 处理消息队列 ===
处理消息 #1: 天气查询
✅ 已处理,时间戳: 2024-10-27 10:00:00
处理消息 #2: 预订机票
✅ 已处理,时间戳: 2024-10-27 10:01:00
处理消息 #3: 闲聊
✅ 已处理,时间戳: 2024-10-27 10:02:00
已处理的消息ID: [1, 2, 3]
ID映射: {1: '天气查询', 2: '预订机票', 3: '闲聊'}
=== 重试机制 ===
尝试 #1/3
❌ 失败,重试中...
尝试 #2/3
✅ 成功!4.4 while 循环:条件驱动
# 代码示例:API 重试逻辑
import time
def call_api_with_retry(max_retries=3, delay=1):
"""模拟带重试的 API 调用"""
attempt = 0
# 【while循环】条件驱动,适合未知循环次数的场景
# 这里用于重试:只要未达上限且未成功,就继续尝试
while attempt < max_retries:
attempt += 1 # 先递增计数器
print(f"API 调用尝试 #{attempt}...")
# 【模拟调用】实际开发中这里会是真实的API请求
success = (attempt == 2)
if success:
# 【成功返回】立即退出函数,无需继续循环
print(" ✅ API 调用成功!")
return {"status": "success", "data": "天气数据"}
else:
print(f" ❌ 失败,{delay}秒后重试...")
# 【指数退避】实际应用中delay应递增(如1,2,4,8秒)
# 这里简化为固定延迟,避免快速耗尽API配额
if attempt < max_retries:
time.sleep(delay)
# 【兜底处理】循环正常结束说明所有重试都失败
# 返回错误而非抛出异常,让调用方决定如何处理
print("\n⚠️ 达到最大重试次数,调用失败")
return {"status": "failed", "error": "Max retries exceeded"}
# 执行
result = call_api_with_retry(max_retries=3, delay=0.5)
print(f"\n最终结果: {result}")
# 【break和continue】控制循环流程
print("\n=== 处理消息(跳过无效消息)===")
messages = ["valid_msg_1", "", "valid_msg_2", None, "valid_msg_3"]
processed_count = 0
for msg in messages:
# 【continue】跳过本次循环,直接进入下一次迭代
# if not msg 等价于 if msg == "" or msg == None
if not msg: # 跳过空消息
print(f"⏭️ 跳过无效消息")
continue # 不执行后续代码,直接下一个msg
print(f"✅ 处理: {msg}")
processed_count += 1
# 【break】彻底退出循环,不再处理后续元素
# 常用于找到目标值、达到处理上限等场景
if processed_count >= 2: # 只处理2条
print("⏸️ 达到处理上限,停止")
break # 退出for循环
print(f"\n总共处理: {processed_count} 条消息")预期结果:
API 调用尝试 #1...
❌ 失败,0.5秒后重试...
API 调用尝试 #2...
✅ API 调用成功!
最终结果: {'status': 'success', 'data': '天气数据'}
=== 处理消息(跳过无效消息)===
✅ 处理: valid_msg_1
⏭️ 跳过无效消息
✅ 处理: valid_msg_2
⏸️ 达到处理上限,停止
总共处理: 2 条消息5. 函数:代码复用的艺术
🚀 30秒快速上手:最简函数
如果你是 Python 小白,先看这个最简单的函数例子:
# 定义函数 - 就像给一段代码起个名字,方便重复使用
def greet(name):
print("你好,", name)
return "欢迎"
# 调用函数
message = greet("小明") # 输出:你好, 小明
print(message) # 输出:欢迎
# 带计算的函数
def add(a, b):
result = a + b
return result
total = add(5, 3) # 调用函数
print(total) # 输出:8💡 新手提示:
def用来定义函数,后面跟函数名和括号- 括号里是参数(输入),
return是返回值(输出) - 定义后要调用才会执行
5.1 函数基础:LangGraph 节点函数
在 LangGraph 中,每个节点都是一个函数,接收状态并返回更新后的状态。
# 代码示例:定义 Agent 节点函数
def process_user_message(state):
"""处理用户消息的节点函数
【LangGraph核心概念】
节点函数是图中的计算单元,接收状态、处理逻辑、返回更新后的状态
这种函数签名(state → state)是LangGraph的标准模式
Args:
state: Agent 状态字典
Returns:
更新后的状态
"""
# 【状态读取】从消息列表获取最新用户输入
# [-1]索引获取列表最后一个元素,这是Python的惯用法
user_message = state["messages"][-1]["content"]
# 【意图识别】简单的关键词匹配
# 实际生产中应使用LLM或分类模型
intent = "unknown"
if "天气" in user_message:
intent = "weather"
elif "预订" in user_message:
intent = "booking"
# 【状态更新】直接修改state字典
# LangGraph会自动合并这些更新到全局状态
state["intent"] = intent
state["processed"] = True
print(f"📥 处理消息: {user_message}")
print(f"🎯 识别意图: {intent}")
# 【必须返回state】这是LangGraph节点函数的契约
return state
# 测试节点函数
initial_state = {
"messages": [
{"role": "user", "content": "今天天气怎么样?"}
],
"intent": None,
"processed": False
}
updated_state = process_user_message(initial_state)
print(f"\n更新后的状态: {updated_state}")预期结果:
📥 处理消息: 今天天气怎么样?
🎯 识别意图: weather
更新后的状态: {'messages': [{'role': 'user', 'content': '今天天气怎么样?'}], 'intent': 'weather', 'processed': True}5.2 参数与返回值
# 代码示例:灵活的函数参数
def create_prompt(
system_role="helpful assistant", # 【默认参数】未传入时使用此值
user_message=None, # 【可选参数】允许为None,后续校验
temperature=0.7, # 【模型参数】OpenAI推荐0-2范围
max_tokens=1000,
**kwargs # 【可变关键字参数】捕获所有额外参数
):
"""构建 LLM 提示词
【参数设计原则】
- 必需参数放前面(user_message)
- 常用参数有默认值(system_role, temperature)
- **kwargs处理未知参数,增强扩展性
Args:
system_role: 系统角色
user_message: 用户消息
temperature: 温度参数
max_tokens: 最大 token 数
**kwargs: 其他参数(如top_p, model等)
Returns:
dict: 提示词配置
"""
# 【参数校验】早失败(fail-fast)原则
# 在函数开始就检查必需参数,避免后续计算浪费
if user_message is None:
raise ValueError("必须提供 user_message")
# 【配置构建】标准化API调用格式
config = {
"system": system_role,
"user": user_message,
"parameters": {
"temperature": temperature,
"max_tokens": max_tokens
}
}
# 【额外参数处理】灵活支持新功能而无需修改函数签名
# 例如:未来OpenAI新增参数,调用方可直接传入
if kwargs:
config["extra"] = kwargs
return config
# 使用函数
prompt1 = create_prompt(user_message="你好")
print("基础调用:")
print(prompt1)
prompt2 = create_prompt(
system_role="专业的天气助手",
user_message="北京天气?",
temperature=0.5,
top_p=0.9, # 额外参数会被**kwargs捕获
model="gpt-5"
)
print("\n完整调用:")
print(prompt2)
# 【返回多个值】Python特性:用元组打包多个返回值
def analyze_sentiment(text):
"""分析文本情感"""
# 【简化分析】实际应使用Transformers情感分析模型
score = 0.8 if "好" in text else 0.3
label = "积极" if score > 0.5 else "消极"
confidence = score
# Python会自动将多个值打包成元组
return score, label, confidence # 返回元组
# 【元组解包】一次性接收多个返回值
# 这比返回字典再提取更简洁
score, label, conf = analyze_sentiment("今天天气真好!")
print(f"\n情感分析: 得分={score}, 标签={label}, 置信度={conf}")预期结果:
基础调用:
{'system': 'helpful assistant', 'user': '你好', 'parameters': {'temperature': 0.7, 'max_tokens': 1000}}
完整调用:
{'system': '专业的天气助手', 'user': '北京天气?', 'parameters': {'temperature': 0.5, 'max_tokens': 1000}, 'extra': {'top_p': 0.9, 'model': 'gpt-4'}}
情感分析: 得分=0.8, 标签=积极, 置信度=0.85.3 Lambda 函数:条件边的利器
# 代码示例:Lambda 在 LangGraph 中的应用
# 【传统函数】适合复杂逻辑
def should_continue(state):
return state.get("continue", True)
# 【Lambda等价写法】适合简单单行逻辑
# 格式: lambda 参数: 表达式
# 优点:简洁,常用于临时函数 缺点:调试困难,不支持多行
should_continue_lambda = lambda state: state.get("continue", True)
# 【Lambda用于排序】最常见的应用场景
messages = [
{"id": 3, "content": "third", "priority": 2},
{"id": 1, "content": "first", "priority": 1},
{"id": 2, "content": "second", "priority": 3}
]
# key参数接受函数,定义排序依据
# lambda x: x["priority"] 表示"按priority字段排序"
sorted_by_priority = sorted(messages, key=lambda x: x["priority"])
print("按优先级排序:")
for msg in sorted_by_priority:
print(f" ID={msg['id']}, Priority={msg['priority']}")
# 【Lambda用于过滤】filter()配合lambda筛选元素
# filter(条件函数, 可迭代对象) → 满足条件的元素
high_priority = list(filter(lambda x: x["priority"] >= 2, messages))
print(f"\n高优先级消息: {[m['id'] for m in high_priority]}")
# 【Lambda用于映射】map()配合lambda转换元素
# map(转换函数, 可迭代对象) → 转换后的新元素
# 注意:推导式[x["id"] for x in messages]更Pythonic
ids = list(map(lambda x: x["id"], messages))
print(f"所有ID: {ids}")
# 【LangGraph条件边示例】传统函数版本
def route_based_on_intent(state):
"""根据意图路由到不同节点"""
# 字典映射:意图 → 节点名
intent_routes = {
"weather": "weather_node",
"booking": "booking_node",
"chat": "llm_node"
}
intent = state.get("intent", "chat")
# 如果意图不在映射中,默认返回llm_node
return intent_routes.get(intent, "llm_node")
# 【Lambda版本】一行实现同样逻辑
# 适合add_conditional_edges()的condition参数
route_lambda = lambda s: {
"weather": "weather_node",
"booking": "booking_node"
}.get(s.get("intent", "chat"), "llm_node")
# 测试
test_state = {"intent": "weather"}
print(f"\n路由结果: {route_based_on_intent(test_state)}")
print(f"Lambda 路由: {route_lambda(test_state)}")预期结果:
按优先级排序:
ID=1, Priority=1
ID=3, Priority=2
ID=2, Priority=3
高优先级消息: [3, 2]
所有ID: [3, 1, 2]
路由结果: weather_node
Lambda 路由: weather_node6. 类型提示:为 LangGraph 做准备
🚀 30秒快速上手:最简类型提示
如果你是 Python 小白,先看这个最简单的类型提示:
# 类型提示 - 告诉别人(和自己)这个变量是什么类型
name: str = "小明" # str 表示字符串(文字)
age: int = 18 # int 表示整数
score: float = 95.5 # float 表示小数
# 函数也可以加类型提示
def add(a: int, b: int) -> int: # -> int 表示返回值是整数
return a + b
result = add(5, 3)
print(result) # 输出:8💡 新手提示:
- 类型提示不是必须的,但能让代码更清晰
:后面写类型,->后面写返回值类型- 常见类型:
str(文字)、int(整数)、float(小数)、bool(真假)
类型提示(Type Hints)是 Python 3.5+ 引入的特性,在 LangGraph 中大量使用。
6.1 基础类型提示
# 代码示例:类型提示基础
from typing import List, Dict, Optional, Union
# 【变量类型提示】增强代码可读性和IDE智能提示
# 格式: 变量名: 类型 = 值
user_name: str = "Alice"
age: int = 25
score: float = 0.85
is_active: bool = True
# 【函数类型提示】声明输入输出类型,避免类型错误
# 格式: def 函数名(参数: 类型) -> 返回类型:
def format_message(role: str, content: str) -> dict:
"""格式化消息
Args:
role: 角色(user/assistant)
content: 消息内容
Returns:
格式化的消息字典
"""
return {"role": role, "content": content}
# 【复合类型】List[元素类型]表示列表中元素的类型
# IDE可以自动提示messages[0]有哪些字典键
messages: List[dict] = [
{"role": "user", "content": "hello"},
{"role": "assistant", "content": "hi"}
]
# 【Dict类型】Dict[键类型, 值类型]
# any表示任意类型(不推荐,应尽量具体化)
state: Dict[str, any] = {
"user_id": "123",
"count": 5
}
# 【Optional类型】等价于Union[指定类型, None]
# 明确告诉调用方:此函数可能返回None,需要处理空值情况
def get_user_name(user_id: str) -> Optional[str]:
"""获取用户名,可能返回 None"""
users = {"123": "Alice", "456": "Bob"}
return users.get(user_id) # dict.get()可能返回None
result = get_user_name("789")
print(f"查询结果: {result}") # None
# 【Union类型】表示"多选一",函数可以接受多种类型
# 常用于处理API参数(可能是字符串或数字)
def process_input(value: Union[str, int, float]) -> str:
"""处理不同类型的输入"""
return f"处理: {value} (类型: {type(value).__name__})"
print(process_input("hello"))
print(process_input(123))
print(process_input(3.14))预期结果:
查询结果: None
处理: hello (类型: str)
处理: 123 (类型: int)
处理: 3.14 (类型: float)6.2 LangGraph 风格的类型提示
# 代码示例:模拟 LangGraph 的类型定义
from typing import TypedDict, List, Literal
# 【TypedDict】定义严格的字典结构(类似TypeScript的Interface)
# 比普通dict更安全:IDE可检查字段拼写错误,mypy可做静态类型检查
class Message(TypedDict):
# Literal限定role只能是这三个值之一,避免拼写错误
role: Literal["user", "assistant", "system"]
content: str
# 【嵌套TypedDict】LangGraph状态的标准定义方式
# 定义清晰的状态结构,便于团队协作和维护
class AgentState(TypedDict):
messages: List[Message] # 消息列表,每个元素都是Message类型
intent: str
next_node: str
# 【类型化函数】明确返回AgentState,IDE可以智能提示字段
def create_initial_state(user_message: str) -> AgentState:
"""创建初始状态
【设计模式】工厂函数,统一状态初始化逻辑
避免在各处重复构建状态结构
"""
return {
"messages": [
{"role": "user", "content": user_message}
],
"intent": "unknown",
"next_node": "classifier"
}
# 【节点函数类型签名】LangGraph标准模式
# 输入AgentState → 处理 → 输出AgentState
def classify_intent(state: AgentState) -> AgentState:
"""意图分类节点"""
user_message = state["messages"][-1]["content"]
# 【状态更新】TypedDict仍然是可变字典,可以直接修改
if "天气" in user_message:
state["intent"] = "weather"
state["next_node"] = "weather_tool"
else:
state["intent"] = "general"
state["next_node"] = "llm"
# 【类型安全】返回类型必须匹配AgentState,否则mypy报错
return state
# 测试
initial = create_initial_state("今天天气怎么样?")
print("初始状态:")
print(initial)
updated = classify_intent(initial)
print("\n分类后:")
print(f"意图: {updated['intent']}")
print(f"下一节点: {updated['next_node']}")预期结果:
初始状态:
{'messages': [{'role': 'user', 'content': '今天天气怎么样?'}], 'intent': 'unknown', 'next_node': 'classifier'}
分类后:
意图: weather
下一节点: weather_tool7. 实战案例:构建简单的意图分类器
让我们综合运用本章所有知识,构建一个完整的意图分类系统。
# 代码示例:完整的意图分类器
from typing import Dict, List, Tuple
class IntentClassifier:
"""简单的意图分类器
【架构设计】
- 基于规则的分类器(适合明确意图场景)
- 实际生产应使用BERT分类模型或Few-Shot LLM
- 演示OOP、类型提示、数据处理的综合应用
"""
def __init__(self):
"""初始化分类器
【构造函数】创建实例时自动调用,初始化实例属性
self.xxx 是实例属性,每个对象独立拥有
"""
# 【意图模式库】字典存储意图与关键词的映射
# 实际应用可从配置文件加载,支持动态更新
self.intent_patterns = {
"weather": ["天气", "温度", "下雨", "晴天"],
"booking": ["预订", "订票", "预约", "安排"],
"calculator": ["计算", "加", "减", "乘", "除", "="],
"greeting": ["你好", "hi", "hello", "早上好"]
}
# 【历史记录】List[Dict]存储所有分类结果,用于统计分析
# 生产中应限制大小或持久化到数据库
self.history: List[Dict] = []
def classify(self, message: str) -> Dict[str, any]:
"""分类消息
【核心算法】关键词匹配 + 得分计算
时间复杂度:O(意图数 × 关键词数),适合小规模场景
Args:
message: 用户消息
Returns:
分类结果字典,包含意图、置信度、详细得分
"""
# 【预处理】统一转小写,提高匹配率
message_lower = message.lower()
scores = {}
# 【得分计算】遍历所有意图,统计关键词命中次数
for intent, keywords in self.intent_patterns.items():
# 列表推导式 + sum:简洁统计匹配数
# sum(1 for ...) 等价于 len([kw for kw in keywords if ...])
score = sum(1 for kw in keywords if kw in message_lower)
if score > 0:
scores[intent] = score
# 【意图判定】找到得分最高的意图
if scores:
# max()的key参数:按元组第二项(分数)排序
# best_intent是(intent, score)元组
best_intent = max(scores.items(), key=lambda x: x[1])
intent, confidence = best_intent[0], best_intent[1] / 5
else:
# 【未匹配处理】返回unknown,置信度0
intent, confidence = "unknown", 0.0
# 【结果封装】标准化输出格式
result = {
"message": message,
"intent": intent,
"confidence": min(confidence, 1.0), # 限制最大值为1.0
"all_scores": scores # 保留所有得分,便于调试
}
# 【历史记录】追加到实例属性,支持后续分析
self.history.append(result)
return result
def batch_classify(self, messages: List[str]) -> List[Dict]:
"""批量分类
【批处理优化】列表推导式简洁实现
实际应用可并行处理提升性能
"""
return [self.classify(msg) for msg in messages]
def get_stats(self) -> Dict[str, int]:
"""获取统计信息
【数据分析】统计各意图出现次数
用于分析用户行为模式、优化意图库
"""
intent_counts = {}
for record in self.history:
intent = record["intent"]
# dict.get()配合默认值:简洁实现计数累加
intent_counts[intent] = intent_counts.get(intent, 0) + 1
return intent_counts
def clear_history(self):
"""清空历史
【内存管理】防止history无限增长导致内存溢出
"""
self.history = []
# ===== 使用示例 =====
# 1. 创建分类器
classifier = IntentClassifier()
# 2. 单条消息分类
test_messages = [
"今天北京天气怎么样?",
"帮我预订明天的机票",
"计算 123 + 456",
"你好,请问有什么可以帮助的吗?",
"这是一个随机消息"
]
print("=== 单条分类 ===")
for msg in test_messages:
result = classifier.classify(msg)
print(f"\n消息: {msg}")
print(f" 意图: {result['intent']}")
print(f" 置信度: {result['confidence']:.0%}")
if result['all_scores']:
print(f" 得分详情: {result['all_scores']}")
# 3. 批量分类
batch_messages = [
"明天会下雨吗?",
"订一张去上海的票",
"早上好!"
]
print("\n\n=== 批量分类 ===")
batch_results = classifier.batch_classify(batch_messages)
for result in batch_results:
print(f"{result['message']} → {result['intent']}")
# 4. 统计分析
print("\n\n=== 统计信息 ===")
stats = classifier.get_stats()
for intent, count in sorted(stats.items(), key=lambda x: x[1], reverse=True):
print(f" {intent}: {count} 条")
print(f"\n总处理消息数: {len(classifier.history)}")预期结果:
=== 单条分类 ===
消息: 今天北京天气怎么样?
意图: weather
置信度: 20%
得分详情: {'weather': 1}
消息: 帮我预订明天的机票
意图: booking
置信度: 20%
得分详情: {'booking': 1}
消息: 计算 123 + 456
意图: calculator
置信度: 20%
得分详情: {'calculator': 1}
消息: 你好,请问有什么可以帮助的吗?
意图: greeting
置信度: 20%
得分详情: {'greeting': 1}
消息: 这是一个随机消息
意图: unknown
置信度: 0%
=== 批量分类 ===
明天会下雨吗? → weather
订一张去上海的票 → booking
早上好! → greeting
=== 统计信息 ===
weather: 2 条
booking: 2 条
greeting: 2 条
calculator: 1 条
unknown: 1 条
总处理消息数: 88. 本章总结
✅ 你已经掌握
变量与数据类型
- 基础类型:str, int, float, bool, None
- 字符串操作:strip(), lower(), split(), f-string
数据结构
- List:对话历史、消息队列
- Dict:Agent 状态、配置
- Tuple:固定格式、多返回值
- Set:去重、集合运算
控制流
- if-elif-else:意图路由
- for 循环:批量处理
- while 循环:重试机制
- 列表/字典推导式
函数
- 函数定义与调用
- 参数(默认、可选、**kwargs)
- Lambda 表达式
- 多返回值
类型提示
- 基础类型:str, int, List, Dict
- 高级类型:Optional, Union, TypedDict
- 函数注解
📚 下一步
准备好了吗?接下来我们将学习:
- 0.2-面向对象与工程实践.md - 构建生产级 Agent
- 0.3-AI开发工具链.md - 掌握 AI 生态系统
💡 练习建议
- 修改意图分类器,添加更多意图类别
- 实现一个简单的对话管理器
- 尝试使用类型提示重构你的代码
🎯 记住:Python 的核心不是语法,而是用它解决实际问题。在 LangGraph 开发中,你会不断使用这些基础知识来构建强大的 AI Agent!