Skip to content

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 步:环境与包管理 ​

bash
# 建议 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可复现安装列表
.envAPI Key,勿提交 Git(配 .env.example)

读取环境变量:

python
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 步:变量、结构与推导式 ​

python
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 示例大量使用类型标注,能读懂即可:

python
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(配置对象常用):

python
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 ​

python
# 同目录模块
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 步:文件、异常与上下文管理器 ​

python
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 工具返回可读错误:

python
class ToolError(Exception):
    def __init__(self, code: str, message: str):
        self.code = code
        super().__init__(message)

第 6 步:HTTP 与 OpenAI SDK(同步) ​

与 2.1 API 与本地部署 对齐:

python
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 形态):

python
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 与流式输出 的后端常用异步:

python
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/O
  • async for 消费流式 chunk
  • FastAPI 路由可写 async def chat(...) 直接对接前端 SSE

第 8 步:本课程常见第三方库速查 ​

库章节用途
openai2.xChat Completions、Embedding
httpx / requests2.x裸 HTTP、调试
python-dotenv2.x.env 加载
langchain*6.xChain、RAG、Agent
langgraph6.6状态图 Agent
chromadb / faiss4.x向量库(视示例而定)
pydantic2.6 / 6.4结构化输出校验

安装冲突时优先查 Python 版本(建议 3.11+)与 requirements.txt pinned 版本。

动手练习 ​

  1. 创建 .venv,安装 openai、python-dotenv,写 chat.py 读 .env 发一条消息并打印 usage。
  2. 写函数 load_messages(path: Path) -> list[dict],从 JSON 文件读取多轮 messages。
  3. 把同步 chat.py 改成 async + stream=True,终端逐字打印回复。
  4. 故意传错 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 边查本节即可。