ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

FastAPI+LangGraph实战:构建智能实验室预约系统全复盘

FastAPI+LangGraph实战:构建智能实验室预约系统全复盘 实验室的预约群又炸了。管理员早上刚发了一条“本周三下午可约”不到十分钟群里就刷了上百条消息有人抢到了黄金时段有人对着“已被占用”的红色提示骂骂咧咧还有人直接私聊管理员说要走后门。这种混乱我实在太熟悉了几乎所有靠人工登记的实验室都逃不过这一劫。所以当我决定自己动手做一个智能实验室预约系统时第一反应不是去网上找一个现成的预约插件而是把这个项目当成一次完整的 AI 工程实践来做后端用 FastAPI 搭 RESTful 服务核心调度逻辑用 LangGraph 编排成可感知上下文的 Agent让系统不仅能“预约”还能在时间冲突时主动给出备选方案。这套组合在当前的技术栈里非常能打FastAPI 负责稳定地把 API 暴露给前端LangGraph 负责把大模型能力嵌进业务流。这篇文章就把我从零开始搭这个项目的过程完整复盘一遍包括选型思路、数据库设计、Agent 图谱搭建、并发控制以及那些文档里查不到的坑。1. 项目从哪来一个真实的实验室预约痛点1.1 为什么实验室预约总能吵起来先说背景。我所在的实验室有 6 台设备、3 个独立房间每周开放 40 个小时的预约窗口但实际使用人数是设备位数的 5 倍以上。原先的预约方式是“微信群 共享表格”管理员用 Excel 手动标时间段谁先回复谁占坑结果就是信息滞后、冲突频发、有人临时放了鸽子别人还补不上。表面上看是个管理问题本质上是个“资源调度 信息对称”的技术问题。传统改法是上个预约系统表单填一下、时间选一下、提交入库看起来能解决问题。但用一段时间你会发现三个硬伤第一用户输入不标准“周三下午”、“3点左右”、“后天上午”这种模糊说法普通表单根本无法处理第二时间冲突时系统只会冷冰冰说“不可预约”不会主动给出相邻空闲时段用户得换好几个条件反复试第三管理员想做一些规则调整比如某台设备夜间限制使用、长时间预约必须审批改起来非常痛苦。这些痛点恰好是大模型 Agent 擅长的地方。自然语言理解可以交给 LLM多轮对话状态管理可以用 LangGraph 的图结构来做规则变更就改图节点不用重构整个系统。这也是我在技术选型时坚持引入 LangGraph 的核心原因它不是花架子是真的能把自然语言预约变成现实。1.2 技术选型FastAPI LangGraph 为什么是这个组合先聊后端框架。实验室预约系统本质上是个 CRUD 业务规则的 Web 服务可选方案有 Django、Flask、FastAPI 三个主流。Django 太重自带 Admin 和 ORM 确实方便但异步支持是后补的而且对于一个预约系统来说有点杀鸡用牛刀Flask 轻量灵活但原生不支持异步后续对接流式响应、WebSocket 这类能力时要额外折腾。FastAPI 的好处非常突出原生 async/await、基于 OpenAPI 自动生成接口文档、Pydantic 做参数校验和序列化写起来快跑起来稳。尤其是自动生成的 Swagger 文档在前后端联调时帮了大忙前端同事拿着/docs页面就能自己试接口不用一遍遍来问参数格式。再补一句热词里高频出现的 FastAPI CORS 问题这个后面实操章节会细说。CORS 是前后端分离项目躲不开的一关配置不对你前端 axios 发请求就报跨域错误但这个跟框架本身没关系是浏览器安全策略FastAPI 提供了现成的 CORSMiddleware用对就行。LangGraph 这边我要先泼一盆冷水不是所有项目都需要上 LangGraph。如果你只是想在 FastAPI 里调用一次大模型做关键词抽取用 LangChain 的 chain 或者直接调 SDK 就够了。但预约系统天然是“多轮状态型”场景用户第一句话说“帮我预约明天下午的设备 A”Agent 需要确认用户身份、解析时间意图、检索设备占用情况、判断与规则的冲突、返回可用时段或者发起二次确认。每一步的输入依赖上一步的输出而且不同分支有跳转逻辑这种场景用 LangGraph 的图式编排就是最合适的选择——节点是处理单元边是状态转移条件整个对话流程一目了然出了问题也好定位。下表整理了我在选型时做的对比直接贴出来供参考对比维度FastAPIFlaskDjango异步支持原生 async/await不支持需额外库3.1 逐步完善自动文档OpenAPI Swagger 自动生成需手动配置需第三方库参数校验Pydantic 模型手动校验Serializer 较重LangGraph 集成自然异步节点友好可集成但不顺畅可集成但偏重适合体量中小型服务优选嵌入式/微服务大型单体业务2. 核心设计预约系统怎么才算“智能”2.1 从“查询空闲”到“主动调度”的设计升级做预约系统的第一反应通常是让用户选设备、选日期、选时间段然后后端查一下有没有冲突没有就写入。这个流程能跑通但谈不上智能。我做的设计是这样升级的把用户任意口语化的预约请求丢给 AgentAgent 先做意图理解和槽位抽取然后调后端接口查空闲再把结果整理成自然语言响应。如果请求的时间段已经被占用Agent 不直接拒绝而是查询最近的两个可替代空闲时段返回给用户由用户确认或修改。这样一来用户在小程序或网页里只需要输入“我要约明天下午两点到四点的设备 B”系统就能自动完成一系列动作而不是让用户在三个下拉框里挨个选。真正的“智能”不一定是大模型有多聪明而是把大模型的语义理解能力嵌进业务流程把原来需要人类管理员判断的事情交给程序自动完成。2.2 数据库表结构设计把预约业务吃透讲真话很多 AI 项目的数据库设计都特别敷衍几张表一把梭。但预约系统不像聊天机器人它要保证数据一致性表结构直接决定后面的并发控制和冲突检测能不能写好。我设计了四张核心表users、equipments、bookings、equipment_maintenance。users表负责用户身份和权限普通用户只能预约和取消自己的预约管理员可以调整所有记录和设置规则。equipments表保存设备基础信息包括设备名称、房间号、是否启用、单次最长占用时长、是否需要审批。bookings表是核心字段包括预约人、设备 ID、开始时间、结束时间、状态待确认/已确认/已取消/已完成、备注、创建时间。关键是给(equipment_id, start_time, end_time)加复合索引这是冲突检测效率的基石。equipment_maintenance表用来记录设备维护计划这块很多预约系统都会漏掉但实际使用中维护周期和可预约时间必须联动否则用户约了一个在检修的设备现场体验极差。我在查询环节就把维护时间段过滤掉Agent 返回的空闲时段也不会包含维护时间。下面是核心bookings表的 SQLite 建表语句示例我在项目里实际用的是 SQLAlchemy 2.0 的 ORM 写法这里简化展示便于理解CREATE TABLE bookings ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL REFERENCES users(id), equipment_id INTEGER NOT NULL REFERENCES equipments(id), start_time DATETIME NOT NULL, end_time DATETIME NOT NULL, status VARCHAR(20) DEFAULT confirmed, remark TEXT DEFAULT , created_at DATETIME DEFAULT CURRENT_TIMESTAMP, CHECK (end_time start_time) ); CREATE INDEX idx_bookings_equipment_time ON bookings(equipment_id, start_time, end_time);2.3 权限与规则给 Agent 加上边界意识系统里不能什么东西都让大模型自由发挥。比如夜间预约规则23:00 到次日 07:00 禁止预约全职管理员值班场景除外这个规则如果只是在 Prompt 里写一句“请遵守实验室规定”效果很不稳定模型偶尔会无视。我把规则做成了 FastAPI 里的一个校验函数Agent 在调用预约节点前先把时间参数传给这个校验器返回通过/不通过并附带原因再决定是否走预约写库分支。这样做的好处是规则放在确定的代码里保证 100% 执行Agent 只负责自然语言理解和用户沟通不负责拍板。所谓“智能系统”是规则引擎和大模型的协同而不是把一切交给大模型。3. 实操一把梭FastAPI 后端从零搭起来3.1 用 uv 管理项目和虚拟环境写 Python 项目最烦的是环境管理以前用pip installvenv容易把系统环境搞乱坑踩多了之后我现在全面切到了 uv。uv 是一个用 Rust 写的 Python 包管理器速度比 pip 快一个量级而且可以一条命令创建虚拟环境、安装依赖、锁定版本。用热词里提到的“uv包管理器创建虚拟环境与fastapi”这套流程在 Windows 和 macOS 上都能跑实测非常稳。核心命令如下# 初始化项目 uv init lab-reservation cd lab-reservation # 创建虚拟环境并激活uv 会自动识别 .python-version uv venv --python 3.11 source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate # 安装依赖 uv add fastapi uvicorn[standard] sqlalchemy aiosqlite pydantic-settings langgraph langchain-openai httpx提示安装时不要用pip install -r requirements.txt直接装uv 的项目依赖写在pyproject.toml里版本锁定更可靠。如果你在 PyCharm 里遇到“安装 FastAPI 失败”大概率是虚拟环境解释器和项目解释器没配对用 uv 创建好.venv后手动在 PyCharm 里选该解释器即可解决。3.2 FastAPI 项目骨架与配置拆分项目目录我习惯按“路由-服务-模型”三层分再加一个core放配置和依赖项lab-reservation/ ├── app/ │ ├── api/ │ │ ├── routes/ │ │ │ ├── auth.py │ │ │ ├── equipment.py │ │ │ └── booking.py │ │ └── deps.py │ ├── core/ │ │ ├── config.py │ │ └── database.py │ ├── models/ │ │ ├── user.py │ │ ├── equipment.py │ │ └── booking.py │ ├── schemas/ │ │ ├── booking.py │ │ └── user.py │ ├── services/ │ │ ├── booking_service.py │ │ └── slot_service.py │ └── main.py ├── agent/ │ ├── graph.py │ ├── nodes.py │ └── state.py └── pyproject.tomlFastAPI 初始化读取配置文件这个问题热词里也提到了我给出一套标准解法使用pydantic-settings读取.env文件把数据库连接、密钥、模型 API Key 全部放到环境变量里不写死在代码中。# app/core/config.py from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): app_name: str Lab Reservation API database_url: str sqliteaiosqlite:///./lab_reservation.db openai_api_key: str model_name: str gpt-4o-mini jwt_secret: str change-me-in-prod jwt_expire_minutes: int 60 * 24 model_config SettingsConfigDict(env_file.env, env_file_encodingutf-8) settings Settings() # app/main.py from fastapi import FastAPI from app.core.config import settings from app.api.routes import booking, equipment, auth app FastAPI(titlesettings.app_name) app.include_router(auth.router, prefix/api/auth, tags[auth]) app.include_router(equipment.router, prefix/api/equipments, tags[equipment]) app.include_router(booking.router, prefix/api/bookings, tags[booking])这种写法的好处是不同环境本地、测试、生产只需要换.env文件内容代码不用改一行。3.3 CORS 和中间件前后端联调避坑CORS 是前后端分离项目里最常见的第一个坎。浏览器的同源策略会拦截跨域请求FastAPI 的解决方案是添加 CORSMiddleware。很多新手在这里的困惑是明明后端接口用 curl 调通了前端 axios 一调就报Access-Control-Allow-Origin缺失这不是后端接口有问题是跨域策略没配好。具体配置如下app.add_middleware( CORSMiddleware, allow_origins[origin for origin in settings.cors_origins], allow_credentialsTrue, allow_methods[*], allow_headers[*], )这里有两个经验点。第一allow_origins别图省事写[*]因为如果你同时开了allow_credentialsTrue浏览器规定通配符是不合法的这在涉及登录态Cookie/Session时必炸。第二如果你的前端跑在http://localhost:5173Vite 默认端口后端跑在http://localhost:8000那么allow_origins里必须包含http://localhost:5173这个 Origin否则就会被拦截。前端排查时候打开浏览器 F12看网络请求的响应头如果请求报 CORS响应头里一定没有Access-Control-Allow-Origin对照这个就能确认是不是配置问题。4. 核心环节用 LangGraph 把预约 Agent 编排出来4.1 LangGraph 和 LangChain 到底有啥区别这是热词里出现率极高的问题也是我在技术分享群被问烂了的问题。LangChain 的核心抽象是 Chain链就是把一连串的调用按顺序串起来上一个的输出是下一个的输入。核心路由控制靠 chain 内部的 Prompt 模板和条件编码但写复杂逻辑时很容易变成一堆 if-else。LangGraph 则引入了图Graph的概念节点是处理函数边是条件转移天然支持环、分支和状态持久化说白了就是给 AI 应用加了“状态机”能力。对于预约系统来说LangChain 的问题是一次预约请求可能需要多轮对话才能拿到完整的槽位信息你总不能让用户一次性把所有信息说全。LangGraph 可以做到第一轮用户说“帮我约设备 A”Agent 发现时间缺失走collect_time节点向用户追问第二轮用户说“明天下午3点”Agent 再检查设备是否存在、是否空闲、是否冲突整个过程的状态都保存在图的State对象里。这种能力用 Chain 做也不是不行但代码会绕到怀疑人生。注意听到“LangChain 和 LangGraph 都过时了”这种说法不必焦虑。作为个人项目和学习来说LangGraph 目前依然是编排 AI Agent 的最主流选择。框架更替是必然的但状态机和图编排的核心思想是通用的学会一种就能快速迁移到新工具。4.2 Agent 状态图设计节点和边的决策逻辑我先定义状态数据结构也就是 LangGraph 里的GraphState# agent/state.py from typing import TypedDict, Annotated, Optional from langgraph.graph.message import add_messages class LabState(TypedDict): messages: Annotated[list, add_messages] user_id: Optional[int] intent: Optional[str] equipment_name: Optional[str] equipment_id: Optional[int] date: Optional[str] start_time: Optional[str] end_time: Optional[str] available_slots: list holidays: list step: str每个字段都有清晰职责messages保存对话历史intent保存解析出来的意图预约/取消/查询/改期equipment_name是设备名字start_time/end_time是解析后的标准化时间available_slots是查询到的可预约时段step是当前状态。节点设计上我拆了 5 个parse_intent_node调用 LLM从用户最新一条消息中抽取意图和设备/时间槽位。把结果写入 state。query_slots_node调用 FastAPI 后端查询接口或直接调数据库服务查目标设备在指定日期哪些时段空闲。check_conflict_node判断用户请求时间段是否落在空闲列表里同时校验是否满足实验室规则夜间禁用、维护中不可约。book_slot_node满足条件时直接写库生成预约成功消息。handle_conflict_node产生冲突时从空闲列表里挑出两个最近的可替代时段生成带选项的回复引导用户重新选择。条件边是图的核心。我用的转移逻辑是parse_intent_node解析后如果equipment或time缺失转移到collect_info_node这个节点可以简单回复“请补充设备或时间段信息”否则去query_slots_node。check_conflict_node如果通过去book_slot_node不通过则去handle_conflict_node。handle_conflict_node返回后用户如果确认新的时间流程回到check_conflict_node重新校验。# agent/graph.py核心骨架 from langgraph.graph import StateGraph, END from agent.nodes import ( parse_intent_node, query_slots_node, check_conflict_node, book_slot_node, handle_conflict_node, collect_info_node ) g StateGraph(LabState) g.add_node(parse_intent, parse_intent_node) g.add_node(collect_info, collect_info_node) g.add_node(query_slots, query_slots_node) g.add_node(check_conflict, check_conflict_node) g.add_node(book_slot, book_slot_node) g.add_node(handle_conflict, handle_conflict_node) g.set_entry_point(parse_intent) g.add_conditional_edges( parse_intent, lambda state: collect_info if not state.get(equipment_id) or not state.get(start_time) else query_slots, {collect_info: collect_info, query_slots: query_slots} ) g.add_conditional_edges( check_conflict, lambda state: book_slot if state.get(available_slots) else handle_conflict, {book_slot: book_slot, handle_conflict: handle_conflict} ) g.add_edge(collect_info, parse_intent) g.add_edge(query_slots, check_conflict) g.add_edge(book_slot, END) g.add_edge(handle_conflict, END) app g.compile()这套状态图跑起来之后你拿中文自然语言请求去测会发现整个对话流程非常清晰。LangGraph 最爽的地方在于每个节点是普通 Python 函数可以写任何业务逻辑调试时只需要打印 state 就能看到每一步发生了什么。4.3 用 LangGraph CLI 快速测试 AgentLangGraph 提供了 LangGraph CLI 和 LangGraph Studio 可视化调试工具网页版可以在浏览器里直接对话测试 Agent还能回放每一步的 state。安装和启动命令# 安装 CLI 工具 uv tool install langgraph-cli # 在项目根目录启动开发服务器默认加载 langgraph.json 配置 langgraph dev如果你只有 API Key 没有本地模型需要在.env里配置OPENAI_API_KEY或者你用的其他兼容 API。LangGraph 的StateGraph本身不绑定任何模型你完全可以用任何 OpenAI 兼容接口这在实际项目中非常实用因为国内各家模型服务基本都提供 OpenAI 兼容的 SDK 接口。5. 硬骨头时间冲突检测和并发控制5.1 时间重叠判断怎么写才不漏时间冲突检测看着简单写起来非常容易漏。两个时间段(a1, a2)和(b1, b2)重叠的完整条件不是“开始时间在对方范围内”那么简单而是要判断四种情况完全覆盖、部分重叠、被包含、完全相等。最保险的判断方法是用“不是不重叠”来反推如果a2 b1或b2 a1则两段时间不重叠除此之外必重叠。写成 SQL Alchemy 查询就是from sqlalchemy import and_, or_, select from app.models.booking import Booking stmt select(Booking).where( Booking.equipment_id equipment_id, Booking.status.in_([confirmed, pending]), and_( Booking.start_time new_end, Booking.end_time new_start ) ) conflict await session.execute(stmt).scalar_one_or_none()这个写法把四种重叠一网打尽而且索引命中率高。如果你是直接用 SQLite 原生 SQL同理。5.2 SQLite 并发写问题与事务控制SQLite 在低并发场景下非常香零配置文件、单文件、好备份但预约系统是个典型的多读多写场景尤其是热门设备在周一早上九点的抢约高峰多个请求同时写可能会报database is locked。解决方案有三个层次第一层开启 WAL 模式允许读写并发。执行PRAGMA journal_modeWAL;读操作不再阻塞写操作。第二层设置busy_timeout让连接等待锁释放而不是立即报错比如设 3000ms。第三层在写关键业务比如预约提交时用事务的原子性保证不会出现脏写。FastAPI 里我用 SQLAlchemy 异步引擎连接 SQLite连接串加参数启用 WALDATABASE_URL sqliteaiosqlite:///./lab_reservation.db?check_same_threadFalse engine create_async_engine( DATABASE_URL, connect_args{timeout: 30}, )如果你预判未来并发量真的特别高几十人同时抢更稳妥的做法是把 SQLite 换成 PostgreSQL。但作为实验室内部系统SQLite WAL 扛住常规并发完全没问题实测 30 人并发抢约也没再出现锁问题。5.3 给 Agent 返回“智能”备选时段冲突检测从来不是预约系统的全部用户体验差异主要在冲突后的处理。普通系统说“对不起该时间段不可预约”我这个系统会返回“设备 A 在 2025-06-18 14:00-16:00 已被占用当前最接近的空闲时段是 16:00-18:00 和次日 09:00-11:00是否需要为您预订”这个能力来自handle_conflict_node的逻辑查到冲突后按时间连续性对一天内所有可用时段进行排序筛出与目标时间段起点差值最小的两个区间。代码实现思路查询目标日期内该设备所有已确认预约得到“已被占用区间”列表再拿整天时间线比如 08:00-22:00减去这些区间剩余的就是空闲块最后跟用户请求的时间段求最近差值排序。这样返回的选项不是随机的而是真正“最近”的可用时间。6. 填坑实录我踩过的那些坑6.1 pydantic v2 与 langchain-core 版本撕裂这是目前 LangChain/LangGraph 初学者最容易遇到的坑。FastAPI 默认用 Pydantic v2而早期版本的langchain-core用的是 Pydantic v1 的 API升级到新版后虽然兼容了 v2但如果你手动指定了旧版本的langchain-core或者混用了pydantic.v1和pydantic经常会出现ValidationError或ImportError。我遇到的实际报错是ImportError: cannot import name ValidationError from pydantic排查了半天最后发现是项目里同时存在两份 pydantic v1 和 v2且 langchain 相关包引用了 pydantic 的别名。解决方案不要手动锁langchain-core老版本尽量让 uv 解析依赖关系装最新版langgraph它会自动带你所需的langchain-core。如果还有包依赖 pydantic v1那就把整个虚拟环境删掉重建避免历史依赖干扰。很多人在这一步浪费时间实际上删.venv重建通常是最快的解法。6.2 LangGraph 安装失败别慌先看 Python 版本LangGraph 要求 Python 3.9 以上但在 3.9 上部分新版本特性表现一般实测最稳的是 Python 3.11 和 3.12。如果你用 uv 创建项目时不小心用了系统默认的 Python 3.8装langgraph大概率失败。解决办法是显式指定 Python 版本uv venv --python 3.11另外Windows 上装langgraph时如果有 C 扩展编译错误比如chromadb之类的伴随依赖建议先装cmake或者直接使用 WSL 环境能少很多折腾。这不是 LangGraph 本身的问题是 Python 生态在 Windows 上常见的依赖编译问题。6.3 FastAPI 初始化读取配置文件的正确姿势这个热词点出了一个很容易被忽略的问题FastAPI 项目多了之后配置管理容易失控。我推荐的模式是前面提过的pydantic-settings.env。但有一点坑如果你用了docker-compose部署.env文件的加载路径可能会漂移。保险做法是在启动命令里显式指定环境变量文件路径ENV_FILE_PATH/path/to/.env uvicorn app.main:app --host 0.0.0.0 --port 8000在SettingsConfigDict里设置env_file的同时也可以用os.environ.get(ENV_FILE_PATH)动态拼路径这个方式在本地开发和 Docker 部署都能适配。另一个习惯是不要把密钥提交到 git 仓库.env一定要加进.gitignore不然一旦仓库公开API Key 就泄露了。6.4 前后端联调与线程问题FastAPI 的async def路由天生跑在事件循环里但如果你在 async 路由里直接调用requests.get这种同步阻塞库整个请求会被卡住。预约系统里有一个需求是调用外部模型服务LangGraph 节点里调 LLM在 FastAPI 侧如果用同步客户端高并发下会拖垮服务。我当时用httpx.AsyncClient替代requests并在路由里await调用响应时间在并发场景下稳定了很多。如果你使用的 LangGraph 是同步的默认执行方式可以考虑在节点函数里用asyncio.run()或者单独跑线程池也可以直接用langgraph对异步节点的支持在定义节点时用async def编写节点函数状态图会以异步方式运行。这个细节在官方文档角落有提到但不仔细看很容易忽略。6.5 调试小技巧打印 state 和 中途回退 的 AgentStateAgent 类系统调试起来比普通后端难因为它跑在图上问题可能出现在任意节点。我强烈建议在开发阶段不要一上来就写完整前端而是先在 FastAPI 里加一个调试路由往 LangGraph 的StateGraph传入初始消息返回时打印每一步的 state 和中间产出app.post(/api/agent/debug) async def debug_agent(payload: dict): initial_state {messages: [{role: user, content: payload[message]}]} result await graph.ainvoke(initial_state) return resultgraph.ainvoke()是 LangGraph 里异步执行整张图的入口它会按状态图跑完全部节点并返回最终 state。调试时直接把 state 打出来看哪个节点没有正确写入字段基本就能定位 90% 的问题。另一条经验是给每个节点加一段print(fnode: x, state: {state})虽然不优雅但开发期极好用。线上再统一替换成 logging。最后再分享一个小技巧这个项目做完以后我最大的体会是FastAPI 和 LangGraph 的组合并不是什么高深莫测的新物种它本质上是“把确定的业务逻辑用 FastAPI 写得干干净净把不确定的语言交互用 LangGraph 写得灵活可控”。你在实际做的时候也守住这个边界Agent 负责理解系统负责规则不要指望大模型能替你保证数据一致性。如果你也要上这类系统建议第一版不要做太复杂先跑通单设备、单用户、标准时间段把图结构和接口跑顺再逐步加多设备、审批流、模糊时间解析。按这个节奏推进基本不会再踩我前面说的那些版本的坑。
返回列表