Python 3.11 x LangChain 预备知识教程

零基础回顾，面向 LangChain 开发者的 Python 速通手册

---

Demo 01 · pyenv — Python 版本管理

什么是 pyenv？

pyenv 是一个用来管理多版本 Python 的工具。你可能需要同时用 Python 3.10 做旧项目、用 Python 3.12 做新项目，pyenv 让你可以在它们之间一键切换，互不影响。

安装 pyenv（macOS 为例）

# 1. 用 Homebrew 安装 pyenv
brew install pyenv

# 2. 把下面这三行加到 ~/.zshrc（如果你用 zsh shell）
echo 'export PYENV_ROOT="$HOME/.pyenv"' >> ~/.zshrc
echo 'command -v pyenv >/dev/null || export PATH="$PYENV_ROOT/bin:$PATH"' >> ~/.zshrc
echo 'eval "$(pyenv init -)"' >> ~/.zshrc

# 3. 让配置生效
source ~/.zshrc

行号	内容	逐词解释	作用
1	brew install pyenv	brew=macOS包管理器, install=安装, pyenv=工具名	用 Homebrew 把 pyenv 安装到系统
2	export PYENV_ROOT="$HOME/.pyenv" | export=导出环境变量, PYENV_ROOT=pyenv根目录变量名, $HOME=用户主目录, .pyenv=文件夹名	告诉系统 pyenv 装在哪里
3	command -v pyenv >/dev/null || export PATH=...	command -v=检查命令是否存在, >/dev/null=丢弃输出, ||=逻辑或, PATH=系统路径变量	如果 pyenv 不在 PATH 里就把它加进去
4	eval "$(pyenv init -)"	eval=执行字符串内容, $(...)=运行命令并取输出	初始化 pyenv（自动补全 + 版本切换）
5	source ~/.zshrc	source=重新加载文件, ~/.zshrc=zsh配置文件	让刚才的修改立刻生效

常用 pyenv 命令

# 安装指定版本的 Python（编译要几分钟）
pyenv install 3.11.7

# 列出已安装的所有 Python 版本
pyenv versions

# 设置全局默认版本（整个系统都用这个）
pyenv global 3.11.7

# 在当前文件夹设定局部版本（只影响这个项目）
pyenv local 3.11.7

# 卸载某个版本
pyenv uninstall 3.10.5

命令	逐词解释	作用
pyenv install 3.11.7	install=安装, 3.11.7=具体版本号	下载并编译安装指定版本的 Python
pyenv versions	versions=列出所有版本	查看当前安装了多少个 Python 版本
pyenv global 3.11.7	global=全局设置	把 3.11.7 设为系统默认 Python
pyenv local 3.11.7	local=局部设置	在当前目录创建一个 .python-version 文件，cd进来自动切换
pyenv uninstall 3.10.5	uninstall=卸载	删除指定版本的 Python

---

Demo 02 · uv — 超快 Python 包管理器

什么是 uv？

uv 是 Astral 公司出品的新一代 Python 包管理器，速度比 pip 快 10~100 倍。它用 Rust 写，支持虚拟环境创建、依赖管理、脚本运行，一条命令全搞定。LangChain 开发者几乎人人用 uv。

安装 uv

# macOS / Linux 一键安装
curl -LsSf https://astral.sh/uv/install.sh | sh

# 或者用 pip 安装（pip install uv）
pip install uv

用 uv 管理项目

# 创建一个新项目（自动生成 pyproject.toml）
uv init my_langchain_project

# 进入项目目录
cd my_langchain_project

# 添加 LangChain 相关依赖
uv add langchain langchain-openai python-dotenv

# 安装依赖（自动创建 .venv 虚拟环境）
uv sync

# 运行 Python 脚本
uv run python main.py

# 更新所有依赖到最新兼容版本
uv update

行号	内容	逐词解释	作用
1	curl -LsSf https://astral.sh/uv/install.sh | sh	curl=下载工具, -LsSf=静默+跟随重定向+失败时报错, | sh=管道给shell执行	下载并执行 uv 安装脚本
2	uv init my_langchain_project	uv=工具名, init=初始化, my_langchain_project=项目文件夹名	创建新项目，生成 pyproject.toml
3	cd my_langchain_project	cd=切换目录, my_langchain_project=目录名	进入项目目录
4	uv add langchain langchain-openai python-dotenv	uv=工具名, add=添加依赖, langchain=LLM应用框架, langchain-openai=LangChain的OpenAI集成, python-dotenv=读取.env环境变量	把这三个包及其依赖加到项目，修改 pyproject.toml
5	uv sync	sync=同步	读取 pyproject.toml，创建/更新虚拟环境，安装所有依赖
6	uv run python main.py	uv run=在虚拟环境中运行, python=解释器, main.py=入口脚本	在隔离的虚拟环境里执行 main.py
7	uv update	update=更新	把所有依赖升级到符合 pyproject.toml 约束的最新版本

---

Demo 03 · 常见数据类型

本节要学什么？

Python 有丰富的数据类型，分为：数值（int/float）、序列（str/list/tuple）、映射（dict）、集合（set）、布尔（bool）、空值（None）。

完整演示

# ========== 数值类型 ==========

age = 37                          # int — 整数（没有大小限制）
temperature = 36.6               # float — 浮点数（小数）
name = "布鲁斯"                    # str — 字符串（可用单引号或双引号）
print(f"我叫{name}，今年{age}岁")  # f-string — 格式化字符串（Python 3.6+，推荐）

# ========== 列表（List）— 可变序列 ==========

fruits = ["apple", "banana", "cherry"]
print(fruits[0])                 # 用索引取值，从0开始
fruits.append("mango")           # 在末尾添加元素
squares = [x**2 for x in range(5)]  # 列表推导式，快速生成列表（LangChain大量使用）
print(squares)                   # 输出: [0, 1, 4, 9, 16]

# ========== 元组（Tuple）— 不可变序列 ==========

coordinates = (120.5, 30.6)       # 定义坐标（不可变）
print(coordinates[0])             # 输出: 120.5

# ========== 字典（Dict）— 键值对 ==========

