5、子图Subgraphs5.1 定义当图开始变复杂时最自然的问题就是能不能把一整张图当成另一张图里的一个节点来复用这正是子图要解决的问题。LangGraph 里的子图可以理解成把一个已经编译好的图嵌入到另一张更大的父图里。所以子图的价值在于复杂流程拆分、模块化复用、父子流程解耦。5.2 为什么需要子图当流程越来越长时如果所有节点都堆在一张图里会出现几个问题图结构越来越难读某个局部流程没法单独测试相似流程难复用不同业务子模块之间耦合越来越重。这时子图就很像“工作流层面的函数抽取”。你可以把它和普通函数封装做一个类比普通函数把一段 Python 逻辑封起来复用子图把一段 LangGraph 工作流封起来复用5.3 三种模式子图最容易从这三种模式入手理解最简单模式把编译后的子图直接当成父图里的节点共享字段模式父图和子图共享部分状态字段状态转换模式父图状态和子图状态结构不同需要代理节点做转换这三种模式正好也是本章三个子图案例的递进顺序。5.4 案例子图作为节点这是最基础的子图案例。它的重点非常单纯子图也可以像普通节点一样被挂进父图当父图执行到这个“节点”时其实就是在执行一整张子图 【案例】子图作为节点将 compile 后的子图直接 add_node 进父图父子共用同一 State 类型时由 Reducer 合并 messages。 对应教程章节第 25 章 - LangGraph 高级特性 → 4、子图Subgraphs 知识点速览 - 这是子图最基础的入门案例重点先理解“编译后的图也可以像节点一样被父图注册”。 - 父子状态结构相同、且 messages 使用 add列表拼接时本例会出现重复前缀正好用来观察“父图和子图各自合并一次”带来的效果。 - 这个案例不是在教“最佳消息合并策略”而是在帮你建立对子图调用链和状态合并路径的第一直觉。 from operator import add from typing import Annotated, TypedDict from langgraph.constants import END from langgraph.graph import StateGraph, START class DiliState(TypedDict): 状态messages 使用 operator.add 合并策略——新返回的列表与原有列表拼接非覆盖。 messages: Annotated[list[str], add] def sub_node(state: DiliState) - DiliState: return {messages: [response from subgraph]} # --- 子图 --- subgraph_builder StateGraph(DiliState) subgraph_builder.add_node(sub_node, sub_node) subgraph_builder.add_edge(START, sub_node) subgraph_builder.add_edge(sub_node, END) subgraph subgraph_builder.compile() # --- 父图节点即子图 --- builder StateGraph(DiliState) builder.add_node(subgraph_node, subgraph) builder.add_edge(START, subgraph_node) builder.add_edge(subgraph_node, END) graph builder.compile() 子图调用的状态传递逻辑当主图调用子图节点时整个过程会触发两次状态合并 第一步主图把初始状态 {messages: [main-graph]} 传递给子图 第二步子图内部执行 sub_node返回 {messages: [response from subgraph]} 由于 add 策略子图会把传入的 [main-graph] 和返回的 [response from subgraph] 拼接 得到 [main-graph, response from subgraph] 第三步子图执行完成后主图会再次应用 add 策略 把主图原有的 [main-graph] 和子图返回的 [main-graph, response from subgraph] 拼接 最终得到 [main-graph, main-graph, response from subgraph] print(graph.invoke({messages: [main-graph]})) print() # 预期形态示例{messages: [main-graph, main-graph, response from subgraph]} print(subgraph.get_graph().draw_mermaid()) print( * 50) print() 【输出示例】 {messages: [main-graph, main-graph, response from subgraph]} --- config: flowchart: curve: linear --- graph TD; __start__([p__start__/p]):::first sub_node(sub_node) __end__([p__end__/p]):::last __start__ -- sub_node; sub_node -- __end__; classDef default fill:#f2f0ff,line-height:1.2 classDef first fill-opacity:0 classDef last fill:#bfb6fc 同名的state的key这个是主图和子图可以共享和相互传递的。# 父图管理整体对话class ParentState(TypedDict):messages: list # 共享对话历史user_id: str # 共享用户信息# 子图只处理特定功能class SubgraphState(TypedDict):messages: list # 共享同一个对话历史同名tool_results: list # 子图内部的工具调用结果私有analysis: str # 子图内部的分析结果私有多智能体和A2A1.2 多智能体定义多智能体不是“多开几个模型调用”这么简单。它指的是把复杂任务拆给多个专精的 Agent让它们分工、路由、协作再共同完成整体任务。和单智能体相比多智能体的核心变化不是“数量变多”而是角色开始分工、上下文开始隔离、控制流开始显式编排。举个最直白的例子单智能体一个 Agent 同时负责查航班、订酒店、回答用户、决定流程多智能体一个主管 Agent 负责调度航班 Agent 只管航班酒店 Agent 只管酒店所以多智能体更适合的不是“任务听起来高级”而是这些场景工具太多一个 Agent 已经选不过来领域太多单个 Agent 上下文太臃肿任务天然可以拆成多个角色希望不同团队各自维护不同能力模块1.3 不必默认上多智能体LangChain 官方多智能体文档也强调不是每个复杂任务都必须上多智能体。很多时候开发者说自己要“multi-agent”实际想要的是下面几类能力更好的上下文管理更清晰的模块边界更高效的并行化更稳定的任务分工但如果任务本身很简单一个单智能体加上合适的工具、提示词和工作流往往就已经够了。本章的判断标准很简单多智能体不是默认更高级而是在单智能体已经开始吃力时才值得引入。1.4 A2A 协议定义A2A 的全称是Agent-to-Agent。它是一种面向 Agent 系统互操作的开放协议目标是让不同 Agent 能以更标准化的方式发现彼此、发送任务、交换消息、返回结果。换句话说A2A 关心的是“Agent 和 Agent 怎么协作”。A2A 里面几个很核心的概念包括Agent Card相当于 Agent 的“名片 / 能力说明”Task一项被发给远程 Agent 的任务Message围绕任务交换的消息Artifact任务过程或结果产出的内容先发现 Agent调用方先读取Agent Card确认对方会什么、支持什么输入输出。再提交 Task把任务目标、上下文消息、必要参数发给远程 Agent。过程中跟状态长任务通常不是一次就结束调用方会通过轮询、流式更新或通知拿到任务进度。最后取结果读取最终Message/Artifact把它当成另一套 Agent 的产出继续接到自己的系统里。A2A 主要回答这几个问题我怎么知道远程有个什么 Agent它会什么我怎么把任务交给它它怎么把中间消息和结果回给我langgraph初步结束接下来进入python的框架fastapi笔记部分。fastapiwith open as这个语法还是不熟import shutil from fastapi import FastAPI, UploadFile app FastAPI() app.post(/uploadfile/) async def create_upload_file(file: UploadFile): # 将上传的文件保存到指定路径 with open(fuploads/{file.filename}, wb) as buffer: shutil.copyfileobj(file.file, buffer) return {filename: file.filename, message: 文件上传成功}我的疑问点你打开uploads/{file.filename}路径下的文件作为buffer那么buffer现在不就是你打开的这个文件了嘛?答也不是一般with open() as xx也就是打开括号里的文件作为xx但是实际每次还要在后面写具体的操作不然xx其实只是个空文件。说白了就是with open as xx只负责打开文件此时的xx其实还是空文件下面代码才是你具体的操作比如把这个文件的内容写入xx这样xx才是真的有了内容这样能增加灵活性详细解释with open(uploads/a.txt, wb) as buffer:这句的含义是打开 uploads/a.txt如果文件不存在创建它用 wb 模式打开允许以二进制方式写入且原文件内容会被清空把这个文件对象/操作入口命名为 buffer。但它不会自动把任何内容写进去。所以刚打开时buffer 对应的文件通常是空的。必须在 with 块中明确操作它例如with open(uploads/a.txt, wb) as buffer:buffer.write(bhello)这样文件才会有 hello。Annotated# Pydantic 里 from pydantic import BaseModel, Field from typing import Annotated class User(BaseModel): age: Annotated[int, Field(gt0, lt120)] # 第二个参数是 Field规则 # FastAPI 里 from fastapi import Query from typing import Annotated def get_items(page: Annotated[int, Query(ge1)]): # 第二个参数是 Query规则 pass安全认证下面介绍下fastapi的安全认证的例子有详细注释from datetime import datetime, timedelta from typing import Annotated from fastapi import Depends, FastAPI, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jose import JWTError, jwt from passlib.context import CryptContext from pydantic import BaseModel # 配置 SECRET_KEY your-secret-key-keep-it-secret # 生产环境使用环境变量 ALGORITHM HS256 ACCESS_TOKEN_EXPIRE_MINUTES 30 # 密码哈希这行是在创建一个密码哈希工具配置 # 后面用于两件事 # pwd_context.hash(secret) # 明文密码 → bcrypt 哈希 # pwd_context.verify(secret, hashed) # 校验明文密码是否匹配哈希 pwd_context CryptContext(schemes[bcrypt], deprecatedauto) # OAuth2 方案 # 告诉 FastAPI 从 Authorization: Bearer token 请求头中获取令牌 oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) # tokenUrltoken 的意思是告诉 FastAPI #客户端想获取 token 时请去调用 /token 这个接口。/token对应后面的login函数 # 数据模型 class Token(BaseModel): access_token: str token_type: str class TokenData(BaseModel): username: str | None None class User(BaseModel): username: str email: str | None None full_name: str | None None disabled: bool | None None class UserInDB(User): hashed_password: str # 模拟数据库 fake_users_db { alice: { username: alice, full_name: Alice Wonderson, email: aliceexample.com, hashed_password: pwd_context.hash(secret), # 密码: secret disabled: False, } } # 工具函数 def verify_password(plain_password: str, hashed_password: str) - bool: 验证密码 return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) - str: 生成密码哈希 return pwd_context.hash(password) def get_user(db: dict, username: str) - UserInDB | None: 从数据库获取用户 if username in db: return UserInDB(**db[username]) return None def authenticate_user(db: dict, username: str, password: str): 验证用户凭据 user get_user(db, username) if not user: return False if not verify_password(password, user.hashed_password): return False return user def create_access_token(data: dict, expires_delta: timedelta | None None): 创建 JWT 访问令牌 to_encode data.copy() expire datetime.utcnow() (expires_delta or timedelta(minutes15)) to_encode.update({exp: expire}) return jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]): 从令牌中获取当前用户依赖函数 credentials_exception HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail无法验证凭据, headers{WWW-Authenticate: Bearer}, ) try: payload jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) username: str payload.get(sub) if username is None: raise credentials_exception except JWTError: raise credentials_exception user get_user(fake_users_db, username) if user is None: raise credentials_exception return user async def get_current_active_user( current_user: Annotated[User, Depends(get_current_user)], ): 获取当前活跃用户 if current_user.disabled: raise HTTPException(status_code400, detail用户已被禁用) return current_user # 路由 app FastAPI() # Depends() 通常会传一个依赖函数例如async def endpoint(db Depends(get_db)): # Depends() 里没有明确写依赖函数是因为 FastAPI 会根据类型注解OAuth2PasswordRequestForm # 自动把这个类本身当成依赖项来处理。可以近似理解为 # form_data: Annotated[ # OAuth2PasswordRequestForm, # Depends(OAuth2PasswordRequestForm), # ] app.post(/token) async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]): 登录获取令牌 user authenticate_user(fake_users_db, form_data.username, form_data.password) if not user: raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail用户名或密码错误, headers{WWW-Authenticate: Bearer}, ) access_token_expires timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES) access_token create_access_token( data{sub: user.username}, expires_deltaaccess_token_expires ) return {access_token: access_token, token_type: bearer} app.get(/users/me) async def read_users_me( current_user: Annotated[User, Depends(get_current_active_user)], ): 获取当前用户信息需要认证 return current_user app.get(/users/me/items) async def read_own_items( current_user: Annotated[User, Depends(get_current_active_user)], ): 获取当前用户的条目需要认证 return [{item_id: Foo, owner: current_user.username}] langgraph笔记（2） fastapi笔记
5、子图Subgraphs5.1 定义当图开始变复杂时最自然的问题就是能不能把一整张图当成另一张图里的一个节点来复用这正是子图要解决的问题。LangGraph 里的子图可以理解成把一个已经编译好的图嵌入到另一张更大的父图里。所以子图的价值在于复杂流程拆分、模块化复用、父子流程解耦。5.2 为什么需要子图当流程越来越长时如果所有节点都堆在一张图里会出现几个问题图结构越来越难读某个局部流程没法单独测试相似流程难复用不同业务子模块之间耦合越来越重。这时子图就很像“工作流层面的函数抽取”。你可以把它和普通函数封装做一个类比普通函数把一段 Python 逻辑封起来复用子图把一段 LangGraph 工作流封起来复用5.3 三种模式子图最容易从这三种模式入手理解最简单模式把编译后的子图直接当成父图里的节点共享字段模式父图和子图共享部分状态字段状态转换模式父图状态和子图状态结构不同需要代理节点做转换这三种模式正好也是本章三个子图案例的递进顺序。5.4 案例子图作为节点这是最基础的子图案例。它的重点非常单纯子图也可以像普通节点一样被挂进父图当父图执行到这个“节点”时其实就是在执行一整张子图 【案例】子图作为节点将 compile 后的子图直接 add_node 进父图父子共用同一 State 类型时由 Reducer 合并 messages。 对应教程章节第 25 章 - LangGraph 高级特性 → 4、子图Subgraphs 知识点速览 - 这是子图最基础的入门案例重点先理解“编译后的图也可以像节点一样被父图注册”。 - 父子状态结构相同、且 messages 使用 add列表拼接时本例会出现重复前缀正好用来观察“父图和子图各自合并一次”带来的效果。 - 这个案例不是在教“最佳消息合并策略”而是在帮你建立对子图调用链和状态合并路径的第一直觉。 from operator import add from typing import Annotated, TypedDict from langgraph.constants import END from langgraph.graph import StateGraph, START class DiliState(TypedDict): 状态messages 使用 operator.add 合并策略——新返回的列表与原有列表拼接非覆盖。 messages: Annotated[list[str], add] def sub_node(state: DiliState) - DiliState: return {messages: [response from subgraph]} # --- 子图 --- subgraph_builder StateGraph(DiliState) subgraph_builder.add_node(sub_node, sub_node) subgraph_builder.add_edge(START, sub_node) subgraph_builder.add_edge(sub_node, END) subgraph subgraph_builder.compile() # --- 父图节点即子图 --- builder StateGraph(DiliState) builder.add_node(subgraph_node, subgraph) builder.add_edge(START, subgraph_node) builder.add_edge(subgraph_node, END) graph builder.compile() 子图调用的状态传递逻辑当主图调用子图节点时整个过程会触发两次状态合并 第一步主图把初始状态 {messages: [main-graph]} 传递给子图 第二步子图内部执行 sub_node返回 {messages: [response from subgraph]} 由于 add 策略子图会把传入的 [main-graph] 和返回的 [response from subgraph] 拼接 得到 [main-graph, response from subgraph] 第三步子图执行完成后主图会再次应用 add 策略 把主图原有的 [main-graph] 和子图返回的 [main-graph, response from subgraph] 拼接 最终得到 [main-graph, main-graph, response from subgraph] print(graph.invoke({messages: [main-graph]})) print() # 预期形态示例{messages: [main-graph, main-graph, response from subgraph]} print(subgraph.get_graph().draw_mermaid()) print( * 50) print() 【输出示例】 {messages: [main-graph, main-graph, response from subgraph]} --- config: flowchart: curve: linear --- graph TD; __start__([p__start__/p]):::first sub_node(sub_node) __end__([p__end__/p]):::last __start__ -- sub_node; sub_node -- __end__; classDef default fill:#f2f0ff,line-height:1.2 classDef first fill-opacity:0 classDef last fill:#bfb6fc 同名的state的key这个是主图和子图可以共享和相互传递的。# 父图管理整体对话class ParentState(TypedDict):messages: list # 共享对话历史user_id: str # 共享用户信息# 子图只处理特定功能class SubgraphState(TypedDict):messages: list # 共享同一个对话历史同名tool_results: list # 子图内部的工具调用结果私有analysis: str # 子图内部的分析结果私有多智能体和A2A1.2 多智能体定义多智能体不是“多开几个模型调用”这么简单。它指的是把复杂任务拆给多个专精的 Agent让它们分工、路由、协作再共同完成整体任务。和单智能体相比多智能体的核心变化不是“数量变多”而是角色开始分工、上下文开始隔离、控制流开始显式编排。举个最直白的例子单智能体一个 Agent 同时负责查航班、订酒店、回答用户、决定流程多智能体一个主管 Agent 负责调度航班 Agent 只管航班酒店 Agent 只管酒店所以多智能体更适合的不是“任务听起来高级”而是这些场景工具太多一个 Agent 已经选不过来领域太多单个 Agent 上下文太臃肿任务天然可以拆成多个角色希望不同团队各自维护不同能力模块1.3 不必默认上多智能体LangChain 官方多智能体文档也强调不是每个复杂任务都必须上多智能体。很多时候开发者说自己要“multi-agent”实际想要的是下面几类能力更好的上下文管理更清晰的模块边界更高效的并行化更稳定的任务分工但如果任务本身很简单一个单智能体加上合适的工具、提示词和工作流往往就已经够了。本章的判断标准很简单多智能体不是默认更高级而是在单智能体已经开始吃力时才值得引入。1.4 A2A 协议定义A2A 的全称是Agent-to-Agent。它是一种面向 Agent 系统互操作的开放协议目标是让不同 Agent 能以更标准化的方式发现彼此、发送任务、交换消息、返回结果。换句话说A2A 关心的是“Agent 和 Agent 怎么协作”。A2A 里面几个很核心的概念包括Agent Card相当于 Agent 的“名片 / 能力说明”Task一项被发给远程 Agent 的任务Message围绕任务交换的消息Artifact任务过程或结果产出的内容先发现 Agent调用方先读取Agent Card确认对方会什么、支持什么输入输出。再提交 Task把任务目标、上下文消息、必要参数发给远程 Agent。过程中跟状态长任务通常不是一次就结束调用方会通过轮询、流式更新或通知拿到任务进度。最后取结果读取最终Message/Artifact把它当成另一套 Agent 的产出继续接到自己的系统里。A2A 主要回答这几个问题我怎么知道远程有个什么 Agent它会什么我怎么把任务交给它它怎么把中间消息和结果回给我langgraph初步结束接下来进入python的框架fastapi笔记部分。fastapiwith open as这个语法还是不熟import shutil from fastapi import FastAPI, UploadFile app FastAPI() app.post(/uploadfile/) async def create_upload_file(file: UploadFile): # 将上传的文件保存到指定路径 with open(fuploads/{file.filename}, wb) as buffer: shutil.copyfileobj(file.file, buffer) return {filename: file.filename, message: 文件上传成功}我的疑问点你打开uploads/{file.filename}路径下的文件作为buffer那么buffer现在不就是你打开的这个文件了嘛?答也不是一般with open() as xx也就是打开括号里的文件作为xx但是实际每次还要在后面写具体的操作不然xx其实只是个空文件。说白了就是with open as xx只负责打开文件此时的xx其实还是空文件下面代码才是你具体的操作比如把这个文件的内容写入xx这样xx才是真的有了内容这样能增加灵活性详细解释with open(uploads/a.txt, wb) as buffer:这句的含义是打开 uploads/a.txt如果文件不存在创建它用 wb 模式打开允许以二进制方式写入且原文件内容会被清空把这个文件对象/操作入口命名为 buffer。但它不会自动把任何内容写进去。所以刚打开时buffer 对应的文件通常是空的。必须在 with 块中明确操作它例如with open(uploads/a.txt, wb) as buffer:buffer.write(bhello)这样文件才会有 hello。Annotated# Pydantic 里 from pydantic import BaseModel, Field from typing import Annotated class User(BaseModel): age: Annotated[int, Field(gt0, lt120)] # 第二个参数是 Field规则 # FastAPI 里 from fastapi import Query from typing import Annotated def get_items(page: Annotated[int, Query(ge1)]): # 第二个参数是 Query规则 pass安全认证下面介绍下fastapi的安全认证的例子有详细注释from datetime import datetime, timedelta from typing import Annotated from fastapi import Depends, FastAPI, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jose import JWTError, jwt from passlib.context import CryptContext from pydantic import BaseModel # 配置 SECRET_KEY your-secret-key-keep-it-secret # 生产环境使用环境变量 ALGORITHM HS256 ACCESS_TOKEN_EXPIRE_MINUTES 30 # 密码哈希这行是在创建一个密码哈希工具配置 # 后面用于两件事 # pwd_context.hash(secret) # 明文密码 → bcrypt 哈希 # pwd_context.verify(secret, hashed) # 校验明文密码是否匹配哈希 pwd_context CryptContext(schemes[bcrypt], deprecatedauto) # OAuth2 方案 # 告诉 FastAPI 从 Authorization: Bearer token 请求头中获取令牌 oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) # tokenUrltoken 的意思是告诉 FastAPI #客户端想获取 token 时请去调用 /token 这个接口。/token对应后面的login函数 # 数据模型 class Token(BaseModel): access_token: str token_type: str class TokenData(BaseModel): username: str | None None class User(BaseModel): username: str email: str | None None full_name: str | None None disabled: bool | None None class UserInDB(User): hashed_password: str # 模拟数据库 fake_users_db { alice: { username: alice, full_name: Alice Wonderson, email: aliceexample.com, hashed_password: pwd_context.hash(secret), # 密码: secret disabled: False, } } # 工具函数 def verify_password(plain_password: str, hashed_password: str) - bool: 验证密码 return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) - str: 生成密码哈希 return pwd_context.hash(password) def get_user(db: dict, username: str) - UserInDB | None: 从数据库获取用户 if username in db: return UserInDB(**db[username]) return None def authenticate_user(db: dict, username: str, password: str): 验证用户凭据 user get_user(db, username) if not user: return False if not verify_password(password, user.hashed_password): return False return user def create_access_token(data: dict, expires_delta: timedelta | None None): 创建 JWT 访问令牌 to_encode data.copy() expire datetime.utcnow() (expires_delta or timedelta(minutes15)) to_encode.update({exp: expire}) return jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]): 从令牌中获取当前用户依赖函数 credentials_exception HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail无法验证凭据, headers{WWW-Authenticate: Bearer}, ) try: payload jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) username: str payload.get(sub) if username is None: raise credentials_exception except JWTError: raise credentials_exception user get_user(fake_users_db, username) if user is None: raise credentials_exception return user async def get_current_active_user( current_user: Annotated[User, Depends(get_current_user)], ): 获取当前活跃用户 if current_user.disabled: raise HTTPException(status_code400, detail用户已被禁用) return current_user # 路由 app FastAPI() # Depends() 通常会传一个依赖函数例如async def endpoint(db Depends(get_db)): # Depends() 里没有明确写依赖函数是因为 FastAPI 会根据类型注解OAuth2PasswordRequestForm # 自动把这个类本身当成依赖项来处理。可以近似理解为 # form_data: Annotated[ # OAuth2PasswordRequestForm, # Depends(OAuth2PasswordRequestForm), # ] app.post(/token) async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]): 登录获取令牌 user authenticate_user(fake_users_db, form_data.username, form_data.password) if not user: raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail用户名或密码错误, headers{WWW-Authenticate: Bearer}, ) access_token_expires timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES) access_token create_access_token( data{sub: user.username}, expires_deltaaccess_token_expires ) return {access_token: access_token, token_type: bearer} app.get(/users/me) async def read_users_me( current_user: Annotated[User, Depends(get_current_active_user)], ): 获取当前用户信息需要认证 return current_user app.get(/users/me/items) async def read_own_items( current_user: Annotated[User, Depends(get_current_active_user)], ): 获取当前用户的条目需要认证 return [{item_id: Foo, owner: current_user.username}]