AI Agent 开发预备知识：从 Python 异步到流式通信

哒咩哒咩129

268人浏览 · 2026-08-08 12:11:26

哒咩哒咩129 · 2026-08-08 12:11:26 发布

AI Agent 开发预备知识：从 Python 异步到流式通信

⚠️ 重要：本文包含 Python 异步编程、类型系统与 Pydantic v2、环境工程（uv/pip/API Key 管理）、HTTP 与 SSE 流式协议 等 AI Agent 开发四大预备知识。

文中涉及的相关代码示例地址：https://github.com/m12305/Langchain-LangGraph-agent — Langchain/LangGraph 学习项目
相关项目推荐：https://github.com/m12305/hello-FastAPI — FastAPI 学习项目

在动手写第一个 LangGraph Agent 之前，有四块基石必须先铺好：异步编程、类型系统、环境工程、HTTP 与流式协议。本文将这四大预备知识串联成一幅完整的地图。

文章目录

AI Agent 开发预备知识：从 Python 异步到流式通信

为什么需要这些预备知识？

当我们用 LangChain/LangGraph 构建 Agent 时，代码长这样：

from typing import Annotated, TypedDict
import asyncio, operator

class AgentState(TypedDict):
    messages: Annotated[list[str], operator.add]

async for event in app.astream_events(  # ← 异步 + 流式
    {"messages": [{"role": "user", "content": "帮我查天气"}]},
    version="v2"
):
    if event["event"] == "on_tool_start":
        print(f"🔧 正在调用工具: {event['name']}")

短短几行代码，涉及了 异步编程 (async for)、类型系统 (TypedDict, Annotated)、环境配置 (API Key 管理)、HTTP/SSE (astream_events 底层协议)。如果你的基础不牢，这段代码就是天书。这篇文章帮你逐个击破。

一、Python 异步编程：让 Agent “同时做多件事”

1.1 一句话理解同步 vs 异步

同步：排队办事。冲咖啡要 3 分钟，烤面包要 2 分钟，你不等咖啡冲完就不去烤面包，总共花 5 分钟。
异步：同时开工。咖啡机和烤面包机同时运转，谁先好先处理谁，总共花 3 分钟。

import asyncio

# 同步：总耗时 = 3s + 2s = 5s
def sync_breakfast():
    make_coffee()   # 阻塞 3 秒
    make_toast()    # 阻塞 2 秒

# 异步：总耗时 ≈ max(3s, 2s) = 3s
async def async_breakfast():
    await asyncio.gather(
        make_coffee_async(),  # 同时执行
        make_toast_async(),   # 同时执行
    )

1.2 核心概念三件套

概念	一句话解释	对应 Python 语法
协程	用 `async def` 定义的"可暂停函数"	`async def foo(): ...`
事件循环	调度中心，不停轮询"谁可以继续往下走了"	`asyncio.run(main())`
Task	协程的包装，提交给事件循环后可并发调度	`asyncio.create_task()`

🔑 最重要的直觉：await 不是"死等"，而是"让出控制权"——告诉事件循环：“这件事需要点时间，你先去处理别的，好了叫我。”

1.3 在 Agent 中的高频用法

# 1. 并发调用多个 LLM
results = await asyncio.gather(
    call_gpt4(prompt),
    call_claude(prompt),
    call_gemini(prompt),
    return_exceptions=True  # 某个挂了不影响其他
)

# 2. 超时降级
try:
    result = await asyncio.wait_for(call_llm(prompt), timeout=2.0)
except asyncio.TimeoutError:
    result = "使用缓存结果..."

# 3. 流式消费（LangGraph 核心）
async for chunk in llm.astream(prompt):
    print(chunk.content, end="", flush=True)

1.4 常见坑速查

❌ 错误	✅ 正确
协程里用 `time.sleep()`	协程里用 `await asyncio.sleep()`
协程里用 `requests.get()`	协程里用 `httpx.AsyncClient`
`await` 写在普通函数里	普通函数调用协程用 `asyncio.run()`
忘写 `await`	开启 mypy 类型检查

