LangChain教程-1、python基础

Python 3.11 x LangChain 预备知识教程

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

Demo 01 · pyenv — Python 版本管理

什么是 pyenv？

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

1. # 1. 用 Homebrew 安装 pyenv
2. brew install pyenv
4. # 2. 把下面这三行加到 ~/.zshrc（如果你用 zsh shell）
5. echo 'export PYENV_ROOT="$HOME/.pyenv"' >> ~/.zshrc
6. echo 'command -v pyenv >/dev/null || export PATH="$PYENV_ROOT/bin:$PATH"' >> ~/.zshrc
7. echo 'eval "$(pyenv init -)"' >> ~/.zshrc
9. # 3. 让配置生效
10. source ~/.zshrc

复制代码

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

1. # 安装指定版本的 Python（编译要几分钟）
2. pyenv install 3.11.7
4. # 列出已安装的所有 Python 版本
5. pyenv versions
7. # 设置全局默认版本（整个系统都用这个）
8. pyenv global 3.11.7
10. # 在当前文件夹设定局部版本（只影响这个项目）
11. pyenv local 3.11.7
13. # 卸载某个版本
14. pyenv uninstall 3.10.5

复制代码

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

什么是 uv？

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

1. # macOS / Linux 一键安装
2. curl -LsSf https://astral.sh/uv/install.sh | sh
4. # 或者用 pip 安装（pip install uv）
5. pip install uv

复制代码

用 uv 管理项目

1. # 创建一个新项目（自动生成 pyproject.toml）
2. uv init my_langchain_project
4. # 进入项目目录
5. cd my_langchain_project
7. # 添加 LangChain 相关依赖
8. uv add langchain langchain-openai python-dotenv
10. # 安装依赖（自动创建 .venv 虚拟环境）
11. uv sync
13. # 运行 Python 脚本
14. uv run python main.py
16. # 更新所有依赖到最新兼容版本
17. uv update

复制代码

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

本节要学什么？

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

1. # ========== 数值类型 ==========
3. age = 37                          # int — 整数（没有大小限制）
4. temperature = 36.6               # float — 浮点数（小数）
5. name = "布鲁斯"                    # str — 字符串（可用单引号或双引号）
6. print(f"我叫{name}，今年{age}岁")  # f-string — 格式化字符串（Python 3.6+，推荐）
8. # ========== 列表（List）— 可变序列 ==========
10. fruits = ["apple", "banana", "cherry"]
11. print(fruits[0])                 # 用索引取值，从0开始
12. fruits.append("mango")           # 在末尾添加元素
13. squares = [x**2 for x in range(5)]  # 列表推导式，快速生成列表（LangChain大量使用）
14. print(squares)                   # 输出: [0, 1, 4, 9, 16]
16. # ========== 元组（Tuple）— 不可变序列 ==========
18. coordinates = (120.5, 30.6)       # 定义坐标（不可变）
19. print(coordinates[0])             # 输出: 120.5
21. # ========== 字典（Dict）— 键值对 ==========
23. person = {
24.     "name": "布鲁斯",
25.     "age": 37,
26.     "city": "深圳"
27. }
28. print(person["name"])                              # 通过键读取值
29. person["email"] = "test@example.com"               # 添加新键值对
30. print(person.get("phone", "未填写"))               # 安全取值，键不存在返回默认值
32. # 字典推导式
33. lengths = {fruit: len(fruit) for fruit in fruits}
34. print(lengths)
36. # ========== 集合（Set）— 无序不重复 ==========
38. tags = {"Python", "LangChain", "AI", "Python"}
39. print(tags)   # 输出: {'Python', 'LangChain', 'AI'}（自动去重）
41. # ========== 布尔（Bool） ==========
43. is_ai_dev = True
44. is_student = False
45. print(is_ai_dev and is_student)    # 输出: False（and=与）
46. print(is_ai_dev or is_student)     # 输出: True（或）
47. print(not is_student)              # 输出: True（非）
49. # ========== 空值 None ==========
51. result = None
52. print(result is None)              # 输出: True（判断None用 is 而非 ==）

复制代码

逐行解析

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

本节要学什么？

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