person = {
    "name": "布鲁斯",
    "age": 37,
    "city": "深圳"
}
print(person["name"])                              # 通过键读取值
person["email"] = "test@example.com"               # 添加新键值对
print(person.get("phone", "未填写"))               # 安全取值，键不存在返回默认值

# 字典推导式
lengths = {fruit: len(fruit) for fruit in fruits}
print(lengths)

# ========== 集合（Set）— 无序不重复 ==========

tags = {"Python", "LangChain", "AI", "Python"}
print(tags)   # 输出: {'Python', 'LangChain', 'AI'}（自动去重）

# ========== 布尔（Bool） ==========

is_ai_dev = True
is_student = False
print(is_ai_dev and is_student)    # 输出: False（and=与）
print(is_ai_dev or is_student)     # 输出: True（或）
print(not is_student)              # 输出: True（非）

# ========== 空值 None ==========

result = None
print(result is None)              # 输出: True（判断None用 is 而非 ==）

逐行解析

行号	内容	逐词解释	作用
1	age = 37	age=变量名, =赋值符, 37=整数字面量	定义一个名为 age 的整数变量
2	temperature = 36.6	temperature=变量名, =赋值符, 36.6=浮点数字面量	定义浮点数变量
3	name = "布鲁斯"	name=变量名, =赋值符, "..."=字符串字面量	定义字符串变量
4	print(f"我叫{name}，今年{age}岁")	f"..."=f-string, {name}=插入变量name	字符串插值输出（Python 3.6+推荐）
8	fruits = ["apple", "banana", "cherry"]	fruits=变量名, [...]=列表字面量	定义列表（有序、可变）
9	fruits[0]	fruits=列表, [0]=索引访问第0个元素	用索引访问列表元素（从0开始）
10	fruits.append("mango")	.append()=列表末尾追加方法, "mango"=元素	在列表末尾添加一个新元素
11	squares = [x**2 for x in range(5)]	x**2=x的平方, for x in=遍历, range(5)=0~4序列	列表推导式：生成0~4各数的平方列表
15	coordinates = (120.5, 30.6)	coordinates=变量名, (...)元组字面量	定义元组（不可变，用于固定组合）
17	person = {"name": "布鲁斯", ...}	{...}=字典字面量, "name"=键, "布鲁斯"=值	定义字典（键值对数据结构）
18	person["name"]	person=字典, ["name"]=键访问	通过键读取字典的值
19	person.get("phone", "未填写")	.get()=字典安全取值方法, "phone"=键, "未填写"=默认值	键不存在时返回默认值而非报错
23	tags = {"Python", "LangChain", "AI", "Python"}	{...}=集合字面量	定义集合（自动去重）
28	is_ai_dev and is_student	is_ai_dev=布尔变量, and=逻辑与运算符	两边都为True才返回True
29	is_ai_dev or is_student	or=逻辑或运算符	任意一边为True就返回True
30	not is_student	not=逻辑非运算符	取反，True变False，False变True
34	result is None	is None=身份比较运算符	判断变量是否为空值（用 is 而非 ==）

---

Demo 04 · 循环判断与或非（条件控制）

本节要学什么？

条件判断（if/elif/else）和循环（for/while）是所有语言的基石。Python 的特别之处：没有花括号，用缩进表示代码块。LangChain 的 Agent 决策逻辑大量用到条件判断。

完整演示

# ========== 条件判断 if / elif / else ==========

temperature = 38

if temperature > 40:
    print("危险！高温预警")
elif temperature > 37:
    print("发烧了，注意休息")
else:
    print("体温正常")

# ========== match 语句（Python 3.10+，类似switch）==========

status = "loading"

match status:
    case "success":
        print("请求成功")
    case "error":
        print("请求失败")
    case "loading":
        print("加载中...")
    case _:
        # _ 是通配符，表示"其他所有情况"
        print("未知状态")

# ========== 逻辑运算符：and / or / not ==========

isLoggedIn = True
isPremium = False

if isLoggedIn and isPremium:
    print("可以访问付费内容")
else:
    print("权限不足")

if isLoggedIn or isPremium:
    print("至少满足一个条件")

if not isPremium:
    print("不是Premium用户")

# ========== 三元表达式（一行if/else）==========

age = 20
category = "成年人" if age >= 18 else "未成年人"
print(category)  # 输出: 成年人

# ========== for 循环 ==========

languages = ["Python", "JavaScript", "Go"]
for lang in languages:
    print(f"我会{lang}")

scores = {"数学": 95, "英语": 88, "语文": 92}
for subject, score in scores.items():
    print(f"{subject}: {score}分")

for i in range(5):   # 0, 1, 2, 3, 4（共5次）
    print(f"第{i}次")

fruits = ["apple", "banana", "cherry"]
for idx, fruit in enumerate(fruits):
    print(f"{idx}: {fruit}")

# ========== while 循环 ==========

count = 0
while count < 3:
    print(f"count = {count}")
    count += 1  # 等价于 count = count + 1

# break 和 continue
for i in range(10):
    if i == 3:
        continue  # 跳过i==3这一次
    if i == 7:
        break     # 满足条件立即退出整个循环
    print(i)

逐行解析

行号	内容	逐词解释	作用
1	if temperature > 40:	if=如果, temperature=变量, >大于, 40=比较值, :=条件结束	如果温度大于40度，执行下一行
2	print("危险！高温预警")	print=打印函数, 4空格缩进=属于if代码块	条件满足时执行（严格4空格缩进是Python语法）
4	elif temperature > 37:	elif=否则如果	第一个if不满足时，尝试第二个条件
7	else:	else=否则	所有条件都不满足时执行
13	match status:	match=模式匹配关键字, status=要匹配的变量	开始一个 match 语句（Python 3.10+）
14	case "success":	case=分支标签, "success"=匹配字符串常量	如果 status 等于 "success"
20	case _:	case=分支, _=通配符	相当于 switch 的 default
23	isLoggedIn and isPremium	and=逻辑与	两边必须同时为True才为True
27	isLoggedIn or isPremium	or=逻辑或	任意一边为True即为True
30	not isPremium	not=逻辑非	把 False 变 True，True 变 False
33	category = "成年人" if age >= 18 else "未成年人"	X if 条件 else Y=三元表达式	条件为True取左边，False取右边
37	for lang in languages:	for=遍历循环, lang=循环变量, in=在...里	依次把列表中每个元素赋值给 lang
40	for subject, score in scores.items():	.items()=字典键值对视图, (subject, score)=元组解包	遍历字典的每一个键值对
43	for i in range(5):	range(5)=生成0~4的整数序列	循环5次，i 分别是 0,1,2,3,4
47	for idx, fruit in enumerate(fruits):	enumerate=枚举函数，同时返回索引和值	既要索引又要值时用 enumerate
51	while count < 3:	while=当...时循环, <3=循环条件	条件为True就一直执行，False时退出
54	count += 1	+=累加赋值，等价于 count = count + 1	把 count 加1（防止无限循环）
57	if i == 3: continue	continue=跳过本次循环	i等于3时跳过本次，不打印
58	if i == 7: break	break=强制退出循环	i等于7时立刻退出整个循环