二、Python 类型系统：让 IDE 帮你写代码

2.1 为什么 Agent 开发特别需要类型？

Agent 的 State 是贯穿所有节点的"灵魂结构"。没有类型：

def agent_node(state):
    # state 里有什么字段？全是猜！
    msg = state["message"]  # 拼写错误，运行时才炸！

有了类型：

class AgentState(TypedDict):
    messages: list[str]     # ← IDE 自动补全
    user_id: str            # ← 拼写错误当场红色波浪线
    turn_count: int         # ← 用错类型当场报错

2.2 TypedDict vs Pydantic：两大 State 定义方式

TypedDict——轻量级，零依赖：

from typing import Annotated, TypedDict
import operator

class AgentState(TypedDict):
    # Annotated[类型, 合并函数] —— LangGraph 的精髓
    messages: Annotated[list[str], operator.add]  # 追加而非覆盖！
    current_tool: str | None  # 没有 Annotated → 直接覆盖
    turn_count: int

Pydantic v2——重量级，带验证：

from pydantic import BaseModel, Field

class AgentState(BaseModel):
    messages: list[str] = Field(default_factory=list)
    user_id: str
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)

    class Config:
        extra = "forbid"  # 严格模式

💡 选型建议：简单场景用 TypedDict，需要复杂的验证（如工具输入参数定义）用 Pydantic。

2.3 Pydantic 不止做 State——工具定义 & 结构化输出

# 1. 定义工具输入 Schema
class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")
    max_results: int = Field(default=5, ge=1, le=20)

# 2. 定义 LLM 的结构化输出
class SentimentResult(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"]
    confidence: float = Field(ge=0.0, le=1.0)
    keywords: list[str]

# chain = prompt | llm.with_structured_output(SentimentResult)

三、环境工程：别把 API Key 提交到 GitHub

3.1 包管理：为什么选 uv？

工具	速度	推荐度	理由
`pip`	慢	⭐⭐	简单，适合快速原型
`poetry`	中	⭐⭐⭐	传统项目首选
`uv`	极快	⭐⭐⭐⭐	AI 项目首选，Rust 实现

# 一行初始化
uv init my-agent && cd my-agent
uv add langchain langchain-openai langgraph
uv run python script.py

3.2 API Key 管理的铁律

❌ 永远不要硬编码 API Key
❌ 永远不要提交 .env 到 Git
✅ 使用 python-dotenv 加载环境变量
✅ 提供 .env.example 作为模板
✅ 企业级方案：Pydantic Settings

# 标准加载方式
from dotenv import load_dotenv
import os

load_dotenv()

# 启动时校验，避免跑到一半才发现 Key 缺失
for key in ["OPENAI_API_KEY", "LANGSMITH_API_KEY"]:
    if not os.getenv(key):
        raise SystemExit(f"缺少环境变量: {key}")

3.3 强烈建议：从第一天起就配好 LangSmith

# .env 中添加
LANGSMITH_API_KEY=lsv2_xxx
LANGSMITH_PROJECT=my-agent
LANGSMITH_TRACING=true

配置后每次 LangChain 调用都会自动上报 Trace 到 LangSmith 后台，你可以可视化地看到 Agent 的调用链——这对调试 Agent 的决策过程至关重要。

四、HTTP 与流式基础：LLM 通信的底层真相

4.1 所有 LLM API 本质上都是一个 HTTP POST

# LangChain 的 invoke() 底层就是这玩意儿
response = requests.post(
    "https://api.openai.com/v1/chat/completions",
    headers={"Authorization": "Bearer sk-xxx"},
    json={"model": "gpt-4o", "messages": [...], "stream": False},
)
data = response.json()
print(data["choices"][0]["message"]["content"])

LangChain 做的，就是把这些原生 HTTP 调用封装成优雅的 llm.invoke("你好")。

4.2 流式 vs 非流式：用户体验的分水岭

非流式 (stream=False)：
Client ──→ POST ──→ Server
       ←── 等待 5 秒，盯白屏... ←──
       ←── 一次性返回完整内容

流式 (stream=True, SSE)：
Client ──→ POST (stream=True) ──→ Server
       ←── data: "床"    ← 0.1s
       ←── data: "前"    ← 0.2s
       ←── data: "明"    ← 0.3s
       ←── data: "月"    ← 0.4s
       ←── data: "光"    ← 0.5s
       ←── data: [DONE]

非流式：用户盯着空白屏幕干等，体验差。流式：0.1 秒就看到第一个字，体感延迟极低。

4.3 为什么 LLM 流式用 SSE 而不 WebSocket？

协议	方向	为什么选/不选
SSE ⭐	服务器→客户端，单向	✅ 方向匹配 LLM 生成模式
WebSocket	双向	❌ 杀鸡用牛刀，LLM 不需要双向
SSE 优势：		普通 HTTP 即可，不需要协议升级

LLM 生成文本是典型的"服务器往客户端单向推数据"场景——SSE 最合适。

4.4 Agent 的流式本质是"决策流"

普通的聊天流式只是逐 token 出字，而 Agent 的流式是"决策的曝光"：

🤔 Agent 正在分析用户请求...
🔧 决定调用工具: search_knowledge_base
   📋 参数: {"query": "退款政策", "top_k": 5}
   ✅ 检索到 5 篇相关文档
💬 正在生成回答: 根据您的订单记录...

# LangGraph 中捕获这些决策事件
async for event in app.astream_events(input, version="v2"):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")   # 逐 token
    elif kind == "on_tool_start":
        print(f"\n🔧 调用工具: {event['name']}")         # 工具调用
    elif kind == "on_tool_end":
        print(f"\n✅ 工具返回: ...")                      # 工具结果

4.5 错误处理：429 和 500 要区别对待

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=30),  # 2s → 4s → 8s
)
async def call_llm_with_retry(prompt: str):
    async with httpx.AsyncClient(timeout=30) as client:
        response = await client.post(url, json=payload)
        if response.status_code == 429:   # 速率限制 → 重试
            raise
        elif response.status_code >= 500: # 服务器错误 → 重试
            raise
        response.raise_for_status()
        return response.json()