1. # ========== 条件判断 if / elif / else ==========
3. temperature = 38
5. if temperature > 40:
6.     print("危险！高温预警")
7. elif temperature > 37:
8.     print("发烧了，注意休息")
9. else:
10.     print("体温正常")
12. # ========== match 语句（Python 3.10+，类似switch）==========
14. status = "loading"
16. match status:
17.     case "success":
18.         print("请求成功")
19.     case "error":
20.         print("请求失败")
21.     case "loading":
22.         print("加载中...")
23.     case _:
24.         # _ 是通配符，表示"其他所有情况"
25.         print("未知状态")
27. # ========== 逻辑运算符：and / or / not ==========
29. isLoggedIn = True
30. isPremium = False
32. if isLoggedIn and isPremium:
33.     print("可以访问付费内容")
34. else:
35.     print("权限不足")
37. if isLoggedIn or isPremium:
38.     print("至少满足一个条件")
40. if not isPremium:
41.     print("不是Premium用户")
43. # ========== 三元表达式（一行if/else）==========
45. age = 20
46. category = "成年人" if age >= 18 else "未成年人"
47. print(category)  # 输出: 成年人
49. # ========== for 循环 ==========
51. languages = ["Python", "JavaScript", "Go"]
52. for lang in languages:
53.     print(f"我会{lang}")
55. scores = {"数学": 95, "英语": 88, "语文": 92}
56. for subject, score in scores.items():
57.     print(f"{subject}: {score}分")
59. for i in range(5):   # 0, 1, 2, 3, 4（共5次）
60.     print(f"第{i}次")
62. fruits = ["apple", "banana", "cherry"]
63. for idx, fruit in enumerate(fruits):
64.     print(f"{idx}: {fruit}")
66. # ========== while 循环 ==========
68. count = 0
69. while count < 3:
70.     print(f"count = {count}")
71.     count += 1  # 等价于 count = count + 1
73. # break 和 continue
74. for i in range(10):
75.     if i == 3:
76.         continue  # 跳过i==3这一次
77.     if i == 7:
78.         break     # 满足条件立即退出整个循环
79.     print(i)

复制代码

逐行解析

行号内容逐词解释作用1if temperature > 40:if=如果, temperature=变量, >大于, 40=比较值, :=条件结束如果温度大于40度，执行下一行2print("危险！高温预警")print=打印函数, 4空格缩进=属于if代码块条件满足时执行（严格4空格缩进是Python语法）4elif temperature > 37:elif=否则如果第一个if不满足时，尝试第二个条件7else:else=否则所有条件都不满足时执行13match status:match=模式匹配关键字, status=要匹配的变量开始一个 match 语句（Python 3.10+）14case "success":case=分支标签, "success"=匹配字符串常量如果 status 等于 "success"20case _:case=分支, _=通配符相当于 switch 的 default23isLoggedIn and isPremiumand=逻辑与两边必须同时为True才为True27isLoggedIn or isPremiumor=逻辑或任意一边为True即为True30not isPremiumnot=逻辑非把 False 变 True，True 变 False33category = "成年人" if age >= 18 else "未成年人"X if 条件 else Y=三元表达式条件为True取左边，False取右边37for lang in languages:for=遍历循环, lang=循环变量, in=在...里依次把列表中每个元素赋值给 lang40for subject, score in scores.items():.items()=字典键值对视图, (subject, score)=元组解包遍历字典的每一个键值对43for i in range(5):range(5)=生成0~4的整数序列循环5次，i 分别是 0,1,2,3,447for idx, fruit in enumerate(fruits):enumerate=枚举函数，同时返回索引和值既要索引又要值时用 enumerate51while count < 3:while=当...时循环,  str:    """    打招呼函数    name: 名字    return: 问候语字符串    """    return f"你好，{name}！"message = greet("布鲁斯")print(message)  # 输出: 你好，布鲁斯！# ========== 默认参数 ==========def powers(base: float, exponent: int = 2) -> float:    """    计算 base 的 exponent 次方    exponent 默认为2（平方）    """    return base ** exponentprint(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 ** 2print(square(5))  # 输出: 25# ========== 装饰器（LangChain工具常用）==========import functoolsdef 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_calldef add(a: int, b: int) -> int:    return a + bprint(add(3, 5))[/code]逐行解析

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

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

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