---

Demo 05 · 函数与类型注解

本节要学什么？

LangChain 大量使用类型注解（Type Hints），让代码可读性更强，也方便 Pylance 等工具做静态检查。Python 3.10+ 还引入了联合类型（Union）和可选类型（Optional）。

完整演示

# ========== 基础函数 ==========

def greet(name: str) -> str:
    """
    打招呼函数
    name: 名字
    return: 问候语字符串
    """
    return f"你好，{name}！"

message = greet("布鲁斯")
print(message)  # 输出: 你好，布鲁斯！

# ========== 默认参数 ==========

def powers(base: float, exponent: int = 2) -> float:
    """
    计算 base 的 exponent 次方
    exponent 默认为2（平方）
    """
    return base ** exponent

print(powers(3))      # 输出: 9.0（只用默认值）
print(powers(3, 3))   # 输出: 27.0（覆盖默认值）

# ========== *args 和 **kwargs（可变参数）==========

def summarize(*args: str) -> str:
    """
    接收任意数量字符串，拼接成一个摘要
    *args 把所有位置参数收集成元组
    """
    return " | ".join(args)

result = summarize("LangChain", "是LLM应用框架", "支持多种模型")
print(result)  # 输出: LangChain | 是LLM应用框架 | 支持多种模型

def print_configs(**kwargs: object) -> None:
    """
    接收任意关键字参数并打印
    **kwargs 把所有关键字参数收集成字典
    """
    for key, value in kwargs.items():
        print(f"{key} = {value}")

print_configs(model="gpt-4o-mini", temperature=0.7, api_key="sk-xxx")

# ========== 类型注解进阶 ==========

from typing import Callable

# Union — 多种可能类型
def process(value: str | int) -> str:
    """value 可以是字符串或整数"""
    return str(value)

# Optional — 可能是None（等价于 Union[T, None]）
def find_user(user_id: int) -> dict[str, str] | None:
    """根据ID查找用户，找不到返回None"""
    users = {1: {"name": "张三", "city": "深圳"}, 2: {"name": "李四", "city": "北京"}}
    return users.get(user_id)  # 找不到返回None，不报错

user = find_user(1)
if user:
    print(user["name"])  # 输出: 张三

# Callable — 描述函数/可调用对象
def apply_twice(func: Callable[[int], int], x: int) -> int:
    """把函数 func 作用于 x 两次"""
    return func(func(x))

print(apply_twice(lambda n: n + 1, 0))  # 输出: 2

# ========== Lambda 匿名函数 ==========

square = lambda x: x ** 2
print(square(5))  # 输出: 25

# ========== 装饰器（LangChain工具常用）==========

import functools

