当前位置：首页 > news >正文

langgraph笔记（2） fastapi笔记

news 2026/8/15 13:25:15

5、子图（Subgraphs）

5.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 # 子图内部的分析结果（私有）

多智能体和A2A

1.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笔记部分。

fastapi

with open as这个语法还是不熟

import shutil from fastapi import FastAPI, UploadFile app = FastAPI() @app.post("/uploadfile/") async def create_upload_file(file: UploadFile): # 将上传的文件保存到指定路径 with open(f"uploads/{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(b"hello")
这样文件才会有 hello。

Annotated：

# Pydantic 里 from pydantic import BaseModel, Field from typing import Annotated class User(BaseModel): age: Annotated[int, Field(gt=0, lt=120)] # 第二个参数是 Field（规则） # FastAPI 里 from fastapi import Query from typing import Annotated def get_items(page: Annotated[int, Query(ge=1)]): # 第二个参数是 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"], deprecated="auto") # ===== OAuth2 方案 ===== # 告诉 FastAPI 从 Authorization: Bearer <token> 请求头中获取令牌： oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # tokenUrl="token" 的意思是告诉 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": "alice@example.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(minutes=15)) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]): """从令牌中获取当前用户（依赖函数）""" credentials_exception = HTTPException( status_code=status.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_code=400, 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_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user.username}, expires_delta=access_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}]

查看全文

http://www.cnnetsun.cn/news/4040780.html