1. # ========== 基础函数 ==========
3. def greet(name: str) -> str:
4.     """
5.     打招呼函数
6.     name: 名字
7.     return: 问候语字符串
8.     """
9.     return f"你好，{name}！"
11. message = greet("布鲁斯")
12. print(message)  # 输出: 你好，布鲁斯！
14. # ========== 默认参数 ==========
16. def powers(base: float, exponent: int = 2) -> float:
17.     """
18.     计算 base 的 exponent 次方
19.     exponent 默认为2（平方）
20.     """
21.     return base ** exponent
23. print(powers(3))      # 输出: 9.0（只用默认值）
24. print(powers(3, 3))   # 输出: 27.0（覆盖默认值）
26. # ========== *args 和 **kwargs（可变参数）==========
28. def summarize(*args: str) -> str:
29.     """
30.     接收任意数量字符串，拼接成一个摘要
31.     *args 把所有位置参数收集成元组
32.     """
33.     return " | ".join(args)
35. result = summarize("LangChain", "是LLM应用框架", "支持多种模型")
36. print(result)  # 输出: LangChain | 是LLM应用框架 | 支持多种模型
38. def print_configs(**kwargs: object) -> None:
39.     """
40.     接收任意关键字参数并打印
41.     **kwargs 把所有关键字参数收集成字典
42.     """
43.     for key, value in kwargs.items():
44.         print(f"{key} = {value}")
46. print_configs(model="gpt-4o-mini", temperature=0.7, api_key="sk-xxx")
48. # ========== 类型注解进阶 ==========
50. from typing import Callable
52. # Union — 多种可能类型
53. def process(value: str | int) -> str:
54.     """value 可以是字符串或整数"""
55.     return str(value)
57. # Optional — 可能是None（等价于 Union[T, None]）
58. def find_user(user_id: int) -> dict[str, str] | None:
59.     """根据ID查找用户，找不到返回None"""
60.     users = {1: {"name": "张三", "city": "深圳"}, 2: {"name": "李四", "city": "北京"}}
61.     return users.get(user_id)  # 找不到返回None，不报错
63. user = find_user(1)
64. if user:
65.     print(user["name"])  # 输出: 张三
67. # Callable — 描述函数/可调用对象
68. def apply_twice(func: Callable[[int], int], x: int) -> int:
69.     """把函数 func 作用于 x 两次"""
70.     return func(func(x))
72. print(apply_twice(lambda n: n + 1, 0))  # 输出: 2
74. # ========== Lambda 匿名函数 ==========
76. square = lambda x: x ** 2
77. print(square(5))  # 输出: 25
79. # ========== 装饰器（LangChain工具常用）==========
81. import functools
83. def log_call(func: Callable) -> Callable:
84.     """打印函数调用信息的装饰器"""
85.     @functools.wraps(func)
86.     def wrapper(*args, **kwargs):
87.         print(f"调用函数: {func.__name__}")
88.         result = func(*args, **kwargs)
89.         print(f"函数返回: {result}")
90.         return result
91.     return wrapper
93. @log_call
94. def add(a: int, b: int) -> int:
95.     return a + b
97. print(add(3, 5))

复制代码

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

本节要学什么？

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

1. def run_task(task_id: str) -> dict:
2.     """执行任务并返回结果。"""
4.     # 先做输入校验，避免后续调用出现难定位的错误
5.     if not task_id:
6.         raise ValueError("task_id 不能为空")
8.     result = {"id": task_id, "status": "ok"}
9.     return result

复制代码

逐行解析

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

本节要学什么？

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

