AI应用开发-python基础
·
第一层：最基础（第 1–3 天）
变量与数据类型
这一层的目标不是"背语法"，而是把每种数据类型的常用操作方法练熟，因为后面写 API 时会反复用到。
数字与字符串：
python
# 数字操作
price = 9.99
quantity = 3
total = price * quantity # 29.97
discount = round(total * 0.9, 2) # 26.97（round 防止浮点误差）
# 字符串操作（高频使用）
name = " 张三 "
name.strip() # "张三"（去掉首尾空白）
name.strip().lower() # "张三"（转小写，处理用户输入时常用）
url = "https://api.example.com/users/123"
url.split("/") # ['https:', '', 'api.example.com', 'users', '123']
url.startswith("https") # True
template = "你好，{}！你有 {} 条新消息。"
template.format("张三", 5) # "你好，张三！你有 5 条新消息。"
# f-string（推荐写法，更直观）
user_id = 42
msg = f"用户 {user_id} 登录成功"
filename = f"report_{user_id:04d}.pdf" # "report_0042.pdf"（补零对齐）列表——AI 应用里最常用的数据结构，存 messages、存 chunks、存检索结果：
python
messages = []
# 添加
messages.append({"role": "user", "content": "你好"})
messages.append({"role": "assistant", "content": "你好！"})
# 读取
messages[0] # 第一条
messages[-1] # 最后一条
messages[-5:] # 最后5条（做 context window 截断时常用）
# 遍历
for msg in messages:
print(msg["role"], msg["content"])
# 列表方法
chunks = ["chunk1", "chunk2", "chunk3"]
len(chunks) # 3
chunks.extend(["chunk4", "chunk5"]) # 合并列表
"chunk1" in chunks # True（成员检测）
chunks.index("chunk2") # 1（找位置）
# 列表推导（后面讲，但先见过）
contents = [msg["content"] for msg in messages if msg["role"] == "user"]字典——API 请求体、JSON 数据、数据库行都是字典：
python
user = {
"id": 1,
"name": "张三",
"email": "zhang@example.com",
"roles": ["user", "admin"],
}
# 读取（安全读法，键不存在时不报错）
user.get("name") # "张三"
user.get("age", 0) # 0（键不存在时返回默认值）
# 修改
user["name"] = "李四"
user["created_at"] = "2024-01-01"
# 遍历
for key, value in user.items():
print(f"{key}: {value}")
# 字典合并（Python 3.9+）
defaults = {"timeout": 30, "retries": 3}
config = {"timeout": 10, "model": "gpt-4o-mini"}
merged = defaults | config # {"timeout": 10, "retries": 3, "model": "gpt-4o-mini"}
# 常用操作
"email" in user # True（检查键是否存在）
user.keys() # 所有键
user.values() # 所有值
user.pop("roles") # 删除并返回值集合——做去重和成员检测：
python
seen_ids = set()
document_ids = [1, 2, 3, 2, 1, 4]
unique_ids = set(document_ids) # {1, 2, 3, 4}
# 快速成员检测（比列表快得多）
if doc_id not in seen_ids:
seen_ids.add(doc_id)
process(doc_id)控制流
python
# if/elif/else
def classify_status_code(code: int) -> str:
if code < 300:
return "成功"
elif code < 400:
return "重定向"
elif code < 500:
return "客户端错误"
else:
return "服务器错误"
# for 循环（最常用的遍历方式）
documents = [{"id": 1, "text": "..."}, {"id": 2, "text": "..."}]
for i, doc in enumerate(documents): # enumerate 同时得到索引和值
print(f"处理第 {i+1}/{len(documents)} 个文档")
process(doc)
# range
for i in range(0, 100, 10): # 0, 10, 20, ..., 90
print(i)
# while（适合"直到满足条件才停"的场景）
retry_count = 0
while retry_count < 3:
try:
result = call_api()
break # 成功就跳出
except Exception:
retry_count += 1
# 三元表达式（一行写简单的if/else）
role_label = "AI" if message["role"] == "assistant" else "用户"函数基础
python
# 基本定义
def add_message(messages: list, role: str, content: str) -> list:
messages.append({"role": role, "content": content})
return messages
# 默认参数
def create_chat_request(
messages: list,
model: str = "gpt-4o-mini", # 有默认值，可不传
temperature: float = 0.7,
max_tokens: int = 1000,
) -> dict:
return {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
}
# 调用时可以用关键字参数（顺序不重要）
request = create_chat_request(
messages=my_messages,
temperature=0.2, # 只覆盖想改的参数
)
# *args：接收任意数量位置参数
def log(*messages):
for msg in messages:
print(f"[LOG] {msg}")
log("启动服务", "连接数据库", "加载模型")
# **kwargs：接收任意数量关键字参数
def build_metadata(**kwargs) -> dict:
return {k: v for k, v in kwargs.items() if v is not None}
build_metadata(source="manual.pdf", page=3, section=None)
# → {"source": "manual.pdf", "page": 3}第二层：写出实用脚本（第 4–7 天）
文件操作与异常处理
python
import json
from pathlib import Path # 比 os.path 更现代
# 文件读写（with 语句保证文件自动关闭）
def load_config(path: str) -> dict:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
def save_results(data: list, path: str):
output = Path(path)
output.parent.mkdir(parents=True, exist_ok=True) # 自动创建目录
with open(output, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# 异常处理
def read_document(filepath: str) -> str:
try:
with open(filepath, "r", encoding="utf-8") as f:
return f.read()
except FileNotFoundError:
print(f"文件不存在: {filepath}")
return ""
except PermissionError:
print(f"无权限读取: {filepath}")
return ""
except Exception as e:
print(f"读取失败: {e}")
raise # 重新抛出，让上层处理
finally:
print("读取操作完成") # 不管成功失败都执行
# 自定义异常（FastAPI 里常用）
class DocumentNotFoundError(Exception):
def __init__(self, doc_id: str):
self.doc_id = doc_id
super().__init__(f"文档不存在: {doc_id}")
class ValidationError(Exception):
pass
# 使用
def get_document(doc_id: str):
doc = db.find(doc_id)
if not doc:
raise DocumentNotFoundError(doc_id)
return doc模块与包管理
python
# 标准库导入（不需要安装）
import os
import sys
import json
import time
import uuid
import datetime
from pathlib import Path
from typing import Optional, List, Dict, Union
from dataclasses import dataclass
from collections import defaultdict, Counter
# 第三方库导入（需要 pip install）
# pip install openai fastapi pydantic python-dotenv
# 从模块导入特定内容
from datetime import datetime, timedelta
from typing import Optional
# 相对导入（在你自己的项目里）
# app/
# services/
# rag.py ← 这里
# llm.py
# models/
# document.py
from .llm import call_llm # 同目录
from ..models.document import Document # 上一级目录
# 虚拟环境管理（每个项目必须做）
# python -m venv .venv
# source .venv/bin/activate (Mac/Linux)
# .venv\Scripts\activate (Windows)
# pip install -r requirements.txt
```
**`requirements.txt`** 的规范写法：
```
# requirements.txt
fastapi==0.115.0
uvicorn[standard]==0.30.0
openai==1.50.0
asyncpg==0.29.0
pydantic==2.9.0
pydantic-settings==2.5.0
python-dotenv==1.0.1
python-multipart==0.0.12 # FastAPI 文件上传必须推导式与函数式工具
python
# 列表推导式（AI 应用里非常高频）
messages = [
{"role": "user", "content": "问题1"},
{"role": "assistant", "content": "回答1"},
{"role": "user", "content": "问题2"},
]
# 提取所有用户消息
user_messages = [m["content"] for m in messages if m["role"] == "user"]
# ["问题1", "问题2"]
# 批量处理 chunks（清理空白）
clean_chunks = [chunk.strip() for chunk in raw_chunks if chunk.strip()]
# 字典推导式
scores = {"chunk_1": 0.92, "chunk_2": 0.78, "chunk_3": 0.65}
high_quality = {k: v for k, v in scores.items() if v > 0.8}
# {"chunk_1": 0.92}
# 嵌套列表展开
nested = [[1, 2], [3, 4], [5, 6]]
flat = [x for row in nested for x in row]
# [1, 2, 3, 4, 5, 6]
# sorted（返回新列表，不修改原列表）
chunks_by_score = sorted(chunks, key=lambda c: c["score"], reverse=True)
most_relevant = chunks_by_score[:3]
# any / all（检测集合条件）
has_system = any(m["role"] == "system" for m in messages)
all_valid = all(len(c["content"]) > 0 for c in chunks)第三层：写 FastAPI 必须的（第 8–12 天）
面向对象编程（OOP）
FastAPI 的 Depends、Pydantic 的 BaseModel、SQLAlchemy 的 ORM 都基于类，必须理解：
python
from dataclasses import dataclass
from typing import Optional
from datetime import datetime
# 基本类定义
class Document:
def __init__(self, filename: str, content: str, doc_type: str = "pdf"):
self.filename = filename
self.content = content
self.doc_type = doc_type
self.created_at = datetime.now()
self._chunk_count = 0 # 约定以 _ 开头的是"私有"属性
def get_summary(self, max_chars: int = 200) -> str:
"""返回内容摘要"""
return self.content[:max_chars] + "..." if len(self.content) > max_chars else self.content
@property
def chunk_count(self) -> int:
"""@property 让方法像属性一样访问"""
return self._chunk_count
@chunk_count.setter
def chunk_count(self, value: int):
if value < 0:
raise ValueError("chunk 数量不能为负")
self._chunk_count = value
@classmethod
def from_dict(cls, data: dict) -> "Document":
"""@classmethod 作为备用构造函数"""
return cls(
filename=data["filename"],
content=data["content"],
doc_type=data.get("doc_type", "unknown"),
)
def __repr__(self) -> str:
return f"Document(filename={self.filename!r}, chunks={self._chunk_count})"
# 继承
class PDFDocument(Document):
def __init__(self, filename: str, content: str, page_count: int):
super().__init__(filename, content, doc_type="pdf") # 调用父类 __init__
self.page_count = page_count
def get_summary(self, max_chars: int = 200) -> str:
base = super().get_summary(max_chars)
return f"[{self.page_count}页] {base}"
# dataclass（更简洁的数据类定义，自动生成 __init__、__repr__）
@dataclass
class SearchResult:
chunk_id: str
content: str
similarity: float
source: str
page: Optional[int] = None
def is_high_quality(self) -> bool:
return self.similarity > 0.85类型提示（重要，FastAPI + Pydantic 全依赖这个）
python
from typing import Optional, Union, Any
from collections.abc import Callable
# 基础类型
def greet(name: str, times: int = 1) -> str:
return (name + "! ") * times
# 复合类型（Python 3.10+ 可以用 | 替代 Union）
def process(data: list[str] | None) -> dict[str, int]:
...
# Optional（等价于 T | None，表示可以是 None）
def find_user(user_id: int) -> Optional[dict]:
...
# Pydantic 模型（FastAPI 里最常用的方式）
from pydantic import BaseModel, Field, field_validator
class ChatRequest(BaseModel):
message: str = Field(..., min_length=1, max_length=4000, description="用户消息")
session_id: Optional[str] = Field(None, description="会话 ID，为空时创建新会话")
temperature: float = Field(0.7, ge=0.0, le=2.0)
stream: bool = True
@field_validator("message")
@classmethod
def message_not_empty(cls, v: str) -> str:
if not v.strip():
raise ValueError("消息内容不能为空")
return v.strip()
class ChatResponse(BaseModel):
answer: str
session_id: str
sources: list[dict] = []
tokens_used: Optional[int] = None
# 用法
req = ChatRequest(message="你好", temperature=0.5)
print(req.model_dump()) # 转成字典
print(req.model_dump_json()) # 转成 JSON 字符串装饰器
理解装饰器是理解 FastAPI 路由 @app.get("/users") 的前提：
python
import time
import functools
import logging
logger = logging.getLogger(__name__)
# 装饰器的本质：一个接收函数并返回函数的函数
def timer(func):
@functools.wraps(func) # 保留原函数的 __name__ 等信息
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
duration = time.time() - start
logger.info(f"{func.__name__} 耗时 {duration:.3f}s")
return result
return wrapper
@timer
def expensive_operation():
time.sleep(0.5)
return "done"
# 等价于：expensive_operation = timer(expensive_operation)
# 带参数的装饰器（多一层嵌套）
def retry(max_attempts: int = 3, delay: float = 1.0):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts - 1:
raise
logger.warning(f"第 {attempt+1} 次失败，{delay}s 后重试: {e}")
time.sleep(delay)
return wrapper
return decorator
@retry(max_attempts=3, delay=2.0)
def call_external_api():
...
# 异步装饰器（FastAPI 里用 async def 时）
def async_timer(func):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
start = time.time()
result = await func(*args, **kwargs)
logger.info(f"{func.__name__} 耗时 {time.time()-start:.3f}s")
return result
return wrapper第四层：写生产级 API（第 13–18 天）
异步编程（async/await）
这是 FastAPI 高性能的核心。同步代码一次只能做一件事；异步代码在等待 IO（网络、数据库）时可以去做别的事：
python
import asyncio
import httpx
import asyncpg
# 基础 async/await
async def fetch_embedding(text: str) -> list[float]:
# await 表示"等这个完成，但期间可以去做别的"
async with httpx.AsyncClient() as client:
response = await client.post(
"https://api.openai.com/v1/embeddings",
json={"model": "text-embedding-3-small", "input": text},
headers={"Authorization": f"Bearer {API_KEY}"},
)
return response.json()["data"][0]["embedding"]
# 并发执行多个任务（比串行快得多）
async def embed_batch(texts: list[str]) -> list[list[float]]:
# 串行：texts 一条一条处理，总耗时 = N × 单条耗时
# 并发：同时发出所有请求，总耗时 ≈ 单条耗时
tasks = [fetch_embedding(text) for text in texts]
return await asyncio.gather(*tasks) # 并发等待所有任务
# 带超时控制
async def safe_fetch(text: str) -> list[float] | None:
try:
return await asyncio.wait_for(
fetch_embedding(text),
timeout=10.0 # 超过10秒抛出 asyncio.TimeoutError
)
except asyncio.TimeoutError:
logger.error(f"embedding 请求超时: {text[:50]}")
return None
# 在 FastAPI 里（async def 路由自动在异步事件循环里运行）
@app.post("/embed")
async def embed_endpoint(text: str):
embedding = await fetch_embedding(text)
return {"embedding": embedding, "dimensions": len(embedding)}
# 运行异步函数（在脚本里，不在 FastAPI 里）
if __name__ == "__main__":
result = asyncio.run(embed_batch(["文本1", "文本2"]))同步 vs 异步对比（直观感受差异）：
python
# 同步版本：总耗时 = 3 × 1秒 = 3秒
import time
def sync_fetch(url): time.sleep(1); return f"result from {url}"
results = [sync_fetch(url) for url in ["url1", "url2", "url3"]]
# 异步版本：总耗时 ≈ 1秒
import asyncio, httpx
async def async_fetch(url):
await asyncio.sleep(1) # 模拟网络等待
return f"result from {url}"
async def main():
results = await asyncio.gather(
async_fetch("url1"),
async_fetch("url2"),
async_fetch("url3"),
)
return results
asyncio.run(main())环境与配置管理
python
# pip install pydantic-settings python-dotenv
from pydantic_settings import BaseSettings
from pydantic import Field
from functools import lru_cache
class Settings(BaseSettings):
# 必填项（没有默认值，不配置就报错）
openai_api_key: str
database_url: str
# 选填项（有默认值）
environment: str = "development"
log_level: str = "INFO"
max_chunk_size: int = 800
embedding_model: str = "text-embedding-3-small"
llm_model: str = "gpt-4o-mini"
llm_temperature: float = 0.7
# Sentry 等可选服务
sentry_dsn: str | None = None
class Config:
env_file = ".env" # 从 .env 文件读取
env_file_encoding = "utf-8"
case_sensitive = False # DATABASE_URL 和 database_url 等价
@lru_cache() # 缓存，避免重复读取文件
def get_settings() -> Settings:
return Settings()
# 在代码里使用
settings = get_settings()
print(settings.openai_api_key)
print(settings.llm_model)
# 在 FastAPI 里通过依赖注入
from fastapi import Depends
@app.get("/info")
def get_info(settings: Settings = Depends(get_settings)):
return {"model": settings.llm_model, "env": settings.environment}生成器（流式输出的基础）
python
# 基本生成器：yield 暂停并返回一个值
def count_up(max_n: int):
n = 0
while n < max_n:
yield n # 暂停，返回 n，等待下一次 next()
n += 1
for i in count_up(5):
print(i) # 0 1 2 3 4
# 生成器表达式（懒求值，不会立即计算所有结果）
large_data = range(10_000_000)
squares = (x**2 for x in large_data if x % 2 == 0) # 还没计算
first_ten = [next(squares) for _ in range(10)] # 只算了10个
# 在 FastAPI 流式响应里的应用（重要！）
async def stream_llm_response(messages: list) -> AsyncGenerator[str, None]:
stream = await openai_client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield f"data: {delta}\n\n" # 每次 yield 一个 SSE 数据块
yield "data: [DONE]\n\n"
@app.post("/chat/stream")
async def chat_stream(body: ChatRequest):
return StreamingResponse(
stream_llm_response(body.messages),
media_type="text/event-stream",
)完整的学习顺序（21 天安排）
|
天数 |
学习内容 |
当天产出 |
|
第 1 天 |
安装 Python、VS Code、运行第一个脚本 |
能跑 |
|
第 2 天 |
变量、数字、字符串操作 |
字符串处理练习 20 道 |
|
第 3 天 |
列表、字典、集合常用方法 |
数据处理脚本 |
|
第 4 天 |
、 、 、 |
猜数字游戏 |
|
第 5 天 |
函数定义、参数、返回值、默认值 |
封装5个工具函数 |
|
第 6 天 |
文件读写、 、 |
JSON 配置读写脚本 |
|
第 7 天 |
异常处理、 、自定义异常 |
健壮的文件处理脚本 |
|
第 8 天 |
模块导入、 、虚拟环境 |
创建第一个项目结构 |
|
第 9 天 |
列表推导式、字典推导式、 、 |
数据变换练习 |
|
第 10–11 天 |
类、 、方法、 、继承 |
设计 类 |
|
第 12 天 |
类型提示、Pydantic 、 |
定义 API 请求/响应模型 |
|
第 13 天 |
装饰器原理、 、带参装饰器 |
写 和 |
|
第 14–15 天 |
、 、 |
异步批量 HTTP 请求 |
|
第 16 天 |
、 、配置管理 |
项目配置模块 |
|
第 17 天 |
生成器、 、异步生成器 |
流式输出 demo |
|
第 18–21 天 |
综合项目：Task Manager API |
FastAPI + PostgreSQL 完整 API |
常用标准库速查
python
# 时间处理
from datetime import datetime, timedelta, timezone
now = datetime.now(timezone.utc)
one_week_later = now + timedelta(days=7)
timestamp = now.isoformat() # "2024-01-15T08:30:00+00:00"
# UUID（数据库主键）
import uuid
new_id = str(uuid.uuid4()) # "550e8400-e29b-41d4-a716-446655440000"
# JSON
import json
data = {"key": "value", "num": 42}
json_str = json.dumps(data, ensure_ascii=False, indent=2)
back = json.loads(json_str)
# 正则表达式
import re
emails = re.findall(r"[\w.+-]+@[\w-]+\.[a-zA-Z]{2,}", text)
clean = re.sub(r"\s+", " ", messy_text).strip()
# 哈希（密码处理）
import hashlib
hash_val = hashlib.sha256("password".encode()).hexdigest()
# 环境变量
import os
api_key = os.getenv("OPENAI_API_KEY", "") # 有默认值
required = os.environ["DATABASE_URL"] # 不存在会报错
# 随机数
import random
random.choice(["a", "b", "c"])
random.shuffle(my_list)
random.randint(1, 100)最容易踩的坑（提前避开）
可变默认参数是经典陷阱——默认值是列表或字典时，所有调用共享同一个对象：
python
# 错误写法（所有调用共享同一个列表）
def add_item(item, items=[]):
items.append(item)
return items
add_item("a") # ["a"]
add_item("b") # ["a", "b"] ← 意外！
# 正确写法
def add_item(item, items=None):
if items is None:
items = []
items.append(item)
return items字符串是不可变的，所有修改方法都返回新字符串：
python
text = "hello"
text.upper() # 返回 "HELLO"，text 本身没变
text = text.upper() # 必须重新赋值才有效整数除法 vs 浮点除法：
python
5 / 2 # 2.5（浮点除法，结果是 float）
5 // 2 # 2（整数除法，向下取整）
5 % 2 # 1（取余）is vs == 的区别：
python
a = [1, 2, 3]
b = [1, 2, 3]
a == b # True（值相等）
a is b # False（不是同一个对象）
# 只有这些情况用 is
x = None
if x is None: # 正确
...
if x == None: # 能跑但不推荐
...掌握这四层知识点后，你写 FastAPI 路由、Pydantic 模型、异步数据库操作时会非常顺手，不会被 Python 语法本身卡住。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
19
0- 0
已为社区贡献8条内容
所有评论(0)