尧图网页设计 
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 异步到流式通信为什么需要这些预备知识一、Python 异步编程让 Agent 同时做多件事1.1 一句话理解同步 vs 异步1.2 核心概念三件套1.3 在 Agent 中的高频用法1.4 常见坑速查二、Python 类型系统让 IDE 帮你写代码2.1 为什么 Agent 开发特别需要类型2.2 TypedDict vs Pydantic两大 State 定义方式2.3 Pydantic 不止做 State——工具定义 结构化输出三、环境工程别把 API Key 提交到 GitHub3.1 包管理为什么选 uv3.2 API Key 管理的铁律3.3 强烈建议从第一天起就配好 LangSmith四、HTTP 与流式基础LLM 通信的底层真相4.1 所有 LLM API 本质上都是一个 HTTP POST4.2 流式 vs 非流式用户体验的分水岭4.3 为什么 LLM 流式用 SSE 而不 WebSocket4.4 Agent 的流式本质是决策流4.5 错误处理429 和 500 要区别对待五、四块基石如何拼成完整地图总结为什么需要这些预备知识当我们用 LangChain/LangGraph 构建 Agent 时代码长这样fromtypingimportAnnotated,TypedDictimportasyncio,operatorclassAgentState(TypedDict):messages:Annotated[list[str],operator.add]asyncforeventinapp.astream_events(# ← 异步 流式{messages:[{role:user,content:帮我查天气}]},versionv2):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 核心概念三件套概念一句话解释对应 Python 语法协程用async def定义的可暂停函数async def foo(): ...事件循环调度中心不停轮询谁可以继续往下走了asyncio.run(main())Task协程的包装提交给事件循环后可并发调度asyncio.create_task()最重要的直觉await不是死等而是让出控制权——告诉事件循环“这件事需要点时间你先去处理别的好了叫我。”1.3 在 Agent 中的高频用法# 1. 并发调用多个 LLMresultsawaitasyncio.gather(call_gpt4(prompt),call_claude(prompt),call_gemini(prompt),return_exceptionsTrue# 某个挂了不影响其他)# 2. 超时降级try:resultawaitasyncio.wait_for(call_llm(prompt),timeout2.0)exceptasyncio.TimeoutError:result使用缓存结果...# 3. 流式消费LangGraph 核心asyncforchunkinllm.astream(prompt):print(chunk.content,end,flushTrue)1.4 常见坑速查❌ 错误✅ 正确协程里用time.sleep()协程里用await asyncio.sleep()协程里用requests.get()协程里用httpx.AsyncClientawait写在普通函数里普通函数调用协程用asyncio.run()忘写await开启 mypy 类型检查二、Python 类型系统让 IDE 帮你写代码2.1 为什么 Agent 开发特别需要类型Agent 的State是贯穿所有节点的灵魂结构。没有类型defagent_node(state):# state 里有什么字段全是猜msgstate[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:intPydantic v2——重量级带验证frompydanticimportBaseModel,FieldclassAgentState(BaseModel):messages:list[str]Field(default_factorylist)user_id:strtemperature:floatField(default0.7,ge0.0,le2.0)classConfig:extraforbid# 严格模式选型建议简单场景用 TypedDict需要复杂的验证如工具输入参数定义用 Pydantic。2.3 Pydantic 不止做 State——工具定义 结构化输出# 1. 定义工具输入 SchemaclassSearchInput(BaseModel):query:strField(description搜索关键词)max_results:intField(default5,ge1,le20)# 2. 定义 LLM 的结构化输出classSentimentResult(BaseModel):sentiment:Literal[positive,negative,neutral]confidence:floatField(ge0.0,le1.0)keywords:list[str]# chain prompt | llm.with_structured_output(SentimentResult)三、环境工程别把 API Key 提交到 GitHub3.1 包管理为什么选 uv工具速度推荐度理由pip慢⭐⭐简单适合快速原型poetry中⭐⭐⭐传统项目首选uv极快⭐⭐⭐⭐AI 项目首选Rust 实现# 一行初始化uv init my-agentcdmy-agent uvaddlangchain langchain-openai langgraph uv run python script.py3.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_KEYlsv2_xxxLANGSMITH_PROJECTmy-agentLANGSMITH_TRACINGtrue配置后每次 LangChain 调用都会自动上报 Trace 到 LangSmith 后台你可以可视化地看到 Agent 的调用链——这对调试 Agent 的决策过程至关重要。四、HTTP 与流式基础LLM 通信的底层真相4.1 所有 LLM API 本质上都是一个 HTTP POST# LangChain 的 invoke() 底层就是这玩意儿responserequests.post(https://api.openai.com/v1/chat/completions,headers{Authorization:Bearer sk-xxx},json{model:gpt-4o,messages:[...],stream:False},)dataresponse.json()print(data[choices][0][message][content])LangChain 做的就是把这些原生 HTTP 调用封装成优雅的llm.invoke(你好)。4.2 流式 vs 非流式用户体验的分水岭非流式 (streamFalse) Client ──→ POST ──→ Server ←── 等待 5 秒盯白屏... ←── ←── 一次性返回完整内容 流式 (streamTrue, SSE) Client ──→ POST (streamTrue) ──→ 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 中捕获这些决策事件asyncforeventinapp.astream_events(input,versionv2):kindevent[event]ifkindon_chat_model_stream:print(event[data][chunk].content,end)# 逐 tokenelifkindon_tool_start:print(f\n 调用工具:{event[name]})# 工具调用elifkindon_tool_end:print(f\n✅ 工具返回: ...)# 工具结果4.5 错误处理429 和 500 要区别对待fromtenacityimportretry,stop_after_attempt,wait_exponentialretry(stopstop_after_attempt(3),waitwait_exponential(multiplier1,min2,max30),# 2s → 4s → 8s)asyncdefcall_llm_with_retry(prompt:str):asyncwithhttpx.AsyncClient(timeout30)asclient:responseawaitclient.post(url,jsonpayload)ifresponse.status_code429:# 速率限制 → 重试raiseelifresponse.status_code500:# 服务器错误 → 重试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 ──→ 转化为决策流事件 │ │ │ └────────────────────────────────────────────────────────┘总结预备知识往往是最容易被跳过的部分——但它决定了你后面是读懂代码还是背住代码。预备知识一句话总结在 Agent 中最直接的体现Python 异步await是让出控制权不是死等ainvoke(),astream_events()类型系统类型让 IDE 帮你看代码而不是你帮 IDE 看代码AgentState,Annotated环境工程API Key 永远不裸奔.env不进 Gitload_dotenv(), LangSmithHTTP/SSELLM 调用的本质是一个 HTTP POST SSE 流streamTrue,astream_events四块基石就位下一站——从零开始构建你的第一个 LangGraph Agent 本文基于AI Agent 学习项目第 0 阶段预备知识整理覆盖 0.1 Python 异步编程、0.2 Python 类型系统、0.3 环境工程、0.4 HTTP 与流式基础 四个章节。* AI Agent 开发预备知识：从 Python 异步到流式通信
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 异步到流式通信为什么需要这些预备知识一、Python 异步编程让 Agent 同时做多件事1.1 一句话理解同步 vs 异步1.2 核心概念三件套1.3 在 Agent 中的高频用法1.4 常见坑速查二、Python 类型系统让 IDE 帮你写代码2.1 为什么 Agent 开发特别需要类型2.2 TypedDict vs Pydantic两大 State 定义方式2.3 Pydantic 不止做 State——工具定义 结构化输出三、环境工程别把 API Key 提交到 GitHub3.1 包管理为什么选 uv3.2 API Key 管理的铁律3.3 强烈建议从第一天起就配好 LangSmith四、HTTP 与流式基础LLM 通信的底层真相4.1 所有 LLM API 本质上都是一个 HTTP POST4.2 流式 vs 非流式用户体验的分水岭4.3 为什么 LLM 流式用 SSE 而不 WebSocket4.4 Agent 的流式本质是决策流4.5 错误处理429 和 500 要区别对待五、四块基石如何拼成完整地图总结为什么需要这些预备知识当我们用 LangChain/LangGraph 构建 Agent 时代码长这样fromtypingimportAnnotated,TypedDictimportasyncio,operatorclassAgentState(TypedDict):messages:Annotated[list[str],operator.add]asyncforeventinapp.astream_events(# ← 异步 流式{messages:[{role:user,content:帮我查天气}]},versionv2):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 核心概念三件套概念一句话解释对应 Python 语法协程用async def定义的可暂停函数async def foo(): ...事件循环调度中心不停轮询谁可以继续往下走了asyncio.run(main())Task协程的包装提交给事件循环后可并发调度asyncio.create_task()最重要的直觉await不是死等而是让出控制权——告诉事件循环“这件事需要点时间你先去处理别的好了叫我。”1.3 在 Agent 中的高频用法# 1. 并发调用多个 LLMresultsawaitasyncio.gather(call_gpt4(prompt),call_claude(prompt),call_gemini(prompt),return_exceptionsTrue# 某个挂了不影响其他)# 2. 超时降级try:resultawaitasyncio.wait_for(call_llm(prompt),timeout2.0)exceptasyncio.TimeoutError:result使用缓存结果...# 3. 流式消费LangGraph 核心asyncforchunkinllm.astream(prompt):print(chunk.content,end,flushTrue)1.4 常见坑速查❌ 错误✅ 正确协程里用time.sleep()协程里用await asyncio.sleep()协程里用requests.get()协程里用httpx.AsyncClientawait写在普通函数里普通函数调用协程用asyncio.run()忘写await开启 mypy 类型检查二、Python 类型系统让 IDE 帮你写代码2.1 为什么 Agent 开发特别需要类型Agent 的State是贯穿所有节点的灵魂结构。没有类型defagent_node(state):# state 里有什么字段全是猜msgstate[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:intPydantic v2——重量级带验证frompydanticimportBaseModel,FieldclassAgentState(BaseModel):messages:list[str]Field(default_factorylist)user_id:strtemperature:floatField(default0.7,ge0.0,le2.0)classConfig:extraforbid# 严格模式选型建议简单场景用 TypedDict需要复杂的验证如工具输入参数定义用 Pydantic。2.3 Pydantic 不止做 State——工具定义 结构化输出# 1. 定义工具输入 SchemaclassSearchInput(BaseModel):query:strField(description搜索关键词)max_results:intField(default5,ge1,le20)# 2. 定义 LLM 的结构化输出classSentimentResult(BaseModel):sentiment:Literal[positive,negative,neutral]confidence:floatField(ge0.0,le1.0)keywords:list[str]# chain prompt | llm.with_structured_output(SentimentResult)三、环境工程别把 API Key 提交到 GitHub3.1 包管理为什么选 uv工具速度推荐度理由pip慢⭐⭐简单适合快速原型poetry中⭐⭐⭐传统项目首选uv极快⭐⭐⭐⭐AI 项目首选Rust 实现# 一行初始化uv init my-agentcdmy-agent uvaddlangchain langchain-openai langgraph uv run python script.py3.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_KEYlsv2_xxxLANGSMITH_PROJECTmy-agentLANGSMITH_TRACINGtrue配置后每次 LangChain 调用都会自动上报 Trace 到 LangSmith 后台你可以可视化地看到 Agent 的调用链——这对调试 Agent 的决策过程至关重要。四、HTTP 与流式基础LLM 通信的底层真相4.1 所有 LLM API 本质上都是一个 HTTP POST# LangChain 的 invoke() 底层就是这玩意儿responserequests.post(https://api.openai.com/v1/chat/completions,headers{Authorization:Bearer sk-xxx},json{model:gpt-4o,messages:[...],stream:False},)dataresponse.json()print(data[choices][0][message][content])LangChain 做的就是把这些原生 HTTP 调用封装成优雅的llm.invoke(你好)。4.2 流式 vs 非流式用户体验的分水岭非流式 (streamFalse) Client ──→ POST ──→ Server ←── 等待 5 秒盯白屏... ←── ←── 一次性返回完整内容 流式 (streamTrue, SSE) Client ──→ POST (streamTrue) ──→ 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 中捕获这些决策事件asyncforeventinapp.astream_events(input,versionv2):kindevent[event]ifkindon_chat_model_stream:print(event[data][chunk].content,end)# 逐 tokenelifkindon_tool_start:print(f\n 调用工具:{event[name]})# 工具调用elifkindon_tool_end:print(f\n✅ 工具返回: ...)# 工具结果4.5 错误处理429 和 500 要区别对待fromtenacityimportretry,stop_after_attempt,wait_exponentialretry(stopstop_after_attempt(3),waitwait_exponential(multiplier1,min2,max30),# 2s → 4s → 8s)asyncdefcall_llm_with_retry(prompt:str):asyncwithhttpx.AsyncClient(timeout30)asclient:responseawaitclient.post(url,jsonpayload)ifresponse.status_code429:# 速率限制 → 重试raiseelifresponse.status_code500:# 服务器错误 → 重试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 ──→ 转化为决策流事件 │ │ │ └────────────────────────────────────────────────────────┘总结预备知识往往是最容易被跳过的部分——但它决定了你后面是读懂代码还是背住代码。预备知识一句话总结在 Agent 中最直接的体现Python 异步await是让出控制权不是死等ainvoke(),astream_events()类型系统类型让 IDE 帮你看代码而不是你帮 IDE 看代码AgentState,Annotated环境工程API Key 永远不裸奔.env不进 Gitload_dotenv(), LangSmithHTTP/SSELLM 调用的本质是一个 HTTP POST SSE 流streamTrue,astream_events四块基石就位下一站——从零开始构建你的第一个 LangGraph Agent 本文基于AI Agent 学习项目第 0 阶段预备知识整理覆盖 0.1 Python 异步编程、0.2 Python 类型系统、0.3 环境工程、0.4 HTTP 与流式基础 四个章节。*