1. # ========== 最简单的类 ==========
3. class Dog:
4.     """一个简单的 Dog 类"""
5.     species = "哺乳动物"  # 类属性（所有实例共享）
7.     def __init__(self, name: str, age: int):
8.         # __init__ = 初始化方法（构造函数）
9.         # self = 当前实例对象本身（类似 this）
10.         self.name = name  # 实例属性
11.         self.age = age
13.     def bark(self) -> str:
14.         """实例方法"""
15.         return f"{self.name} 在叫：汪汪！"
17.     def __str__(self) -> str:
18.         """定义 print(对象) 时的输出格式"""
19.         return f"Dog(name={self.name}, age={self.age})"
21. # 创建对象（实例化）
22. dog1 = Dog("旺财", 3)
23. print(dog1.bark())       # 输出: 旺财 在叫：汪汪！
24. print(dog1.species)      # 输出: 哺乳动物（类属性）
25. print(str(dog1))         # 输出: Dog(name=旺财, age=3)
27. # ========== 继承 ==========
29. class LangChainTool:
30.     """LangChain 工具的基类"""
31.     name: str   # 类型注解：name 是字符串
32.     description: str
34.     def __init__(self, name: str, description: str):
35.         self.name = name
36.         self.description = description
38.     def run(self, query: str) -> str:
39.         """子类必须实现这个方法"""
40.         raise NotImplementedError("子类必须实现 run 方法")
42.     def __repr__(self) -> str:
43.         return f"{self.__class__.__name__}(name={self.name!r})"
45. class SearchTool(LangChainTool):
46.     """搜索工具（继承 LangChainTool）"""
48.     def __init__(self, name: str, description: str, api_key: str):
49.         super().__init__(name, description)  # 调用父类构造函数
50.         self.api_key = api_key               # 自己新增的属性
52.     def run(self, query: str) -> str:
53.         """实现父类的抽象方法（多态）"""
54.         return f"搜索「{query}」的结果（使用 {self.name}）"
56. search = SearchTool(
57.     name="WebSearch",
58.     description="搜索互联网信息",
59.     api_key="sk-xxx"
60. )
61. print(search.run("LangChain 教程"))
62. print(repr(search))
64. # ========== dataclass（Python 3.7+，LangChain大量使用）==========
66. from dataclasses import dataclass, field
67. from datetime import datetime
69. @dataclass
70. class LLMConfig:
71.     """LLM 配置数据类（自动生成 __init__ / __repr__ / __eq__）"""
72.     model: str                          # 必需字段
73.     temperature: float = 0.7           # 带默认值
74.     max_tokens: int = 2048
75.     created_at: datetime = field(default_factory=datetime.now)  # 工厂函数生成默认值
77. config = LLMConfig(model="gpt-4", temperature=0.9)
78. print(config)
80. config.temperature = 0.5               # dataclass 默认可变，可修改
81. print(config.temperature)

复制代码

逐行解析

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

本节要学什么？

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

1. import logging
2. from typing import Optional
4. # ========== 配置日志（比 print 强一万倍）==========
6. logging.basicConfig(
7.     level=logging.INFO,                        # 记录 INFO 及以上级别
8.     format="%(asctime)s | %(levelname)-8s | %(message)s",
9.     datefmt="%Y-%m-%d %H:%M:%S"
10. )
11. logger = logging.getLogger(__name__)           # __name__=当前模块名
13. logger.info("LangChain 应用启动")
14. logger.warning("这是警告级别日志")
15. logger.error("这是错误级别日志")
17. # ========== try / except / finally ==========
19. def call_llm(prompt: str, api_key: Optional[str]) -> str:
20.     """
21.     模拟调用 LLM
22.     演示异常处理的完整结构
23.     """
24.     try:
25.         if not api_key:
26.             raise ValueError("API Key 不能为空！")
28.         logger.info(f"发送请求: {prompt[:20]}...")
29.         result = f"LLM 对「{prompt}」的回复"
30.         logger.info("请求成功")
31.         return result
33.     except ValueError as e:
34.         # 捕获 ValueError 异常（参数错误）
35.         logger.error(f"参数错误: {e}")
36.         raise  # 重新抛出，让调用者知道发生了错误
38.     except (ConnectionError, TimeoutError) as e:
39.         # 同时捕获多种异常（网络类错误）
40.         logger.error(f"网络错误: {e}")
41.         return "网络异常，请稍后重试"
43.     except Exception as e:
44.         # 兜底：捕获所有未预料的异常（避免程序崩溃）
45.         logger.exception("发生了未知错误！")  # exception 会打印堆栈
46.         return "系统错误，请联系管理员"
48.     finally:
49.         # 不管有没有异常，finally 都会执行（适合清理工作）
50.         logger.info("请求处理完毕")
52. # ========== with 上下文管理器（自动关闭文件）==========
54. def read_config_file(filepath: str) -> str:
55.     """
56.     with 语句：自动管理资源（文件、网络连接等）
57.     离开 with 块时自动 close()，不需要手动处理
58.     """
59.     with open(filepath, "r", encoding="utf-8") as f:
60.         content = f.read()
61.     # 文件已自动关闭
62.     return content
64. # 写入 .env 配置文件（LangChain 常用 .env 存储 API Key）
65. env_content = """OPENAI_API_KEY=sk-xxx
66. MODEL_NAME=gpt-4o-mini
67. TEMPERATURE=0.7
68. """
69. with open(".env", "w", encoding="utf-8") as f:
70.     f.write(env_content)
72. # ========== 断言（开发时检查假设）==========
74. def divide(a: float, b: float) -> float:
75.     """除法，b 不能为 0"""
76.     if b == 0:
77.         raise ValueError("除数不能为 0！")
78.     return a / b
80. print(divide(10, 2))   # 输出: 5.0
81. # print(divide(10, 0))  # 抛出 ValueError: 除数不能为 0！

