Python 基础
🎯 学习目标
- 搭建虚拟环境,用
pip管理本课程依赖- 掌握阅读 LangChain / RAG 示例所需的语法:结构、函数、类、类型标注
- 会用
json、pathlib、httpx处理配置与 API 调用- 了解
async/await与流式响应的关系,遇语法问题时知道查什么
引言
非主线选修,但第六章 LangChain、第四章 RAG 与多数 Agent 后端示例以 Python 为主。有 JS/Java 背景的同学建议 45–60 分钟 速览本节,然后边做 实战 · 流式 API 与 LangChain 概览 边查。
全栈同学常见分工:Python 跑 RAG 索引、批处理、LangGraph;Node/Next.js 跑 API 与流式 UI——两栈都要会读示例代码。
章节正文
第 1 步:环境与包管理
# 建议 Python 3.11+
python3 --version
# 项目内虚拟环境(推荐)
cd my-agent
python3 -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install --upgrade pip
pip install openai python-dotenv httpx langchain langchain-openai
pip freeze > requirements.txt # 提交锁版本,团队 pip install -r requirements.txt| 文件 | 作用 |
|---|---|
.venv/ | 隔离依赖,勿提交 Git |
requirements.txt | 可复现安装列表 |
.env | API Key,勿提交 Git(配 .env.example) |
读取环境变量:
import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.environ["OPENAI_API_KEY"] # 缺失会 KeyError
base_url = os.getenv("OPENAI_BASE_URL") # 缺失返回 None
model = os.getenv("OPENAI_MODEL", "gpt-4o-mini") # 带默认值第 2 步:变量、结构与推导式
name = "LLM"
nums = [1, 2, 3]
user = {"id": 1, "role": "admin", "tags": ["rag", "agent"]}
unique = {1, 2, 3} # set
point = (10, 20) # tuple,不可变
# 遍历
for n in nums:
if n > 1:
print(n)
for k, v in user.items():
print(k, v)
# 列表推导 — 数据处理里极常见
squares = [x * x for x in nums if x > 1] # [4, 9]
tags_upper = [t.upper() for t in user["tags"]]
# 字典推导
index = {t: i for i, t in enumerate(user["tags"])}None 与布尔:空列表 []、None、0 在 if 中为假;判断「有值」写 if items: 而非 if items is not None:(除非要区分空列表与 None)。
第 3 步:函数、类型标注与类
LangChain 示例大量使用类型标注,能读懂即可:
def greet(name: str, lang: str = "zh") -> str:
return f"Hello, {name}"
def call_api(messages: list[dict[str, str]], model: str = "gpt-4o-mini") -> str:
"""messages 形如 [{"role": "user", "content": "..."}]"""
return "mock reply"
class ChatClient:
def __init__(self, api_key: str, base_url: str | None = None):
self.api_key = api_key
self.base_url = base_url
def send(self, message: str) -> str:
return message
# 解包
first, *rest = [1, 2, 3, 4] # first=1, rest=[2,3,4]
a, b = {"x": 1, "y": 2}.values()dataclass(配置对象常用):
from dataclasses import dataclass
@dataclass
class ModelConfig:
provider: str
model: str
temperature: float = 0.7
cfg = ModelConfig(provider="openai", model="gpt-4o-mini")
print(cfg.model)结构化输出场景会见到 Pydantic(第六章 Output Parser),此处知道「用模型类校验 JSON 字段」即可。
第 4 步:模块、路径与 JSON
# 同目录模块
from utils.prompts import build_system_prompt
# 标准库
import json
from pathlib import Path
config_path = Path(__file__).resolve().parent / "config.json"
raw = config_path.read_text(encoding="utf-8")
data = json.loads(raw)
# 写回
config_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")RAG 流水线里常见:Path("docs").glob("**/*.md") 遍历文档目录。
第 5 步:文件、异常与上下文管理器
try:
with open("config.json", encoding="utf-8") as f:
data = json.load(f)
except FileNotFoundError:
data = {}
except json.JSONDecodeError as e:
raise ValueError(f"invalid config: {e}") from e
finally:
pass # 清理资源(少用,with 已够用)自定义异常便于 Agent 工具返回可读错误:
class ToolError(Exception):
def __init__(self, code: str, message: str):
self.code = code
super().__init__(message)第 6 步:HTTP 与 OpenAI SDK(同步)
与 2.1 API 与本地部署 对齐:
from openai import OpenAI
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "你是简洁中文助手。"},
{"role": "user", "content": "什么是 Token?"},
],
max_tokens=200,
)
print(resp.choices[0].message.content)
print(resp.usage)裸 httpx(调试 curl、理解 HTTP 形态):
import httpx
r = httpx.post(
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"},
json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hi"}]},
timeout=60.0,
)
r.raise_for_status()
print(r.json()["choices"][0]["message"]["content"])务必设置 timeout:LLM 响应可能数十秒。
第 7 步:async/await 与流式(入门)
2.4 SSE 与流式输出 的后端常用异步:
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
async def stream_chat(prompt: str) -> None:
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
print(delta, end="", flush=True)
asyncio.run(stream_chat("用三句话介绍 RAG"))要点:
async def定义协程;await等待 I/Oasync for消费流式 chunk- FastAPI 路由可写
async def chat(...)直接对接前端 SSE
第 8 步:本课程常见第三方库速查
| 库 | 章节 | 用途 |
|---|---|---|
openai | 2.x | Chat Completions、Embedding |
httpx / requests | 2.x | 裸 HTTP、调试 |
python-dotenv | 2.x | .env 加载 |
langchain* | 6.x | Chain、RAG、Agent |
langgraph | 6.6 | 状态图 Agent |
chromadb / faiss | 4.x | 向量库(视示例而定) |
pydantic | 2.6 / 6.4 | 结构化输出校验 |
安装冲突时优先查 Python 版本(建议 3.11+)与 requirements.txt pinned 版本。
动手练习
- 创建
.venv,安装openai、python-dotenv,写chat.py读.env发一条消息并打印usage。 - 写函数
load_messages(path: Path) -> list[dict],从 JSON 文件读取多轮messages。 - 把同步
chat.py改成async+stream=True,终端逐字打印回复。 - 故意传错 Key,用
try/except捕获并打印友好错误(区分 401 与其它异常)。
常见问题
Q:pip install 报权限错误?
不要用 sudo pip。确认已 source .venv/bin/activate,提示符前有 (.venv)。
Q:相对导入 ImportError?
脚本直接 python tools/foo.py 时包路径可能不对。开发期可用 python -m package.module,或把项目根加入 PYTHONPATH。
Q:要和 Node 全栈怎么配合?
Python 提供 /rag/query 或 gRPC 服务;Next.js 只调你的 HTTP 接口,Key 留在 Python 侧。详见 Node 基础 与 6.7 框架选型。
本节小结
跑通本课程 Python 示例需要:虚拟环境、dict/list 与推导式、带类型标注的函数与类、pathlib/json、异常处理、openai SDK 与基础 async 流式。不必先系统学完 Python 再动手——边做 Lab 边查本节即可。