跳到主要内容

Python 工程基础——智能体的"骨骼支架"

Python 不仅是 Agentic AI 开发的首选语言,更是将想法快速固化为可测试、可维护、可扩展的生产级代码的唯一途径。

前置知识

  • 无,从零开始
  • 建议已了解:基本编程概念(变量、函数、循环)

核心概念

为什么选 Python

Agentic AI 工程的本质是胶水 + 编排——将 LLM API、向量库、外部工具、业务逻辑粘合在一起。Python 在这一场景中的优势是压倒性的:

维度PythonNode.jsGoJava
AI 生态★★★★★★★★★★★★★
原型速度★★★★★★★★★★★★★★
类型安全★★★★★★★★★★★★★★★★★
异步支持★★★★★★★★★★★★★★★★
生产成熟度★★★★★★★★★★★★★★★★★★

关键洞察:对于 Agent 工程,"胶水 + 编排"场景下 Python 的原型速度远超其他语言,且拥有最完善的 AI 生态。

2026 年 Agentic AI 工程师必须掌握的 Python 能力


1. 类型注解——Agent 的第一道防线

Agent 系统中,类型注解不只是"好习惯",而是防止 LLM 产生幻觉参数的硬屏障

# 差的写法——没有类型,LLM 无法理解工具的契约
def search(query, limit=5):
return db.search(query)[:limit]

# 好的写法——完整的类型 + Pydantic 验证
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
"""搜索工具的输入契约。"""
query: str = Field(
description="搜索关键词,尽量简洁,建议不超过 50 字",
max_length=200,
min_length=1,
)
limit: int = Field(
default=5,
ge=1, le=20,
description="返回结果数量,1-20 之间",
)
category: str | None = Field(
default=None,
description="可选:按类别过滤,如 '技术文档'、'FAQ'",
)

class SearchResult(BaseModel):
"""搜索工具的输出契约。"""
title: str
url: str
snippet: str
score: float = Field(description="相关度分数,0-1 之间", ge=0, le=1)

class SearchOutput(BaseModel):
results: list[SearchResult]
count: int
error: str = Field(default="", description="错误信息,无错误则为空")

为什么重要

  • LLM 通过 Schema 理解工具的输入输出格式
  • Pydantic 在运行时自动验证,防止类型错误
  • Field 中的 description 直接作为 LLM 的工具描述

2. 异步编程——Agent 的并发引擎

Agent 系统大量调用外部 API(LLM、向量库、搜索服务),异步是提升吞吐量的关键

import asyncio
import aiohttp

async def fetch_url(session: aiohttp.ClientSession, url: str, timeout: float = 10.0) -> str:
"""异步获取 URL 内容——网络密集型操作。"""
try:
async with asyncio.timeout(timeout):
async with session.get(url) as resp:
resp.raise_for_status()
return await resp.text()
except asyncio.TimeoutError:
return f"超时: {url}"
except aiohttp.ClientError as e:
return f"请求失败: {url} - {e}"

async def fetch_all(urls: list[str]) -> list[str]:
"""并发获取多个 URL——典型 Agent 工具调用场景。"""
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
return await asyncio.gather(*tasks, return_exceptions=True)

# 同步 vs 异步性能对比
# 10 个 URL,每个耗时 ~200ms
# 同步: 10 × 200ms = 2000ms
# 异步: max(200ms) ≈ 250ms(含 overhead)
# 加速比: ~8x

核心原则

任务类型推荐方式原因
网络 I/O(API 调用、HTTP 请求)async/await不阻塞,高并发
向量检索async + 异步向量库与 LLM 调用并行
CPU 密集型(嵌入计算、图像处理)后台任务队列(Celery)不阻塞事件循环
数据库写入异步 ORM 或连接池避免锁等待

3. 项目结构分层——可维护性的基础

my-agent/
├── pyproject.toml # 项目元数据 + 依赖锁定
├── .env.example # 环境变量模板
├── .env # 本地配置(gitignore)

