






















零基础回顾,面向 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
| 行号 | 内容 | 逐词解释 | 作用 |
|---|---|---|---|
# 安装指定版本的 Python(编译要几分钟)
pyenv install 3.11.7
# 列出已安装的所有 Python 版本
pyenv versions
# 设置全局默认版本(整个系统都用这个)
pyenv global 3.11.7
# 在当前文件夹设定局部版本(只影响这个项目)
pyenv local 3.11.7
# 卸载某个版本
pyenv uninstall 3.10.5
| 命令 | 逐词解释 | 作用 |
|---|---|---|
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
| 行号 | 内容 | 逐词解释 | 作用 |
|---|---|---|---|
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 而非 ==)
| 行号 | 内容 | 逐词解释 | 作用 |
|---|---|---|---|
条件判断(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)
| 行号 | 内容 | 逐词解释 | 作用 |
|---|---|---|---|
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))
| 行号 | 内容 | 逐词解释 | 作用 |
|---|---|---|---|
在团队代码里,推荐遵循这个顺序:
docstring("""...""")。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)
| 行号 | 内容 | 逐词解释 | 作用 |
|---|---|---|---|
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!
| 行号 | 内容 | 逐词解释 | 作用 |
|---|---|---|---|
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)
| 行号 | 内容 | 逐词解释 | 作用 |
|---|---|---|---|
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:)
| 行号 | 内容 | 逐词解释 | 作用 |
|---|---|---|---|
上下文管理器(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)]
| 行号 | 内容 | 逐词解释 | 作用 |
|---|---|---|---|
前面 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())
timeout,并在错误时 raise_for_status()。.env、API Key、用户敏感数据提交到 Git 或写进日志。mypy 或 pyright 做静态检查。| 语法 | 含义 | 在 LangChain 中的用途 |
|---|---|---|
下一步:建议直接用
uv init my_langchain_project创建项目,然后:uv add langchain langchain-openai uv run python -c "from langchain_openai import ChatOpenAI; print('LangChain 环境 OK!')"环境验证通过后,去 LangChain 官方文档 开始你的 LLM 应用开发之旅!
此内容由惯性聚合(RSS阅读器)自动聚合整理,仅供阅读参考。 原文来自 — 版权归原作者所有。