复制代码

逐行解析

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

本节要学什么？

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

1. import json          # JSON 序列化/反序列化（Python 内置）
2. import yaml          # 需要: uv add pyyaml
3. from pathlib import Path
4. from dotenv import load_dotenv  # 需要: uv add python-dotenv
6. # ========== 读取和写入 JSON ==========
8. config = {
9.     "model": "gpt-4o-mini",
10.     "temperature": 0.7,
11.     "max_tokens": 2048,
12.     "tools": ["search", "calculator", "wikipedia"]
13. }
15. # 写入 JSON 文件（indent=4 让格式更易读）
16. with open("config.json", "w", encoding="utf-8") as f:
17.     json.dump(config, f, ensure_ascii=False, indent=4)
19. # 读取 JSON 文件
20. with open("config.json", "r", encoding="utf-8") as f:
21.     loaded = json.load(f)  # json.load() 从文件读取，json.loads() 从字符串读取
23. print(loaded["model"])  # 输出: gpt-4
24. print(json.dumps(loaded, ensure_ascii=False))  # 字符串转 JSON
26. # ========== 读取和写入 YAML ==========
28. chain_config = {
29.     "chain_type": "LLMChain",
30.     "prompt": {
31.         "template": "请把以下中文翻译成英文：{text}",
32.         "input_variables": ["text"]
33.     },
34.     "llm": {"model_name": "gpt-4o-mini", "temperature": 0.5}
35. }
37. # 写入 YAML 文件
38. with open("chain.yaml", "w", encoding="utf-8") as f:
39.     yaml.dump(chain_config, f, allow_unicode=True, default_flow_style=False)
41. # 读取 YAML 文件
42. with open("chain.yaml", "r", encoding="utf-8") as f:
43.     loaded_yaml = yaml.safe_load(f)  # safe_load 只允许基本类型，避免执行任意代码
45. print(loaded_yaml["chain_type"])   # 输出: LLMChain
46. print(loaded_yaml["llm"]["model_name"])  # 输出: gpt-4o-mini
48. # ========== .env 文件与 python-dotenv ==========
50. # .env 文件内容（不要提交到 Git！）
51. # OPENAI_API_KEY=sk-xxx
52. # MODEL_NAME=gpt-4o-mini
54. load_dotenv()  # 加载 .env 文件到环境变量
56. import os
57. api_key = os.getenv("OPENAI_API_KEY")  # 从环境变量读取
58. model = os.getenv("MODEL_NAME", "gpt-4o-mini")  # 不存在时用默认值
60. masked_key = f"{api_key[:6]}***" if api_key else None
61. print(f"API Key: {masked_key}")
62. print(f"Model: {model}")
64. # ========== pathlib（更现代的文件路径操作）==========
66. project_root = Path.cwd()  # 教程示例用当前工作目录（脚本/Notebook 都可运行）
67. config_path = project_root / "config.json"  # 拼接路径（自动处理 / 或 \）
68. data_dir = project_root / "data"
70. # 创建目录（parents=True 不报错，exist_ok=True 目录存在不报错）
71. data_dir.mkdir(parents=True, exist_ok=True)
73. print(config_path.exists())  # 检查文件是否存在
74. print(config_path.suffix)     # 获取文件扩展名（.json）
75. print(config_path.stem)       # 获取不含扩展名的文件名（config）

复制代码

逐行解析

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

本节要学什么？

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