├── src/
│ └── my_agent/
│ ├── __init__.py
│ ├── main.py # 入口:FastAPI 应用启动
│ ├── config.py # 配置管理(Pydantic Settings)
│ │
│ ├── domain/ # 领域逻辑
│ │ ├── models.py # 数据模型
│ │ └── services.py # 业务逻辑
│ │
│ ├── agents/ # 智能体实现
│ │ ├── graph.py # LangGraph 状态图定义
│ │ ├── nodes.py # 节点实现(检索、综合等)
│ │ └── prompts/ # 提示词模板
│ │ ├── system.txt
│ │ └── qa.txt
│ │
│ ├── tools/ # 工具封装
│ │ ├── search.py # 搜索工具
│ │ ├── web.py # 网络工具
│ │ └── database.py # 数据库工具
│ │
│ ├── rag/ # RAG 模块
│ │ ├── chunker.py # 分块策略
│ │ ├── retriever.py# 检索器
│ │ └── vector_store.py # 向量库
│ │
│ ├── memory/ # 记忆模块
│ │ ├── short_term.py
│ │ └── long_term.py
│ │
│ └── guardrails/ # 安全护栏
│ ├── input.py # 输入验证
│ └── output.py # 输出验证

├── tests/ # 测试
│ ├── test_tools.py
│ ├── test_agents.py
│ └── test_guardrails.py

└── scripts/ # 运维脚本
├── ingest.py # 文档导入
└── eval.py # 评估运行

分层原则

  • 入口文件只负责启动,不包含业务逻辑
  • 领域逻辑不依赖框架,保持可测试性
  • 智能体图只定义状态和流程,不实现具体功能
  • 工具封装是 LLM 与外部世界的接口层

4. 依赖管理与配置

# pyproject.toml
[project]
name = "my-agent"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
"langgraph>=0.2.0",
"langchain>=0.3.0",
"pydantic>=2.0",
"pydantic-settings>=2.0",
"fastapi>=0.115.0",
"uvicorn[standard]>=0.30.0",
"aiohttp>=3.10.0",
"structlog>=24.0",
]

[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.24",
"ruff>=0.6.0",
]
# src/my_agent/config.py
from pydantic_settings import BaseSettings
from pydantic import Field

class Settings(BaseSettings):
"""应用配置——所有敏感信息通过环境变量注入。"""

# LLM 配置
openai_api_key: str = Field(..., description="OpenAI API Key")
openai_model: str = Field(default="gpt-4o", description="使用的模型")

# 数据库配置
database_url: str = Field(..., description="PostgreSQL 连接字符串")

# RAG 配置
vector_store_type: str = Field(default="chromadb", description="向量库类型")
embedding_model: str = Field(default="text-embedding-3-small")

# 服务配置
host: str = Field(default="0.0.0.0")
port: int = Field(default=8000)

# 安全配置
max_query_length: int = Field(default=2000, description="最大查询长度")
max_iterations: int = Field(default=20, description="最大迭代次数")

model_config = {
"env_file": ".env",
"env_file_encoding": "utf-8",
"extra": "forbid", # 拒绝未知配置——防止拼写错误
}

settings = Settings()

工程视角

Python 版本选择

