企业官网建设流程全解析

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 与流式协议。本文将这四大预备知识串联成一幅完整的地图。

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

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

fromtypingimportAnnotated,TypedDictimportasyncio,operatorclassAgentState(TypedDict):messages:Annotated[list[str],operator.add]asyncforeventinapp.astream_events(# ← 异步 + 流式{"messages":[{"role":"user","content":"帮我查天气"}]},version="v2"):ifevent["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 分钟。

importasyncio# 同步：总耗时 = 3s + 2s = 5sdefsync_breakfast():make_coffee()# 阻塞 3 秒make_toast()# 阻塞 2 秒# 异步：总耗时 ≈ max(3s, 2s) = 3sasyncdefasync_breakfast():awaitasyncio.gather(make_coffee_async(),# 同时执行make_toast_async(),# 同时执行)

1.2 核心概念三件套

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

1.3 在 Agent 中的高频用法

# 1. 并发调用多个 LLMresults=awaitasyncio.gather(call_gpt4(prompt),call_claude(prompt),call_gemini(prompt),return_exceptions=True# 某个挂了不影响其他)# 2. 超时降级try:result=awaitasyncio.wait_for(call_llm(prompt),timeout=2.0)exceptasyncio.TimeoutError:result="使用缓存结果..."# 3. 流式消费（LangGraph 核心）asyncforchunkinllm.astream(prompt):print(chunk.content,end="",flush=True)

1.4 常见坑速查

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

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

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

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

有了类型：

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

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

TypedDict——轻量级，零依赖：

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

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

frompydanticimportBaseModel,FieldclassAgentState(BaseModel):messages:list[str]=Field(default_factory=list)user_id:strtemperature:float=Field(default=0.7,ge=0.0,le=2.0)classConfig:extra="forbid"# 严格模式

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

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

# 1. 定义工具输入 SchemaclassSearchInput(BaseModel):query:str=Field(description="搜索关键词")max_results:int=Field(default=5,ge=1,le=20)# 2. 定义 LLM 的结构化输出classSentimentResult(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？

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

3.2 API Key 管理的铁律

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

# 标准加载方式fromdotenvimportload_dotenvimportos load_dotenv()# 启动时校验，避免跑到一半才发现 Key 缺失forkeyin["OPENAI_API_KEY","LANGSMITH_API_KEY"]:ifnotos.getenv(key):raiseSystemExit(f"缺少环境变量:{key}")

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

# .env 中添加LANGSMITH_API_KEY=lsv2_xxxLANGSMITH_PROJECT=my-agentLANGSMITH_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？

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

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

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

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

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

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

fromtenacityimportretry,stop_after_attempt,wait_exponential@retry(stop=stop_after_attempt(3),wait=wait_exponential(multiplier=1,min=2,max=30),# 2s → 4s → 8s)asyncdefcall_llm_with_retry(prompt:str):asyncwithhttpx.AsyncClient(timeout=30)asclient:response=awaitclient.post(url,json=payload)ifresponse.status_code==429:# 速率限制 → 重试raiseelifresponse.status_code>=500:# 服务器错误 → 重试raiseresponse.raise_for_status()returnresponse.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 ──→ 转化为"决策流"事件 │ │ │ └────────────────────────────────────────────────────────┘

总结

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

四块基石就位，下一站——从零开始构建你的第一个 LangGraph Agent 🚀

本文基于"AI Agent 学习项目"第 0 阶段（预备知识）整理，覆盖 0.1 Python 异步编程、0.2 Python 类型系统、0.3 环境工程、0.4 HTTP 与流式基础 四个章节。*