五、四块基石如何拼成完整地图？

来一个全景视角——当你写下 async for event in app.astream_events(...) 时，背后发生了什么：

┌────────────────────────────────────────────────────────┐
│              LangGraph Agent 执行全景                  │
├────────────────────────────────────────────────────────┤
│                                                        │
│  环境工程 (.env)                                       │
│  └─→ API Key 管理 ──→ 安全地连上 LLM Provider         │
│                                                        │
│  异步编程 (asyncio)                                    │
│  └─→ astream_events() ──→ 不阻塞地消费事件流          │
│  └─→ asyncio.gather() ──→ 并发调用多个工具            │
│                                                        │
│  类型系统 (TypedDict/Pydantic)                         │
│  └─→ AgentState ──→ 所有节点共享同一份"记忆"          │
│  └─→ Annotated[list, add] ──→ 消息追加而非覆盖        │
│  └─→ Pydantic Model ──→ 工具参数自动校验              │
│                                                        │
│  HTTP/SSE (httpx)                                      │
│  └─→ POST /chat/completions ──→ 每次 LLM 调用         │
│  └─→ SSE stream ──→ 逐 token 接收响应                 │
│  └─→ astream_events ──→ 转化为"决策流"事件            │
│                                                        │
└────────────────────────────────────────────────────────┘

总结

预备知识往往是最容易被跳过的部分——但它决定了你后面是"读懂代码"还是"背住代码"。

预备知识	一句话总结	在 Agent 中最直接的体现
Python 异步	`await` 是让出控制权，不是死等	`ainvoke()`, `astream_events()`
类型系统	类型让 IDE 帮你看代码，而不是你帮 IDE 看代码	`AgentState`, `Annotated`
环境工程	API Key 永远不裸奔，`.env` 不进 Git	`load_dotenv()`, LangSmith
HTTP/SSE	LLM 调用的本质是一个 HTTP POST + SSE 流	`stream=True`, `astream_events`