版本推荐度原因
3.12+★★★★★uv 原生支持、类型注解增强、性能优化
3.11★★★★兼容性好、性能提升 10-60%
3.10★★★最低要求,`X
3.9 及以下不推荐,缺少现代类型语法

依赖管理工具对比

工具速度锁定兼容性推荐场景
uv★★★★★Python 3.8+新项目首选,速度比 pip/poetry 快 10-100x
poetry★★★Python 3.8+已有项目继续使用
pip + pip-tools★★全兼容简单场景
conda★★全兼容需要非 Python 依赖(CUDA、C 库)

常见反模式

反模式症状修复
意大利面条代码所有逻辑在 main.py,超过 500 行按 domain/tools/agents 分层
硬编码密钥api_key = "sk-xxx" 直接写在代码里.env + Pydantic Settings
同步阻塞多个 API 串行调用,总耗时 = 各耗时之和改用 asyncio.gather 并行
无类型注解函数签名 def process(data):所有函数加类型注解
裸 exceptexcept: pass 吞掉所有异常捕获具体异常类型 + 日志

面试视角

Q: 为什么 Agent 项目首选 Python?

满分回答框架

  • Agent 本质是胶水 + 编排——连接 LLM、工具、知识库
  • Python 的 AI 生态最完善(LangChain、Pydantic、FastAPI、aiohttp)
  • 原型速度远超其他语言(写一个 LLM 调用 vs Java 的 boilerplate)
  • 生产部署不输其他语言(FastAPI 性能接近 Go,Docker 容器化成熟)
  • 对于高并发、低延迟场景(如实时推理服务),Go/Rust 更合适

Q: 异步编程中如何避免阻塞事件循环?

满分回答框架

  • 网络 I/O:直接用 async/await(aiohttp、asyncpg)
  • CPU 密集型:asyncio.to_thread() 或 Celery 后台队列
  • 第三方同步库:asyncio.to_thread() 包装
  • 避免在 async 函数中使用 time.sleep()(用 asyncio.sleep()

实战环节:搭建 Agent 项目骨架

目标

完成一个结构清晰、可运行的 Python Agent 项目骨架,包含完整的依赖管理、配置、异步工具和测试。

环境要求

  • Python 3.12+
  • uv 包管理器(curl -LsSf https://astral.sh/uv/install.sh | sh

步骤

1. 初始化项目

uv init my-agent
cd my-agent
uv add "langgraph>=0.2.0" "pydantic>=2.0" "pydantic-settings>=2.0" \
"fastapi>=0.115.0" "uvicorn[standard]>=0.30.0" "aiohttp>=3.10.0" \
"structlog>=24.0"
uv add --dev pytest pytest-asyncio ruff

2. 创建目录结构

mkdir -p src/my_agent/{domain,agents,tools,rag,memory,guardrails} tests scripts
touch src/my_agent/__init__.py

3. 编写配置模块

创建 src/my_agent/config.py(参考上文 Settings 类),创建 .env.example

OPENAI_API_KEY=sk-xxx
DATABASE_URL=postgresql://localhost:5432/agent_db

4. 编写第一个异步工具

创建 src/my_agent/tools/search.py

import aiohttp
from pydantic import BaseModel, Field

class WeatherResult(BaseModel):
city: str
temperature: float
description: str

async def search_weather(city: str) -> WeatherResult:
"""搜索指定城市的天气——异步工具示例。"""
# 模拟 API 调用(实际应调用 OpenWeatherMap 等)
import asyncio
await asyncio.sleep(0.1) # 模拟网络延迟
return WeatherResult(city=city, temperature=25.0, description="晴天")

5. 编写测试

创建 tests/test_tools.py

import pytest
from src.my_agent.tools.search import search_weather

@pytest.mark.asyncio
async def test_search_weather():
result = await search_weather("北京")
assert result.city == "北京"
assert result.temperature > 0

6. 运行测试

uv run pytest tests/ -v

验证成功

  • uv run pytest tests/ -v 全部通过
  • uv run ruff check src/ 无 lint 错误
  • 项目结构包含 domain/agents/tools/rag/memory/guardrails 目录
  • .env.example 存在,.env.gitignore

思考题

  1. 如果一个工具既有同步版本又有异步版本,Agent 应该如何选择?
  2. Pydantic Settings 的 extra: "forbid" 在生产环境有什么利弊?
  3. 如何设计一个通用的重试装饰器,适用于所有异步工具?

上一阶段:无(第一阶段) | 下一阶段:LLM 基础 →