1. import requests   # 需要: uv add requests
2. import os
3. from dotenv import load_dotenv
5. load_dotenv()
6. api_key = os.getenv("OPENAI_API_KEY")
8. # ========== 最简单的 GET 请求 ==========
10. response = requests.get("https://httpbin.org/get")
11. print(response.status_code)  # HTTP 状态码（200=成功，404=未找到，500=服务器错误）
12. print(response.text)        # 响应原始文本
13. print(response.json())       # 把响应体解析成 Python 字典（自动 JSON 解码）
15. # ========== POST 请求（带 body）— LangChain 调用模型的本质 ==========
17. headers = {
18.     "Authorization": f"Bearer {api_key}",   # Bearer Token 认证方式
19.     "Content-Type": "application/json"        # 告诉服务器发送的是 JSON
20. }
22. payload = {
23.     "model": "gpt-4",
24.     "messages": [
25.         {"role": "system", "content": "你是一个有帮助的助手"},
26.         {"role": "user", "content": "用一句话解释什么是 LangChain"}
27.     ],
28.     "temperature": 0.7,
29.     "max_tokens": 500
30. }
32. response = requests.post(
33.     "https://api.openai.com/v1/chat/completions",
34.     headers=headers,
35.     json=payload,          # requests 自动把字典转成 JSON 并设置 Content-Type
36.     timeout=30             # 超时时间（秒），防止请求卡死
37. )
39. print(f"状态码: {response.status_code}")   # 200 表示成功
40. result = response.json()
41. print(result["choices"][0]["message"]["content"])  # 提取 LLM 的回复
43. # ========== 查询参数（URL 问号后面的参数）==========
45. params = {
46.     "q": "LangChain 教程",
47.     "page": 1,
48.     "per_page": 10
49. }
50. response = requests.get("https://httpbin.org/get", params=params)
51. print(response.url)   # 打印完整 URL（含编码后的参数）
53. # ========== 处理错误状态码 ==========
55. response = requests.get("https://httpbin.org/status/404", timeout=5)
56. print(response.status_code)  # 输出: 404
58. # response.raise_for_status()  # 有错误时抛出异常，没错误时什么都不做
60. if response.status_code == 200:
61.     print("请求成功")
62. elif response.status_code == 404:
63.     print("资源不存在")
64. elif response.status_code >= 500:
65.     print("服务器错误，稍后重试")
66. else:
67.     print(f"其他错误，状态码: {response.status_code}")
69. # ========== requests.Session（保持连接，提升性能）==========
71. session = requests.Session()
72. session.headers.update({"Authorization": f"Bearer {api_key}"})  # 全局headers，所有请求复用
74. # 连续发多个请求时，Session 复用 TCP 连接，速度更快
75. for i in range(3):
76.     resp = session.get(f"https://httpbin.org/get?request_id={i}", timeout=5)
77.     print(f"请求 {i}: {resp.status_code}")
79. session.close()  # 关闭 Session（也可以用 with: with requests.Session() as session:）

复制代码

逐行解析

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

本节要学什么？

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