def log_call(func: Callable) -> Callable:
    """打印函数调用信息的装饰器"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print(f"调用函数: {func.__name__}")
        result = func(*args, **kwargs)
        print(f"函数返回: {result}")
        return result
    return wrapper

@log_call
def add(a: int, b: int) -> int:
    return a + b

print(add(3, 5))

逐行解析

行号	内容	逐词解释	作用
1	def greet(name: str) -> str:	def=定义函数, greet=函数名, name: str=参数注解, -> str=返回注解, :=结束	定义一个接收字符串参数、返回字符串的函数
2	"""..."""	多行字符串（docstring）	函数的文档说明，供 help() 和 IDE 使用
5	return f"你好，{name}！"	return=返回语句, f-string=格式化字符串	把结果返回给调用者
11	def powers(base: float, exponent: int = 2) -> float:	float=浮点类型, int=整数类型, exponent=2=默认参数	exponent 有默认值，调用时可省略
17	base ** exponent	**=幂运算符	计算 base 的 exponent 次方
21	def summarize(*args: str) -> str:	*args=可变位置参数, str=元组内元素都是字符串	接收任意数量字符串参数
22	`"	".join(args)`	"
28	def print_configs(**kwargs: object) -> None:	**kwargs=可变关键字参数, object=任意类型的基类	允许不同类型的配置值（字符串、数字、布尔）
35	from typing import Callable	typing=类型标准库模块, Callable=函数类型注解工具	导入函数签名注解工具
37	def process(value: str | int) -> str:	str|int=联合类型写法（Python 3.10+）	参数可以是字符串或整数
41	def find_user(user_id: int) -> dict[str, str] | None:	dict[...]标准泛型, | None=可为空	返回值可能是字典，也可能是 None
44	Callable[[int], int]	Callable[[参数类型], 返回类型]	描述一个接收int返回int的可调用对象
48	lambda n: n + 1	lambda=匿名函数关键字, n=参数, n+1=返回值	定义一个一行的小函数
52	def log_call(func: Callable) -> Callable:	func=函数类型参数, 返回Callable	定义一个装饰器函数
53	@functools.wraps(func)	functools.wraps=装饰器保留原函数元信息	让 wrapper 伪装成原函数（保留名字和文档）
54	def wrapper(*args, **kwargs):	wrapper=嵌套内部函数, *args/**kwargs=透传所有参数	包装原函数，增加额外逻辑
56	@log_call	@装饰器语法糖	应用装饰器，等价于 add = log_call(add)

---

注释放置规范（你这个问题非常关键）

在团队代码里，推荐遵循这个顺序：

1. 方法级说明放在方法定义下面一行，用 docstring（"""..."""）。
2. 代码块说明放在代码块上方一行，而不是塞进块内部。
3. 只有当某一行逻辑非常反直觉时，才在行尾写短注释。

def run_task(task_id: str) -> dict:
    """执行任务并返回结果。"""

    # 先做输入校验，避免后续调用出现难定位的错误
    if not task_id:
        raise ValueError("task_id 不能为空")

    result = {"id": task_id, "status": "ok"}
    return result

结论：你说得对，块注释大多数情况下放在“方法内部的代码块上方”更清晰；方法整体说明用 docstring，不建议用大段 # 注释放在函数体里替代 docstring。

---

Demo 06 · 类与面向对象

本节要学什么？

LangChain 的核心概念（LLM、Chain、Agent、Tool）全都是类。理解类、对象、继承，是看懂 LangChain 源码的基础。

完整演示

# ========== 最简单的类 ==========

class Dog:
    """一个简单的 Dog 类"""
    species = "哺乳动物"  # 类属性（所有实例共享）

    def __init__(self, name: str, age: int):
        # __init__ = 初始化方法（构造函数）
        # self = 当前实例对象本身（类似 this）
        self.name = name  # 实例属性
        self.age = age

    def bark(self) -> str:
        """实例方法"""
        return f"{self.name} 在叫：汪汪！"

    def __str__(self) -> str:
        """定义 print(对象) 时的输出格式"""
        return f"Dog(name={self.name}, age={self.age})"

# 创建对象（实例化）
dog1 = Dog("旺财", 3)
print(dog1.bark())       # 输出: 旺财 在叫：汪汪！
print(dog1.species)      # 输出: 哺乳动物（类属性）
print(str(dog1))         # 输出: Dog(name=旺财, age=3)

# ========== 继承 ==========

class LangChainTool:
    """LangChain 工具的基类"""
    name: str   # 类型注解：name 是字符串
    description: str

    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description

    def run(self, query: str) -> str:
        """子类必须实现这个方法"""
        raise NotImplementedError("子类必须实现 run 方法")

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(name={self.name!r})"

class SearchTool(LangChainTool):
    """搜索工具（继承 LangChainTool）"""

    def __init__(self, name: str, description: str, api_key: str):
        super().__init__(name, description)  # 调用父类构造函数
        self.api_key = api_key               # 自己新增的属性

    def run(self, query: str) -> str:
        """实现父类的抽象方法（多态）"""
        return f"搜索「{query}」的结果（使用 {self.name}）"

search = SearchTool(
    name="WebSearch",
    description="搜索互联网信息",
    api_key="sk-xxx"
)
print(search.run("LangChain 教程"))
print(repr(search))

# ========== dataclass（Python 3.7+，LangChain大量使用）==========

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class LLMConfig:
    """LLM 配置数据类（自动生成 __init__ / __repr__ / __eq__）"""
    model: str                          # 必需字段
    temperature: float = 0.7           # 带默认值
    max_tokens: int = 2048
    created_at: datetime = field(default_factory=datetime.now)  # 工厂函数生成默认值

config = LLMConfig(model="gpt-4", temperature=0.9)
print(config)

config.temperature = 0.5               # dataclass 默认可变，可修改
print(config.temperature)

逐行解析

行号	内容	逐词解释	作用
1	class Dog:	class=定义类关键字, Dog=类名（首字母大写）	定义一个名为 Dog 的类
3	species = "哺乳动物"	species=类属性名	类属性：所有 Dog 实例共享
5	def __init__(self, name: str, age: int):	init=特殊方法（构造器）, self=当前实例, name: str=参数注解	创建实例时自动调用的初始化方法
7	self.name = name	self.name=实例属性, =右边name=参数值	把参数存到当前实例的属性里
11	def bark(self) -> str:	def=定义方法, bark=方法名, self=实例引用	定义实例方法（需要创建实例来调用）
12	return f"{self.name} 在叫：汪汪！"	self.name=访问当前实例的 name 属性	方法内部通过 self 访问实例数据
15	def __str__(self) -> str:	str=Python特殊方法（魔术方法）	定义 print(对象) 时的输出格式
23	dog1 = Dog("旺财", 3)	Dog=类名, (...)=传给 init 的参数	实例化：创建一个 Dog 对象
28	class LangChainTool:	继承语法	定义一个基类
30	name: str	类型注解（非赋值）	声明属性类型，IDE 可做静态检查
34	raise NotImplementedError(...)	raise=抛出异常, NotImplementedError=未实现异常	强制子类必须重写此方法
40	class SearchTool(LangChainTool):	SearchTool=子类名, (LangChainTool)=父类名	定义子类，继承父类所有属性和方法
42	super().__init__(name, description)	super()=获取父类引用, .init()=调用父类构造器	在子类构造函数中初始化从父类继承的属性
45	def run(self, query: str) -> str:	重写（override）父类方法	子类提供具体实现（多态的基础）
55	@dataclass	@dataclass=装饰器	自动为类生成 init / repr / eq 等方法
56	model: str	必需字段（无默认值）	dataclass 会自动生成带这些参数的 init
57	temperature: float = 0.7	带默认值的字段	创建实例时可不传，用默认值
58	field(default_factory=datetime.now)	field=字段选项, default_factory=工厂函数	每次实例化都自动调用函数生成默认值

---

Demo 07 · 异常处理与日志

本节要学什么？

LangChain 应用运行时可能遇到：API Key 错误、网络超时、模型返回格式异常。优雅地处理这些错误，不让程序崩溃，是生产级代码的必备能力。

完整演示

import logging
from typing import Optional

# ========== 配置日志（比 print 强一万倍）==========

logging.basicConfig(
    level=logging.INFO,                        # 记录 INFO 及以上级别
    format="%(asctime)s | %(levelname)-8s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger(__name__)           # __name__=当前模块名

logger.info("LangChain 应用启动")
logger.warning("这是警告级别日志")
logger.error("这是错误级别日志")

# ========== try / except / finally ==========

def call_llm(prompt: str, api_key: Optional[str]) -> str:
    """
    模拟调用 LLM
    演示异常处理的完整结构
    """
    try:
        if not api_key:
            raise ValueError("API Key 不能为空！")

        logger.info(f"发送请求: {prompt[:20]}...")
        result = f"LLM 对「{prompt}」的回复"
        logger.info("请求成功")
        return result

    except ValueError as e:
        # 捕获 ValueError 异常（参数错误）
        logger.error(f"参数错误: {e}")
        raise  # 重新抛出，让调用者知道发生了错误

    except (ConnectionError, TimeoutError) as e:
        # 同时捕获多种异常（网络类错误）
        logger.error(f"网络错误: {e}")
        return "网络异常，请稍后重试"

    except Exception as e:
        # 兜底：捕获所有未预料的异常（避免程序崩溃）
        logger.exception("发生了未知错误！")  # exception 会打印堆栈
        return "系统错误，请联系管理员"

    finally:
        # 不管有没有异常，finally 都会执行（适合清理工作）
        logger.info("请求处理完毕")

# ========== with 上下文管理器（自动关闭文件）==========

def read_config_file(filepath: str) -> str:
    """
    with 语句：自动管理资源（文件、网络连接等）
    离开 with 块时自动 close()，不需要手动处理
    """
    with open(filepath, "r", encoding="utf-8") as f:
        content = f.read()
    # 文件已自动关闭
    return content

# 写入 .env 配置文件（LangChain 常用 .env 存储 API Key）
env_content = """OPENAI_API_KEY=sk-xxx
MODEL_NAME=gpt-4o-mini
TEMPERATURE=0.7
"""
with open(".env", "w", encoding="utf-8") as f:
    f.write(env_content)

# ========== 断言（开发时检查假设）==========

def divide(a: float, b: float) -> float:
    """除法，b 不能为 0"""
    if b == 0:
        raise ValueError("除数不能为 0！")
    return a / b

print(divide(10, 2))   # 输出: 5.0
# print(divide(10, 0))  # 抛出 ValueError: 除数不能为 0！

逐行解析

行号	内容	逐词解释	作用
1	import logging	import=导入模块, logging=Python标准日志模块	导入日志模块
4	logging.basicConfig(...)	basicConfig=日志基础配置函数	设置日志级别、输出格式等
5	level=logging.INFO	level=日志级别, logging.INFO=信息级别	记录 INFO 及以上级别的日志
6	format="%(asctime)s | %(levelname)-8s | %(message)s"	format=格式字符串, asctime=时间, levelname=级别, message=消息	定义每条日志的输出格式
8	logger = logging.getLogger(__name__)	getLogger=获取日志记录器, name=当前模块名	创建以模块名为名的 logger
17	raise ValueError("API Key 不能为空！")	raise=抛出异常, ValueError=值错误异常类型	主动抛出异常，表示参数不合法
20	try:	try=尝试执行块	标记可能发生异常的代码块
21	logger.info(f"发送请求: {prompt[:20]}...")	[:20]=字符串切片，取前20个字符	只打印前20字，避免日志过长
24	except ValueError as e:	except=捕获异常, ValueError=异常类型, as e=把异常绑定到变量e	捕获 ValueError 类型的异常
28	except (ConnectionError, TimeoutError) as e:	(A, B)=异常组，同时捕获多种异常	一次捕获多个不同类型的异常
32	except Exception as e:	Exception=所有常规异常的基类	兜底捕获，避免有异常没被处理
33	logger.exception(...)	exception=打印堆栈信息的 error	既记录错误又打印完整调用栈
36	finally:	finally=无论是否异常都执行	适合做清理工作（关闭文件、释放锁等）
42	with open(filepath, "r", encoding="utf-8") as f:	with=上下文管理器, open=打开文件, "r"=读模式, encoding="utf-8"=字符编码	打开文件并绑定到变量 f，自动管理资源
44	content = f.read()	f=文件对象, .read()=读取全部内容方法	把文件内容全部读入字符串
52	if b == 0: raise ValueError(...)	if=条件判断, raise=抛出异常, ValueError=参数值错误	运行时参数校验（生产环境更稳定，不依赖 assert）

---

Demo 08 · 文件操作、JSON、YAML 与 .env

本节要学什么？

LangChain 的配置文件、Prompt 模板、Chain 配置几乎都会涉及JSON 和 YAML 文件。API Key 等敏感信息存在 .env 文件里，而不是写死在代码中。

完整演示

import json          # JSON 序列化/反序列化（Python 内置）
import yaml          # 需要: uv add pyyaml
from pathlib import Path
from dotenv import load_dotenv  # 需要: uv add python-dotenv

# ========== 读取和写入 JSON ==========

config = {
    "model": "gpt-4o-mini",
    "temperature": 0.7,
    "max_tokens": 2048,
    "tools": ["search", "calculator", "wikipedia"]
}

# 写入 JSON 文件（indent=4 让格式更易读）
with open("config.json", "w", encoding="utf-8") as f:
    json.dump(config, f, ensure_ascii=False, indent=4)

# 读取 JSON 文件
with open("config.json", "r", encoding="utf-8") as f:
    loaded = json.load(f)  # json.load() 从文件读取，json.loads() 从字符串读取

print(loaded["model"])  # 输出: gpt-4
print(json.dumps(loaded, ensure_ascii=False))  # 字符串转 JSON

# ========== 读取和写入 YAML ==========

chain_config = {
    "chain_type": "LLMChain",
    "prompt": {
        "template": "请把以下中文翻译成英文：{text}",
        "input_variables": ["text"]
    },
    "llm": {"model_name": "gpt-4o-mini", "temperature": 0.5}
}

# 写入 YAML 文件
with open("chain.yaml", "w", encoding="utf-8") as f:
    yaml.dump(chain_config, f, allow_unicode=True, default_flow_style=False)

# 读取 YAML 文件
with open("chain.yaml", "r", encoding="utf-8") as f:
    loaded_yaml = yaml.safe_load(f)  # safe_load 只允许基本类型，避免执行任意代码

print(loaded_yaml["chain_type"])   # 输出: LLMChain
print(loaded_yaml["llm"]["model_name"])  # 输出: gpt-4o-mini

# ========== .env 文件与 python-dotenv ==========

# .env 文件内容（不要提交到 Git！）
# OPENAI_API_KEY=sk-xxx
# MODEL_NAME=gpt-4o-mini

load_dotenv()  # 加载 .env 文件到环境变量

import os
api_key = os.getenv("OPENAI_API_KEY")  # 从环境变量读取
model = os.getenv("MODEL_NAME", "gpt-4o-mini")  # 不存在时用默认值

masked_key = f"{api_key[:6]}***" if api_key else None
print(f"API Key: {masked_key}")
print(f"Model: {model}")

# ========== pathlib（更现代的文件路径操作）==========

project_root = Path.cwd()  # 教程示例用当前工作目录（脚本/Notebook 都可运行）
config_path = project_root / "config.json"  # 拼接路径（自动处理 / 或 \）
data_dir = project_root / "data"

# 创建目录（parents=True 不报错，exist_ok=True 目录存在不报错）
data_dir.mkdir(parents=True, exist_ok=True)

print(config_path.exists())  # 检查文件是否存在
print(config_path.suffix)     # 获取文件扩展名（.json）
print(config_path.stem)       # 获取不含扩展名的文件名（config）

逐行解析

行号	内容	逐词解释	作用
1	import json	import=导入, json=Python标准库JSON模块	导入 JSON 处理模块
2	import yaml	yaml=第三方YAML库（需安装 pyyaml）	导入 YAML 处理模块
3	from pathlib import Path	pathlib=路径操作标准库, Path=路径类	导入路径操作工具
4	from dotenv import load_dotenv	dotenv=环境变量加载库（需安装 python-dotenv）	导入 .env 文件加载函数
11	json.dump(config, f, ensure_ascii=False, indent=4)	json.dump=把Python对象写入文件, ensure_ascii=False=保留中文, indent=4=缩进4空格	把字典写入 JSON 文件（格式化）
15	loaded = json.load(f)	json.load=从文件反序列化JSON	把 JSON 文件内容读成 Python 字典
16	json.loads()	json.loads=从字符串反序列化JSON	把 JSON 字符串解析成 Python 对象
24	yaml.dump(chain_config, f, allow_unicode=True, ...)	yaml.dump=把Python对象写入YAML文件, allow_unicode=True=保留中文	把字典写入 YAML 文件
28	yaml.safe_load(f)	yaml.safe_load=安全加载YAML（只允许基本类型）	从文件加载 YAML（防止任意代码执行）
35	load_dotenv()	load_dotenv=加载.env文件到环境变量	把 .env 文件中的变量加载到 os.environ
37	os.getenv("OPENAI_API_KEY")	os.getenv=读取环境变量, "OPENAI_API_KEY"=变量名	从环境变量读取 API Key
38	os.getenv("MODEL_NAME", "gpt-4o-mini")	第二个参数=默认值（变量不存在时使用）	安全读取，带默认值兜底
43	Path.cwd()	cwd=current working directory（当前工作目录）	在脚本和 Notebook 中都稳定可用
44	project_root / "config.json"	/ = Path 对象的路径拼接运算符	拼接路径（自动适配不同操作系统）
45	data_dir = project_root / "data"	同上	拼接 data 目录路径
47	data_dir.mkdir(parents=True, exist_ok=True)	.mkdir=创建目录, parents=True=自动创建父目录, exist_ok=True=存在不报错	安全创建目录（幂等操作）
49	config_path.exists()	.exists()=检查路径是否存在	返回 True 或 False
50	config_path.suffix	.suffix=获取文件扩展名	返回 ".json"

---

Demo 09 · HTTP 请求（requests）与 API 调用

本节要学什么？

LangChain 的本质就是调用大模型 API。无论是 OpenAI、Anthropic 还是本地模型，都离不开 HTTP 请求。requests 库是 Python 最流行的 HTTP 客户端，LangChain 底层也用它。

完整演示

import requests   # 需要: uv add requests
import os
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")

# ========== 最简单的 GET 请求 ==========

response = requests.get("https://httpbin.org/get")
print(response.status_code)  # HTTP 状态码（200=成功，404=未找到，500=服务器错误）
print(response.text)        # 响应原始文本
print(response.json())       # 把响应体解析成 Python 字典（自动 JSON 解码）

# ========== POST 请求（带 body）— LangChain 调用模型的本质 ==========

headers = {
    "Authorization": f"Bearer {api_key}",   # Bearer Token 认证方式
    "Content-Type": "application/json"        # 告诉服务器发送的是 JSON
}

payload = {
    "model": "gpt-4",
    "messages": [
        {"role": "system", "content": "你是一个有帮助的助手"},
        {"role": "user", "content": "用一句话解释什么是 LangChain"}
    ],
    "temperature": 0.7,
    "max_tokens": 500
}

response = requests.post(
    "https://api.openai.com/v1/chat/completions",
    headers=headers,
    json=payload,          # requests 自动把字典转成 JSON 并设置 Content-Type
    timeout=30             # 超时时间（秒），防止请求卡死
)

print(f"状态码: {response.status_code}")   # 200 表示成功
result = response.json()
print(result["choices"][0]["message"]["content"])  # 提取 LLM 的回复

# ========== 查询参数（URL 问号后面的参数）==========

params = {
    "q": "LangChain 教程",
    "page": 1,
    "per_page": 10
}
response = requests.get("https://httpbin.org/get", params=params)
print(response.url)   # 打印完整 URL（含编码后的参数）

# ========== 处理错误状态码 ==========

response = requests.get("https://httpbin.org/status/404", timeout=5)
print(response.status_code)  # 输出: 404

# response.raise_for_status()  # 有错误时抛出异常，没错误时什么都不做

if response.status_code == 200:
    print("请求成功")
elif response.status_code == 404:
    print("资源不存在")
elif response.status_code >= 500:
    print("服务器错误，稍后重试")
else:
    print(f"其他错误，状态码: {response.status_code}")

# ========== requests.Session（保持连接，提升性能）==========

session = requests.Session()
session.headers.update({"Authorization": f"Bearer {api_key}"})  # 全局headers，所有请求复用

# 连续发多个请求时，Session 复用 TCP 连接，速度更快
for i in range(3):
    resp = session.get(f"https://httpbin.org/get?request_id={i}", timeout=5)
    print(f"请求 {i}: {resp.status_code}")

session.close()  # 关闭 Session（也可以用 with: with requests.Session() as session:）

逐行解析

行号	内容	逐词解释	作用
1	import requests	requests=Python最流行的HTTP客户端库	导入发送HTTP请求的工具
5	api_key = os.getenv("OPENAI_API_KEY")	os.getenv=读取环境变量	安全读取 API Key（不硬编码在代码里）
9	requests.get("https://httpbin.org/get")	.get=发送GET请求, URL=请求地址	发送一个简单的 GET 请求
10	response.status_code	status_code=HTTP状态码属性	查看请求是否成功（200=成功）
11	response.text	text=响应体原始文本	获取响应的文本内容
12	response.json()	.json()=JSON响应解析方法	把响应体自动解析成 Python 字典
16	headers = {...}	headers=HTTP请求头字典	告诉服务器：我是谁（认证）、我在发什么类型的数据
17	"Authorization": f"Bearer {api_key}"	Authorization=认证头字段名, Bearer=令牌类型, f-string=插入API密钥	API 认证的标准方式（大多数大模型API都用这个）
18	"Content-Type": "application/json"	Content-Type=内容类型, application/json=JSON格式	告诉服务器 body 是 JSON 格式
21	payload = {...}	payload=请求载荷（body数据）	定义发给 API 的数据
23	"role": "system"	role=角色字段, system=系统角色（定义AI行为）	定义系统提示词
24	"role": "user"	user=用户角色	用户发送的消息
29	requests.post(..., json=payload, timeout=30)	.post=发送POST请求, json=payload=自动序列化+设置Content-Type, timeout=超时秒数	发送 POST 请求（带 JSON body 和超时保护）
37	params = {"q": "LangChain 教程", ...}	params=URL查询参数	问号后面的键值对（?key=value）
38	response = requests.get(..., params=params)	params=查询参数	requests 自动把字典拼成 ?q=...&page=...
39	response.url	.url=最终发送的完整URL	打印实际发出的 URL（已编码）
43	response.raise_for_status()	.raise_for_status()=有错误时抛出异常	把 HTTP 错误码转成 Python 异常（方便 try/except 处理）
58	session = requests.Session()	Session=会话对象	创建持久化会话（复用 TCP 连接）
59	session.headers.update({...})	.headers.update=批量更新请求头	给会话设置全局 headers（所有请求自动带上）
63	session.close()	.close()=关闭会话	关闭 TCP 连接（释放资源）

---

Demo 10 · 上下文管理器、生成器与迭代器

本节要学什么？

上下文管理器（with 语句）我们已经见过。生成器是 Python 里非常优雅的惰性求值方式——在 LangChain 里处理海量文档流时，生成器能让你不用一次性把全部数据加载到内存，这一点在高并发场景下至关重要。

完整演示

# ========== 生成器函数（yield）==========

def count_to_5():
    """用 yield 返回值，每次返回一个后暂停函数"""
    for i in range(1, 6):
        yield i  # yield=生成并暂停，返回一个值后函数暂停在这里
        print(f"已yield {i}")  # 下次调用时从这里继续

# 生成器是惰性的：不会立刻执行函数体，只是创建一个生成器对象
gen = count_to_5()
print(gen)  # 输出: <generator object count_to_5 at 0x...>

# 每次 next() 取一个值（节省内存，适合处理大数据）
print(next(gen))  # 输出: 1（遇到yield暂停）
print(next(gen))  # 输出: 2（从暂停处继续，再遇到yield暂停）
print(next(gen))  # 输出: 3

# ========== 生成器表达式（类似列表推导式，但惰性）==========

# 列表推导式：一次性把所有平方算出来（占内存）
squares_list = [x**2 for x in range(1000000)]

# 生成器表达式：只记录规则，需要时才算（省内存）
squares_gen = (x**2 for x in range(1000000))
print(squares_gen)  # <generator object <genexpr> at 0x...>
print(next(squares_gen))  # 输出: 0
print(next(squares_gen))  # 输出: 1

# ========== 在 LangChain 中用生成器处理文档流 ==========

def stream_documents(documents: list[str]):
    """
    模拟流式读取文档（LangChain 的 RetrievalQA 会用到）
    每次 yield 一段文本，而不是一次性返回所有文本
    """
    for doc in documents:
        # 模拟把文档分成小段，一段一段地 yield
        words = doc.split()
        for word in words:
            yield word + " "

documents = ["LangChain 是一个应用框架", "它可以构建 LLM 应用", "支持多种模型"]
for chunk in stream_documents(documents):
    print(chunk, end="")  # 流式输出，不用等全部处理完

# ========== 上下文管理器（用类实现）==========

class Timer:
    """测量代码执行时间"""

    def __init__(self, name: str = "任务"):
        self.name = name
        self.start = None
        self.end = None

    def __enter__(self):
        """进入 with 块时执行（类似 try 块开头）"""
        import time
        self.start = time.time()
        print(f"[{self.name}] 开始")
        return self  # with 的 as 子句会收到这个返回值

    def __exit__(self, exc_type, exc_val, exc_tb):
        """离开 with 块时执行（类似 finally）"""
        import time
        self.end = time.time()
        elapsed = self.end - self.start
        print(f"[{self.name}] 结束，耗时 {elapsed:.4f} 秒")
        return False  # 返回 False 或 None 表示不拦截异常

# 使用 with 上下文管理器
with Timer("LangChain 文档处理") as timer:
    # 这里写要计时的代码
    total = sum(range(1000000))
    print(f"计算结果: {total}")

# ========== itertools（生成器工具库）==========

import itertools

# count() — 无限计数器（永不停止）
# counter = itertools.count(1)
# print(next(counter))  # 1, 2, 3, 4, ...

# islice — 从无限生成器里取前N个（不卡死）
limited = itertools.islice(itertools.count(1), 5)
print(list(limited))  # 输出: [1, 2, 3, 4, 5]

# chain — 把多个可迭代对象串起来
chain = itertools.chain([1, 2], ["a", "b"], [True, False])
print(list(chain))  # 输出: [1, 2, 'a', 'b', True, False]

# groupby — 按key分组
data = sorted([("cat", 1), ("dog", 2), ("cat", 3), ("dog", 4)], key=lambda x: x[0])
for key, group in itertools.groupby(data, key=lambda x: x[0]):
    print(f"{key}: {list(group)}")
# 输出: cat: [('cat', 1), ('cat', 3)]
#       dog: [('dog', 2), ('dog', 4)]

逐行解析

行号	内容	逐词解释	作用
1	def count_to_5():	def=定义函数, count_to_5=函数名	定义一个生成器函数（注意没有 return，只有 yield）
2	"""用 yield 返回值..."""	docstring=函数文档说明	说明生成器函数的特性
4	yield i	yield=生成并暂停关键字, i=要生成的值	返回一个值并暂停函数执行，下次调用从暂停处继续
11	gen = count_to_5()	gen=生成器对象变量名	调用生成器函数不会执行函数体，只返回一个生成器对象
13	next(gen)	next()=从生成器取下一个值	恢复函数执行到下一个 yield，返回其值
18	squares_gen = (x**2 for x in range(1000000))	(表达式 for x in 可迭代对象)=生成器表达式（圆括号）	创建惰性生成器（不立即计算，不占内存）
26	def stream_documents(documents: list[str]):	list[str]=类型注解（Python 3.9+，等价于 List[str]）	定义流式文档处理函数（返回生成器）
28	words = doc.split()	.split()=按空格分割字符串成列表	把文档分成单词列表
29	yield word + " "	yield=逐个返回每个单词+空格	流式返回（一次一个词）
32	for chunk in stream_documents(documents):	for=遍历生成器	逐个处理流式数据
33	print(chunk, end="")	end=""=不换行，持续输出	流式打印（同一行不断追加）
37	class Timer:	class=定义类关键字, Timer=类名	定义一个上下文管理器类
39	def __enter__(self):	enter=进入with块时自动调用的方法	做准备工作（开始计时）
41	return self	return self=把 self 作为 with 的 as 子句的值	让 with as timer 能拿到 timer 本身
43	def __exit__(self, exc_type, exc_val, exc_tb):	exit=离开with块时自动调用的方法, exc_type=异常类型, exc_val=异常值, exc_tb=堆栈	做清理工作（停止计时），返回 False 不拦截异常
50	with Timer("LangChain 文档处理") as timer:	with...as=上下文管理器语法	进入时调用 enter，离开时调用 exit
60	itertools.islice(itertools.count(1), 5)	itertools.count=无限计数器, islice=无限迭代器切片, 5=只取前5个	从无限生成器安全取出前N个元素
63	itertools.chain([1, 2], ["a", "b"], ...)	chain=链式连接, [...]=多个可迭代对象	把多个序列串成一个序列
66	itertools.groupby(data, key=lambda x: x[0])	groupby=按key分组, key=分组依据函数	把相邻的同类元素分组（数据需先排序）

---

Demo 11 · 生产实践补充（LangChain 项目强烈建议）

本节要学什么？

前面 10 个 Demo 解决“会写”。这一节补“写得稳”：异步并发、重试、敏感信息处理、静态检查。

最小示例

import asyncio
import os
from typing import Any

import httpx


async def fetch_json(client: httpx.AsyncClient, url: str) -> dict[str, Any]:
    """异步请求 + 超时 + 状态码检查。"""
    resp = await client.get(url, timeout=10.0)
    resp.raise_for_status()
    return resp.json()


async def main() -> None:
    # 不要打印完整密钥，最多打印前缀
    api_key = os.getenv("OPENAI_API_KEY")
    masked = f"{api_key[:6]}***" if api_key else None
    print("OPENAI_API_KEY:", masked)

    async with httpx.AsyncClient() as client:
        # 并发执行多个请求（真实项目里可并发查多个检索源）
        urls = ["https://httpbin.org/get", "https://httpbin.org/uuid"]
        tasks = [fetch_json(client, u) for u in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    for idx, item in enumerate(results):
        if isinstance(item, Exception):
            print(f"任务 {idx} 失败: {item}")
        else:
            print(f"任务 {idx} 成功: keys={list(item.keys())[:3]}")


if __name__ == "__main__":
    asyncio.run(main())

实战注意事项

1. 所有外部请求都要显式 timeout，并在错误时 raise_for_status()。
2. 不要把 .env、API Key、用户敏感数据提交到 Git 或写进日志。
3. 优先给函数写返回类型注解，结合 mypy 或 pyright 做静态检查。
4. LLM 输出永远当“不可信输入”处理，先校验结构再使用。
5. 把“可恢复错误”（超时、429、5xx）和“不可恢复错误”（参数非法）分开处理。

---

附录 · LangChain 必知 Python 语法速查卡

语法	含义	在 LangChain 中的用途
f"Hello, {name}"	f-string 格式化	拼 Prompt 模板
*args, **kwargs	可变参数	装饰器、Tool 的 run 方法
typing.Optional[T]	可选类型（等于 Union[T, None]）	LLM 返回值、Tool 执行结果
typing.Callable[[T], R]	函数类型注解	装饰器、Callback 回调
@dataclass	自动生成 __init__ 等	LLMConfig、ToolConfig 等配置类
with open(...) as f:	上下文管理器	读取 prompt 模板文件、.env
yield	生成器关键字	流式输出（Streaming）
super().__init__()	调用父类构造器	自定义 Tool、Custom Chain
isinstance(x, type)	类型检查	判断 LLM 返回的是否为字符串
functools.wraps	保留函数元信息	写装饰器时保持原函数名称不变
pathlib.Path	路径操作	拼接配置文件路径
.env + load_dotenv()	环境变量隔离	管理 API Key（不提交到 Git）

---

下一步：建议直接用 uv init my_langchain_project 创建项目，然后：

uv add langchain langchain-openai
uv run python -c "from langchain_openai import ChatOpenAI; print('LangChain 环境 OK!')"

环境验证通过后，去 LangChain 官方文档 开始你的 LLM 应用开发之旅！