Python 3.11 x LangChain 预备知识教程

Python 3.11 x LangChain 预备知识教程零基础回顾 面向 LangChain 开发者的 Python 速通手册 pyenv 是一个用来管理多版本 Python 的工具 你可能需要同时用 Python 3 10 做旧项目 用 Python 3 12 做新项目 pyenv 让你可以在它们之间一键切换 互不影响 1 用 Homebrew 安装 pyenv brew install pyenv 2 把下面这三行加到

大家好,我是讯享网,很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。



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


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

# 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配置文件 让刚才的修改立刻生效
# 安装指定版本的 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

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

# macOS / Linux 一键安装 curl -LsSf https://astral.sh/uv/install.sh | sh # 或者用 pip 安装(pip install uv) pip install 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 约束的最新版本

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 = [x2 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"] = "" # 添加新键值对 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 = [x2 for x in range(5)] x2=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 而非 ==)

条件判断(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时立刻退出整个循环

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。


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=工厂函数 每次实例化都自动调用函数生成默认值

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)

LangChain 的配置文件、Prompt 模板、Chain 配置几乎都会涉及JSONYAML 文件。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”

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 连接(释放资源)

上下文管理器(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) # 输出: 
             
    
               
                 # 每次 next() 取一个值(节省内存,适合处理大数据) print(next(gen)) # 输出: 1(遇到yield暂停) print(next(gen)) # 输出: 2(从暂停处继续,再遇到yield暂停) print(next(gen)) # 输出: 3 # ========== 生成器表达式(类似列表推导式,但惰性)========== # 列表推导式:一次性把所有平方算出来(占内存) squares_list = [x2 for x in range()] # 生成器表达式:只记录规则,需要时才算(省内存) squares_gen = (x2 for x in range()) print(squares_gen) # 
                
                  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()) 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 = (x2 for x in range()) (表达式 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=分组依据函数 把相邻的同类元素分组(数据需先排序)

前面 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. 优先给函数写返回类型注解,结合 mypypyright 做静态检查。
  4. LLM 输出永远当“不可信输入”处理,先校验结构再使用。
  5. 把“可恢复错误”(超时、429、5xx)和“不可恢复错误”(参数非法)分开处理。

语法 含义 在 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 应用开发之旅!

小讯
上一篇 2026-03-28 10:53
下一篇 2026-03-28 10:51

相关推荐

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容,请联系我们,一经查实,本站将立刻删除。
如需转载请保留出处:https://51itzy.com/kjqy/250132.html