1. # ========== 生成器函数（yield）==========
3. def count_to_5():
4.     """用 yield 返回值，每次返回一个后暂停函数"""
5.     for i in range(1, 6):
6.         yield i  # yield=生成并暂停，返回一个值后函数暂停在这里
7.         print(f"已yield {i}")  # 下次调用时从这里继续
9. # 生成器是惰性的：不会立刻执行函数体，只是创建一个生成器对象
10. gen = count_to_5()
11. print(gen)  # 输出: <generator object count_to_5 at 0x...>
13. # 每次 next() 取一个值（节省内存，适合处理大数据）
14. print(next(gen))  # 输出: 1（遇到yield暂停）
15. print(next(gen))  # 输出: 2（从暂停处继续，再遇到yield暂停）
16. print(next(gen))  # 输出: 3
18. # ========== 生成器表达式（类似列表推导式，但惰性）==========
20. # 列表推导式：一次性把所有平方算出来（占内存）
21. squares_list = [x**2 for x in range(1000000)]
23. # 生成器表达式：只记录规则，需要时才算（省内存）
24. squares_gen = (x**2 for x in range(1000000))
25. print(squares_gen)  # <generator object <genexpr> at 0x...>
26. print(next(squares_gen))  # 输出: 0
27. print(next(squares_gen))  # 输出: 1
29. # ========== 在 LangChain 中用生成器处理文档流 ==========
31. def stream_documents(documents: list[str]):
32.     """
33.     模拟流式读取文档（LangChain 的 RetrievalQA 会用到）
34.     每次 yield 一段文本，而不是一次性返回所有文本
35.     """
36.     for doc in documents:
37.         # 模拟把文档分成小段，一段一段地 yield
38.         words = doc.split()
39.         for word in words:
40.             yield word + " "
42. documents = ["LangChain 是一个应用框架", "它可以构建 LLM 应用", "支持多种模型"]
43. for chunk in stream_documents(documents):
44.     print(chunk, end="")  # 流式输出，不用等全部处理完
46. # ========== 上下文管理器（用类实现）==========
48. class Timer:
49.     """测量代码执行时间"""
51.     def __init__(self, name: str = "任务"):
52.         self.name = name
53.         self.start = None
54.         self.end = None
56.     def __enter__(self):
57.         """进入 with 块时执行（类似 try 块开头）"""
58.         import time
59.         self.start = time.time()
60.         print(f"[{self.name}] 开始")
61.         return self  # with 的 as 子句会收到这个返回值
63.     def __exit__(self, exc_type, exc_val, exc_tb):
64.         """离开 with 块时执行（类似 finally）"""
65.         import time
66.         self.end = time.time()
67.         elapsed = self.end - self.start
68.         print(f"[{self.name}] 结束，耗时 {elapsed:.4f} 秒")
69.         return False  # 返回 False 或 None 表示不拦截异常
71. # 使用 with 上下文管理器
72. with Timer("LangChain 文档处理") as timer:
73.     # 这里写要计时的代码
74.     total = sum(range(1000000))
75.     print(f"计算结果: {total}")
77. # ========== itertools（生成器工具库）==========
79. import itertools
81. # count() — 无限计数器（永不停止）
82. # counter = itertools.count(1)
83. # print(next(counter))  # 1, 2, 3, 4, ...
85. # islice — 从无限生成器里取前N个（不卡死）
86. limited = itertools.islice(itertools.count(1), 5)
87. print(list(limited))  # 输出: [1, 2, 3, 4, 5]
89. # chain — 把多个可迭代对象串起来
90. chain = itertools.chain([1, 2], ["a", "b"], [True, False])
91. print(list(chain))  # 输出: [1, 2, 'a', 'b', True, False]
93. # groupby — 按key分组
94. data = sorted([("cat", 1), ("dog", 2), ("cat", 3), ("dog", 4)], key=lambda x: x[0])
95. for key, group in itertools.groupby(data, key=lambda x: x[0]):
96.     print(f"{key}: {list(group)}")
97. # 输出: cat: [('cat', 1), ('cat', 3)]
98. #       dog: [('dog', 2), ('dog', 4)]

复制代码

实战注意事项

• 所有外部请求都要显式 timeout，并在错误时 raise_for_status()。
  
• 不要把 .env、API Key、用户敏感数据提交到 Git 或写进日志。
  
• 优先给函数写返回类型注解，结合 mypy 或 pyright 做静态检查。
  
• LLM 输出永远当“不可信输入”处理，先校验结构再使用。
  
• 把“可恢复错误”（超时、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 模板文件、.envyield生成器关键字流式输出（Streaming）super().__init__()调用父类构造器自定义 Tool、Custom Chainisinstance(x, type)类型检查判断 LLM 返回的是否为字符串functools.wraps保留函数元信息写装饰器时保持原函数名称不变pathlib.Path路径操作拼接配置文件路径.env + load_dotenv()环境变量隔离管理 API Key（不提交到 Git）

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

1. import asyncio
2. import os
3. from typing import Any
5. import httpx
8. async def fetch_json(client: httpx.AsyncClient, url: str) -> dict[str, Any]:
9.     """异步请求 + 超时 + 状态码检查。"""
10.     resp = await client.get(url, timeout=10.0)
11.     resp.raise_for_status()
12.     return resp.json()
15. async def main() -> None:
16.     # 不要打印完整密钥，最多打印前缀
17.     api_key = os.getenv("OPENAI_API_KEY")
18.     masked = f"{api_key[:6]}***" if api_key else None
19.     print("OPENAI_API_KEY:", masked)
21.     async with httpx.AsyncClient() as client:
22.         # 并发执行多个请求（真实项目里可并发查多个检索源）
23.         urls = ["https://httpbin.org/get", "https://httpbin.org/uuid"]
24.         tasks = [fetch_json(client, u) for u in urls]
25.         results = await asyncio.gather(*tasks, return_exceptions=True)
27.     for idx, item in enumerate(results):
28.         if isinstance(item, Exception):
29.             print(f"任务 {idx} 失败: {item}")
30.         else:
31.             print(f"任务 {idx} 成功: keys={list(item.keys())[:3]}")
34. if __name__ == "__main__":
35.     asyncio.run(main())

复制代码

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

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！