这份路线图面向已经熟悉 Java 后端开发的人,目标不是把 Python 学成另一门“全栈语言”,而是尽快掌握 Python Agent 生态中最常见的工程能力。
最终目标:
- 能阅读和运行 LangChain、LangGraph、AutoGen、LlamaIndex 等 Python 生态示例。
- 能用 Python 快速做 Agent 原型、数据处理、评估脚本。
- 能理解 Python Agent 生态中的主流工程范式。
需要掌握的核心主题:
- Python 基础语法
- FastAPI
- Pydantic
- asyncio
- Jupyter Notebook
- requests / httpx
- pandas
- 基础脚本能力
建议总周期:4 到 6 周。
如果每天只有 1 小时,按 6 周推进;如果每天有 2 到 3 小时,可以压缩到 4 周。
Python 在 Agent 生态中的使用方式,通常更接近“脚本 + 小型服务 + 数据处理 + 原型实验”的组合,而不是传统 Java 企业应用的大型分层工程。
| Java 常见概念 | Python 对应概念 | 学习重点 |
|---|---|---|
| Maven / Gradle | pip / uv / poetry / pyproject.toml | 会创建虚拟环境、安装依赖、固定版本 |
| package / classpath | module / package / import path | 理解文件即模块、目录即包 |
| POJO / DTO | dict / dataclass / Pydantic model | 区分轻量数据结构和强校验模型 |
| Jackson | Pydantic / json | JSON 序列化、反序列化、类型转换 |
| Bean Validation | Pydantic validators | 请求入参、配置、LLM 输出结构化校验 |
| Spring Boot Controller | FastAPI route | 快速暴露 HTTP API |
| RestTemplate / WebClient | requests / httpx | 同步和异步 HTTP 调用 |
| CompletableFuture / Reactor | asyncio / await | 协程、并发 IO、任务编排 |
| Stream API | list/dict comprehension / generator / pandas | 数据清洗、转换、聚合 |
| JUnit | pytest | 写可运行的脚本级测试和评估测试 |
| main 方法 | if name == "main" | 命令行脚本入口 |
核心转变:
- Python 更重视“快速表达”和“运行时组合”。
- 类型提示很重要,但它不是 Java 那样的强制编译期类型系统。
- Agent 生态里的很多对象都是“可调用对象、配置对象、消息对象、状态对象”的组合。
- 阅读 Python 示例时,要先抓住数据流和调用链,再纠结框架细节。
建议统一使用一个干净的项目目录和虚拟环境。
推荐掌握:
- Python 虚拟环境:
venv或uv - 依赖文件:
requirements.txt或pyproject.toml - 代码编辑器:VS Code / PyCharm 均可
- Notebook 环境:Jupyter Notebook 或 JupyterLab
- 常用命令:
python --version
python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn pydantic requests httpx pandas jupyter pytest
pip freeze > requirements.txtWindows PowerShell 激活虚拟环境:
.\.venv\Scripts\Activate.ps1建议建立这样的练习目录:
python-agent-learning/
notebooks/
scripts/
services/
data/
tests/
requirements.txt
README.md
目标:能无障碍阅读 Python 示例,能写简单脚本和函数。
建议用时:5 到 7 天。
- 基本类型:
int、float、str、bool、None - 容器类型:
list、tuple、dict、set - 控制流:
if、for、while、break、continue - 函数:默认参数、关键字参数、
*args、**kwargs - 推导式:list comprehension、dict comprehension
- 异常处理:
try、except、finally、自定义异常 - 文件操作:
open、pathlib.Path - 模块导入:
import、from ... import ... - 类和对象:
class、实例属性、类属性、继承 - 类型提示:
list[str]、dict[str, Any]、Optional、Callable - dataclass:轻量数据对象
- Python 缩进是语法,不是格式偏好。
- Python 变量是名字绑定,不是 Java 引用变量的完全等价物。
- 可变默认参数要避免:
def bad(items=[]):
items.append("x")
return items推荐写法:
def good(items=None):
if items is None:
items = []
items.append("x")
return items- Python 的
is判断对象身份,==判断值相等。 - 字典和列表在 Agent 示例中极常见,要非常熟。
- 不要一上来用复杂 OO 设计,Python 示例更偏函数式和数据流。
- 写一个脚本读取 JSON 文件,过滤出满足条件的数据,再输出成新的 JSON 文件。
- 写一个函数,把一批对话消息转换成 OpenAI / LangChain 常见的 message 格式。
- 用
pathlib扫描一个目录,统计.py、.md、.json文件数量。 - 写一个
dataclass表示一次模型调用记录:输入、输出、耗时、是否成功。
你应该能读懂下面这类代码:
from dataclasses import dataclass
from pathlib import Path
import json
@dataclass
class EvalItem:
question: str
answer: str
score: float | None = None
def load_items(path: str) -> list[EvalItem]:
rows = json.loads(Path(path).read_text(encoding="utf-8"))
return [EvalItem(**row) for row in rows]目标:能写日常自动化脚本、批处理脚本、实验脚本。
建议用时:3 到 5 天。
- 命令行参数:
argparse - 环境变量:
os.environ - 文件路径:
pathlib - JSON / CSV 读写:
json、csv - 日志:
logging - 时间处理:
datetime - 简单配置:
.env、YAML、JSON - 进度条:
tqdm - 测试:
pytest基础
- 批量跑提示词实验
- 批量调用模型 API
- 清洗知识库文档
- 把 Markdown / PDF / CSV 转成 chunk
- 统计评估结果
- 从日志中抽取失败样本
- 生成小规模 benchmark 报告
import argparse
import json
import logging
from pathlib import Path
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def main(input_path: str, output_path: str) -> None:
data = json.loads(Path(input_path).read_text(encoding="utf-8"))
result = [{"text": item["text"].strip()} for item in data if item.get("text")]
Path(output_path).write_text(
json.dumps(result, ensure_ascii=False, indent=2),
encoding="utf-8",
)
logger.info("Wrote %s records to %s", len(result), output_path)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--input", required=True)
parser.add_argument("--output", required=True)
args = parser.parse_args()
main(args.input, args.output)- 写一个
clean_texts.py,输入 JSONL,输出清洗后的 JSONL。 - 写一个
sample_dataset.py,从大文件中随机抽样 N 条。 - 写一个
merge_eval_results.py,合并多个评估结果文件。 - 用
pytest给其中一个脚本的核心函数写测试。
目标:能调用 LLM API、Agent 服务、内部 HTTP 接口,并理解同步和异步 HTTP 客户端差异。
建议用时:3 到 5 天。
requests 适合快速同步调用。
必须掌握:
GET/POST- headers
- query params
- JSON body
- timeout
- status code
- exception handling
- session 复用
示例:
import requests
def call_api(url: str, payload: dict) -> dict:
response = requests.post(
url,
json=payload,
timeout=30,
)
response.raise_for_status()
return response.json()httpx 同时支持同步和异步,是很多现代 Python 项目常用选择。
同步示例:
import httpx
def call_api(url: str, payload: dict) -> dict:
with httpx.Client(timeout=30) as client:
response = client.post(url, json=payload)
response.raise_for_status()
return response.json()异步示例:
import httpx
async def call_api(url: str, payload: dict) -> dict:
async with httpx.AsyncClient(timeout=30) as client:
response = await client.post(url, json=payload)
response.raise_for_status()
return response.json()- 用
requests调一个公开 HTTP API,保存返回结果。 - 用
httpx.Client封装一个同步 API client。 - 用
httpx.AsyncClient并发请求 10 个 URL。 - 给 API client 加上 timeout、重试、错误日志。
你应该能读懂大多数 Python Agent 示例里的:
- 模型 API 调用
- embedding API 调用
- tool API 调用
- webhook 调用
- 流式响应的基础结构
目标:能看懂和编写强类型数据模型,能用于 API 入参、配置、LLM 结构化输出校验。
建议用时:4 到 6 天。
BaseModel- 字段类型
- 默认值
- 可选字段
- 嵌套模型
- list / dict 字段
- 字段校验
- model dump / JSON 序列化
- 配置读取
- 错误信息理解
示例:
from pydantic import BaseModel, Field
class ToolCall(BaseModel):
name: str
arguments: dict = Field(default_factory=dict)
class AgentMessage(BaseModel):
role: str
content: str
tool_calls: list[ToolCall] = Field(default_factory=list)Pydantic 大致承担了这些角色的一部分:
- DTO
- JSON schema
- Jackson 反序列化
- Bean Validation
- 配置对象
- LLM structured output schema
但它比 Java DTO 更动态,常用于运行时把不稳定输入校验成稳定对象。
Pydantic 在 Agent 生态中非常常见:
- 定义工具入参 schema
- 定义 Agent state
- 定义结构化输出
- 校验模型返回 JSON
- 定义评估数据集 item
- 定义服务端请求和响应
- 定义
ChatMessage、ToolCall、AgentState三个模型。 - 读取一份 JSON,校验成
list[ChatMessage]。 - 故意制造错误数据,观察 Pydantic 错误信息。
- 把 Pydantic model 转成 dict 和 JSON。
- 用 Pydantic 定义一个 LLM 输出格式,比如:
{
"answer": "string",
"confidence": 0.8,
"citations": ["string"]
}目标:能理解 Python 异步代码,能写并发 HTTP 调用、并发 Agent 评估脚本。
建议用时:5 到 7 天。
async defawait- coroutine
- event loop
asyncio.runasyncio.gatherasyncio.create_task- timeout
- semaphore 限流
- 异步异常处理
| Java | Python |
|---|---|
CompletableFuture<T> |
coroutine / task |
future.get() |
await |
CompletableFuture.allOf |
asyncio.gather |
| WebClient | httpx.AsyncClient |
| Executor / thread pool | event loop / task |
| backpressure / rate limit | semaphore / queue |
import asyncio
import httpx
async def fetch(client: httpx.AsyncClient, url: str) -> str:
response = await client.get(url)
response.raise_for_status()
return response.text
async def main(urls: list[str]) -> list[str]:
async with httpx.AsyncClient(timeout=30) as client:
tasks = [fetch(client, url) for url in urls]
return await asyncio.gather(*tasks)
if __name__ == "__main__":
results = asyncio.run(main(["https://example.com"]))
print(len(results))需要异步的常见情况:
- 并发调用多个工具
- 并发跑评估集
- 并发抓取网页
- 并发调用 embedding API
- FastAPI 异步接口
- LangGraph / AutoGen 中的异步节点或异步消息流
- 用
asyncio.gather并发请求 20 个 URL。 - 用
asyncio.Semaphore限制最大并发为 5。 - 写一个异步评估脚本:并发处理 100 条问题,每条输出一个 JSONL 结果。
- 对失败任务做异常捕获,不影响其他任务继续执行。
目标:能用 Python 快速搭建 Agent 原型服务和 API 包装层。
建议用时:5 到 7 天。
- 创建 app
- 定义 route
- GET / POST
- path params / query params / request body
- Pydantic 请求和响应模型
- 依赖注入基础
- 异常处理
- 中间件基础
- 启动服务:
uvicorn - 自动生成 OpenAPI 文档
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class ChatRequest(BaseModel):
message: str
class ChatResponse(BaseModel):
answer: str
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest) -> ChatResponse:
return ChatResponse(answer=f"Echo: {request.message}")启动:
uvicorn main:app --reload| Spring Boot | FastAPI |
|---|---|
@RestController |
FastAPI() + route decorator |
@PostMapping |
@app.post |
@RequestBody |
Pydantic request model |
@PathVariable |
path parameter |
@RequestParam |
query parameter |
@ControllerAdvice |
exception handler |
| Swagger/OpenAPI starter | FastAPI 内置 OpenAPI |
POST /chatPOST /agent/invokePOST /agent/streamPOST /tools/{tool_name}/invokePOST /eval/runGET /health
- 写一个
/health接口。 - 写一个
/chat接口,接收用户问题,返回固定回答。 - 把请求和响应都改成 Pydantic model。
- 接入一个假 tool,比如天气查询、数据库查询或搜索函数。
- 写一个
/eval/run接口,接收问题列表,返回批量结果。
目标:能用 Notebook 做数据探索、Prompt 实验、Agent 调试和结果分析。
建议用时:2 到 4 天。
- cell 执行模型
- Markdown cell
- 变量状态
- 安装和选择 kernel
- 读取 CSV / JSON
- 展示 pandas DataFrame
- 简单可视化
- 导出结果
- 快速验证 Prompt
- 查看中间变量
- 对比模型输出
- 分析评估结果
- 查看 pandas 表格
- 做一次性实验记录
- 长期维护的核心业务逻辑
- 复杂服务端工程
- 大规模并发任务
- 需要严格代码审查的生产逻辑
建议原则:
- Notebook 用来探索。
- 稳定逻辑沉淀到
.py文件。 - 评估结果和实验结论写清楚。
- 重要实验固定随机种子和依赖版本。
- 新建一个 Notebook,读取一份 JSONL 评估结果。
- 用 pandas 展示前 10 行。
- 统计成功率、平均分、失败原因分布。
- 画一个简单柱状图。
- 把稳定的数据处理函数移动到
scripts/目录。
目标:能处理评估数据、日志数据、表格数据和实验结果。
建议用时:5 到 7 天。
DataFrameSeries- 读取 CSV / JSON / JSONL
- 查看数据:
head、info、describe - 选择列
- 过滤行
- 新增列
- 缺失值处理
- 排序
- 分组聚合:
groupby - join / merge
- 导出 CSV / JSON
import pandas as pd
df = pd.read_json("eval_results.jsonl", lines=True)
summary = (
df.groupby("model")
.agg(
avg_score=("score", "mean"),
count=("score", "count"),
success_rate=("success", "mean"),
)
.reset_index()
)
summary.to_csv("summary.csv", index=False)pandas 常用于:
- 读取评估集
- 合并多轮实验结果
- 统计模型表现
- 找出失败样本
- 按 prompt 版本分组对比
- 生成报表
- 观察延迟、token、成本
- 读取一份
eval_results.jsonl。 - 按模型名统计平均分。
- 按 prompt 版本统计成功率。
- 找出所有失败样本并导出。
- 合并
dataset.csv和eval_results.csv。 - 生成一个
summary.csv。
目标:能读懂 LangChain、LangGraph、AutoGen、LlamaIndex 的示例结构,知道它们分别解决什么问题。
建议用时:7 到 10 天。
这些框架虽然 API 不同,但常见抽象类似:
- message:用户、系统、助手、工具消息
- model / llm:模型调用对象
- prompt:提示词模板
- tool:可被模型或 Agent 调用的函数
- retriever:检索器
- vector store:向量库
- chain / workflow:顺序调用链
- graph:有状态的流程图
- state:Agent 执行状态
- memory:短期或长期上下文
- callback / tracing:日志、追踪、观测
- structured output:结构化输出
- evaluator:评估器
重点理解:
- model 调用
- prompt template
- output parser
- tool
- retriever
- chain composition
- LCEL 风格的管道组合
最小目标:
- 能跑通一个 chat 示例。
- 能加一个自定义 tool。
- 能做一个简单 RAG 示例。
- 能看懂 prompt -> model -> parser 的链路。
重点理解:
- graph
- node
- edge
- conditional edge
- state
- checkpoint
- human-in-the-loop
最小目标:
- 能定义一个包含 2 到 3 个节点的图。
- 能让 state 在节点之间传递。
- 能根据条件选择下一步。
- 能看懂 ReAct Agent 示例的大致执行流。
重点理解:
- agent
- multi-agent conversation
- message passing
- tool execution
- group chat
- termination condition
最小目标:
- 能跑通两个 Agent 对话示例。
- 能让一个 Agent 调用工具。
- 能理解多 Agent 协作的消息流。
重点理解:
- document
- node / chunk
- index
- retriever
- query engine
- response synthesizer
- metadata
最小目标:
- 能加载本地文档。
- 能构建一个简单索引。
- 能执行查询。
- 能理解文档切分、检索、生成回答的流程。
目标:用 Python 做一个最小评估工具。
功能:
- 读取
dataset.jsonl - 对每条样本调用一个模拟 LLM 函数
- 输出
eval_results.jsonl - 用 pandas 生成
summary.csv - 支持命令行参数
涉及能力:
- Python 基础
- 脚本能力
- JSONL
- pandas
- pytest
目标:用 asyncio 和 httpx 并发调用 API。
功能:
- 读取待处理任务
- 最大并发数可配置
- 每条任务有 timeout
- 失败自动记录错误
- 输出成功和失败结果
涉及能力:
- asyncio
- httpx
- logging
- argparse
- 错误处理
目标:搭一个最小 Agent API。
接口:
GET /healthPOST /chatPOST /tools/searchPOST /eval/run
功能:
- Pydantic 定义请求和响应
- 一个假 LLM 函数
- 一个工具函数
- 一个批量评估函数
涉及能力:
- FastAPI
- Pydantic
- asyncio
- httpx
- 基础工程结构
目标:用 Notebook 分析 Agent 输出质量。
功能:
- 读取多轮评估结果
- 对比不同 prompt 版本
- 统计成功率、平均分、平均耗时
- 展示失败样本
- 导出报告 CSV
涉及能力:
- Jupyter Notebook
- pandas
- 数据分析
- 实验记录
目标:理解 Python Agent / RAG 项目的基本结构。
功能:
- 读取本地 Markdown 文档
- 切分文本
- 构造简单检索器
- 根据 query 返回相关片段
- 把检索结果拼进 prompt
- 输出回答
涉及能力:
- Python 基础
- 文本处理
- Pydantic
- 脚本能力
- LlamaIndex 或 LangChain 入门
重点:
- 基础语法
- 容器
- 函数
- 文件
- JSON
- 类型提示
- dataclass
产出:
- 3 个小脚本
- 1 个 JSON 数据处理工具
- 1 个 pytest 测试文件
重点:
- argparse
- logging
- requests
- httpx
- Pydantic
产出:
- 一个 API client
- 一个 Pydantic 数据模型集合
- 一个批处理脚本
重点:
- async / await
- gather
- semaphore
- FastAPI route
- request / response model
产出:
- 一个异步批处理器
- 一个 FastAPI Agent 原型服务
重点:
- Jupyter Notebook
- pandas DataFrame
- groupby
- merge
- 评估结果分析
产出:
- 一个评估分析 Notebook
- 一个 summary 报表
- 一个失败样本分析文件
重点:
- prompt
- model
- parser
- tool
- retriever
- RAG
产出:
- 一个 LangChain chat + tool 示例
- 一个 LlamaIndex 本地文档问答示例
重点:
- graph state
- node
- edge
- multi-agent message flow
- tool execution
产出:
- 一个 LangGraph 多节点工作流
- 一个 AutoGen 双 Agent 协作示例
| 主题 | 最低掌握标准 |
|---|---|
| Python 基础语法 | 能读写函数、类、dict/list 操作、文件和 JSON 处理 |
| 基础脚本能力 | 能写带参数、日志、输入输出文件的命令行脚本 |
| requests / httpx | 能封装同步和异步 API 调用,处理 timeout 和错误 |
| Pydantic | 能定义嵌套模型,校验 JSON,作为 FastAPI 入参和响应 |
| asyncio | 能用 gather 并发跑任务,能用 semaphore 控制并发 |
| FastAPI | 能写一个带 Pydantic 模型的可运行 API 服务 |
| Jupyter Notebook | 能做实验、展示数据、分析评估结果 |
| pandas | 能读取、过滤、分组、聚合、导出实验数据 |
| Agent 框架 | 能跑通官方示例,能看懂 message、tool、state、workflow |
读 LangChain、LangGraph、AutoGen、LlamaIndex 示例时,不要从每个类的源码开始。建议按以下顺序:
- 看输入是什么:用户问题、文档、消息、状态。
- 看输出是什么:回答、结构化 JSON、工具结果、状态更新。
- 找模型调用位置。
- 找 prompt 或 system message。
- 找 tool 定义。
- 找状态对象或消息对象。
- 找执行入口:
invoke、ainvoke、run、stream、async for。 - 最后再看框架特有抽象。
常见执行入口:
result = chain.invoke(input_data)
result = await chain.ainvoke(input_data)
for chunk in chain.stream(input_data):
print(chunk)
async for chunk in chain.astream(input_data):
print(chunk)常见方式:
- 环境变量
.env- Pydantic settings
- YAML / JSON 配置文件
建议:
- API key 不要写死在代码里。
- 配置读取集中管理。
- 脚本参数和环境变量边界要清楚。
常见方式:
- 简单内部数据:
dict或dataclass - 边界输入输出:Pydantic model
- 表格数据:pandas DataFrame
- Agent 状态:TypedDict / Pydantic / dataclass
Agent tool 通常是普通 Python 函数加上 schema 描述。
好工具函数应具备:
- 明确入参
- 明确返回值
- 可测试
- 不依赖隐藏全局状态
- 错误信息可读
评估脚本通常包括:
- dataset loader
- model / agent runner
- evaluator
- result writer
- summary reporter
推荐输出 JSONL:
{"id":"1","input":"...","output":"...","score":0.8,"success":true}Agent 原型也要尽早记录:
- 输入
- 输出
- 工具调用
- 错误
- 耗时
- token 用量
- 模型名
- prompt 版本
做一个“Agent 实验台”小项目。
功能清单:
scripts/run_eval.py:批量运行评估集。scripts/summarize_eval.py:用 pandas 汇总结果。services/app.py:FastAPI 提供/chat和/eval/run。notebooks/analyze_eval.ipynb:分析结果。models.py:Pydantic 数据模型。clients.py:httpx API client。agent.py:最小 Agent 流程。tests/:核心函数测试。
推荐目录:
agent-lab/
data/
dataset.jsonl
notebooks/
analyze_eval.ipynb
scripts/
run_eval.py
summarize_eval.py
services/
app.py
agent_lab/
__init__.py
agent.py
clients.py
models.py
tools.py
tests/
test_models.py
test_tools.py
requirements.txt
README.md
完成后应具备的能力:
- 能快速看懂 Python Agent 示例。
- 能把示例改成自己的小原型。
- 能把原型包装成 API。
- 能并发跑评估。
- 能分析结果并输出报告。
- 能理解 Agent 工程中的模型、消息、工具、状态、评估、追踪这些主线。
- 能熟练使用 list / dict / set / tuple。
- 能写函数、类、dataclass。
- 能读写 JSON、CSV、文本文件。
- 能理解类型提示。
- 能用 pytest 写基础测试。
- 能用 argparse 写命令行参数。
- 能处理输入输出路径。
- 能写日志。
- 能处理异常。
- 能批量处理文件。
- 能用 requests 发送同步请求。
- 能用 httpx 发送同步和异步请求。
- 能处理 timeout。
- 能处理非 2xx 响应。
- 能封装简单 API client。
- 能定义 BaseModel。
- 能定义嵌套模型。
- 能校验 JSON 数据。
- 能理解校验错误。
- 能和 FastAPI 配合使用。
- 能解释 async / await。
- 能用 asyncio.run。
- 能用 gather 并发执行任务。
- 能用 semaphore 限制并发。
- 能处理异步任务异常。
- 能创建 app。
- 能写 GET / POST 接口。
- 能使用 Pydantic 请求和响应模型。
- 能启动 uvicorn。
- 能打开 OpenAPI 文档调试接口。
- 能创建 Notebook。
- 能读取和展示数据。
- 能记录实验过程。
- 能把稳定逻辑迁移到
.py文件。
- 能读取 CSV / JSONL。
- 能过滤和选择数据。
- 能 groupby 聚合。
- 能 merge 数据。
- 能导出结果。
- 能运行 LangChain 示例。
- 能运行 LangGraph 示例。
- 能运行 AutoGen 示例。
- 能运行 LlamaIndex 示例。
- 能解释 message、tool、state、workflow、retriever 的含义。
学习 Python Agent 生态时,不要追求一次性学完所有框架。更好的顺序是:
- 先把 Python 脚本、HTTP、Pydantic、asyncio 学扎实。
- 再用 FastAPI 把一个 Agent 原型服务化。
- 然后用 pandas 和 Notebook 建立评估闭环。
- 最后再进入 LangChain、LangGraph、AutoGen、LlamaIndex。
真正重要的不是记住每个框架的 API,而是理解这条主线:
输入数据 -> prompt / message -> model -> tool / retriever -> state -> output -> evaluation
掌握这条主线后,Python Agent 生态里的大部分示例都会变得容易阅读、运行和改造。