FastAPI + LangChain Agent + RAG：打造共享办公智能助手（制度问答 / 工位查询 / 参观预约）

一、项目背景

二、数据模型：会员 + 工位

app/models/gongwei.py 定义了两张表：

from tortoise import models, fields

# 会员表
class member(models.Model):
id = fields.IntField(pk=True, auto_increment=True)
name = fields.CharField(max_length=20) # 姓名
phone = fields.CharField(max_length=11) # 手机号
is_delete = fields.IntField(default=0) # 逻辑删除

class Meta:
table = 'member'

# 工位表
class desk(models.Model):
id = fields.IntField(pk=True, auto_increment=True)
name = fields.CharField(max_length=20) # 工位名称
zone = fields.CharField(max_length=20) # 区域（阳光区/静音区）
monthly_price = fields.IntField(default=100) # 月租金
status = fields.IntField(default=1) # 1空闲 / 0已占用
create_time = fields.DateField(auto_now_add=True)
is_delete = fields.IntField(default=0) # 逻辑删除

class Meta:
table = 'desk'

和上一篇的回收项目对比，这里有个值得注意的差异：

同一套业务骨架，字段风格不完全一致，这是真实项目的常态——模型设计跟着业务走，不必强求统一模板。

三、RAG 知识库：从 docx 到 md 的减法

app/routers/gongwei/case.py 的链路和上次完全一样：读文件 → 分片 → 嵌入 → 入库，

import os
import chromadb
from openai import OpenAI

raw_key = os.getenv("DASHSCOPE_API_KEY")
api_key = raw_key.strip()

# 1. 读取 md 文件：原生文件读取，不再依赖 docx 库
def load_file(file_path):
with open(file_path, "r", encoding="utf-8") as f:
all_text = f.read()
return all_text

# 2. 文本分片：每 200 字符一片
def split_text(all_text):
chunks = []
for i in range(0, len(all_text), 200):
chunks.append(all_text[i:i + 200])
return chunks

# 3. 文本向量化：qwen3.7-text-embedding，每批 20 条
def text_embedding(chunks):
client = OpenAI(
api_key=api_key,
base_url="https://ws-xfbad2xxmhwytrdx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)
if isinstance(chunks, str):
chunks = [chunks]
batch_size = 20
embeddings = []
for i in range(0, len(chunks), batch_size):
batch_chunks = chunks[i:i + batch_size]
completion = client.embeddings.create(
model="qwen3.7-text-embedding",
input=batch_chunks,
)
batch_embeddings = [item.embedding for item in completion.data]
embeddings.extend(batch_embeddings)
return embeddings

# 4. 向量入库
def save_to_vector_db(chunks):
embeddings = text_embedding(chunks)
client = chromadb.PersistentClient(path=r"E:\\shijian\\P4\\zuoye\\demo0719\\app\\routers\\gongwei")
collection = client.get_or_create_collection(
name="cowork_handbook",
embedding_function=None # 向量外部算好，入库不二次嵌入
)
ids = [f"chunk_{i}" for i in range(len(chunks))]
collection.add(ids=ids, embeddings=embeddings, documents=chunks)

if __name__ == '__main__':
all_text = load_file(r"E:\\shijian\\P4\\zuoye\\demo0719\\app\\routers\\gongwei\\共享办公入驻手册.md")
save_to_vector_db(split_text(all_text))
print("构建完成")

相比上次的三个关键优化/注意点：

知识库内容来自《共享办公入驻手册.md》，覆盖六大类制度：门禁与到访、工位与物品、网络与会议、押金与退租、宠物与行为规范、发票与续费。

四、工位 CRUD 接口

app/routers/gongwen_api.py 提供工位列表和添加两个接口：

from fastapi import APIRouter, Query
from app.models import desk
from app.schemas.gongwei import GongWei

gongwei_router = APIRouter(prefix="/gongwei", tags=["工位"])

@get 列表：分页 + 模糊搜索 + 状态筛选
@post 添加：创建工位

列表接口的筛选逻辑值得一提：

@gongwei_router.get("/list", summary="工位列表")
async def gongwei_list(name: str = None, status: int = None, page: int = Query(1, ge=1),
size: int = Query(3, ge=2)):
query = desk.all()
if name:
query = query.filter(desk.name.contains(name)) # 模糊搜索
if status:
query = query.filter(desk.status == status) # 状态筛选
offset = (page – 1) * size
gongwei = await query.filter().offset(offset).limit(size)
gongwei_list = []
for i in gongwei:
gongwei_list.append({
"name": i.name,
"zone": i.zone,
"monthly_price": i.monthly_price,
"status": i.status,
})
return {"code": "200", "msg": "查询成功", "data": gongwei_list}

细节：desk.name.contains(name) 是 Tortoise 的模糊查询写法，等价于 SQL 的 LIKE '%name%'；分页用 offset/limit 组合，是 FastAPI 查询接口的标配。Pydantic 的 Query(1, ge=1) 还顺手做了参数校验——页码最小为 1，每页最小为 2。

五、核心：Agent 智能助手

app/routers/gongwen_llm.py，三个工具分别对应制度问答、工位查询、参观预约。

5.1 工具一：制度问答（rag_search）

会员问"押金要交多少"“能带宠物吗”，向量化问题 → Chroma 检索手册最相关 2 段 → 拼 prompt 交给大模型：

@tool(description="查询共享办公入驻制度手册，押金/门禁/宠物/退租等制度问题调用，参数 query：用户提问")
def rag_search(query: str) –> str:
chroma = chromadb.PersistentClient(path=r"E:\\shijian\\P4\\zuoye\\demo0719\\app\\routers\\gongwei")
collection = chroma.get_collection(name="cowork_handbook")
query_str = text_embedding(query)
result = collection.query(query_embeddings=query_str, n_results=2, include=["documents"])
retsult_text = "\\n".join(result["documents"][0])
pormpt = f"""
根据共享办公入驻制度手册，押金/门禁/宠物/退租等制度问题调用，
用户输入问题为：{query}
请根据查询结果，给出详细回答：{retsult_text}
"""
resp = llm.invoke([{"role": "user", "content": pormpt}])
return resp.content

和上次的 rag_search 对比，这次提示词相对宽松（“根据查询结果给出详细回答”），没有上次那么强的"禁止编造"约束——提示词的严格程度应该跟业务风险挂钩：涉及危险品处置（上次）必须硬约束，制度问答（本次）可以宽松些。这是设计上的取舍，不是失误。

5.2 工具二：查工位（search_desks）

会员说"阳光区有没有空位"，或者只说"看看有哪些工位"，AI 都能处理：

@tool(description="查询工位，参数zone传区域名称(阳光区/静音区)，不传则查询所有可租工位")
async def search_desks(zone: str = None):
query = desk.filter(is_delete=0, status=1) # 只看未删除 + 空闲
if zone:
query = query.filter(zone=zone) # 按区域过滤
desks = await query.all()
if not desks:
return "暂时没有空闲工位"
text = ""
for i in desks:
text += f"工位编号：{i.id}，区域：{i.zone}，工位状态：{i.status}(1表示空闲，0表示已占用)，月租金:{i.monthly_price}\\n"
return text

注意 zone: str = None 是可选参数——LangChain 工具支持默认值，Agent 会根据用户问题决定传不传：提到区域就过滤，没提就全查。返回文本里把 1/0 的含义写清楚，模型才知道怎么组织成"人话"回复给用户。

5.3 工具三：创建参观预约（create_visit_booking）

预约参观是业务核心，去重逻辑用 Redis 实现——同会员同一天只能预约 1 次：

@tool(description="创建参观预约；member_id会员ID，desk_id工位ID，visit_date日期YYYY-MM-DD；同会员同一天仅1次预约")
async def create_visit_booking(member_id: int, desk_id: int, visit_date: str):
members = await member.filter(id=member_id).first()
if not members:
return "会员不存在"
desks = await desk.filter(id=desk_id).first()
if not desks:
return "工位不存在"
if desks.status != 1:
return "工位已占用"

# Redis 去重：同一会员同一天只允许 1 次预约
key = f"member_id:desk_id:{visit_date}"
fishe = await r.lrange(key, 0, –1)
for i in fishe:
info = json.loads(i)
if info["member_id"] == member_id:
return "同会员同一天仅1次预约"

booking_no = f"D{datetime.datetime.now().strftime('%Y%m%d%H%M%S')}"
desk_dict = {
"booking_no": booking_no,
"member_id": member_id,
"member_name": members.name,
"visit_date": visit_date,
"desk_id": desk_id,
"desk_name": desks.name,
"create_time": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
}
await r.lpush(key, json.dumps(desk_dict, ensure_ascii=False))
return f"预约成功！预约编号：{booking_no}，工位名称：{desks.name}"

三层校验层层递进：会员存在 → 工位存在且空闲 → Redis 查重。跟前一篇的 create_pickup_order 对比，去重策略换了个思路：

LPUSH + LRANGE 的做法更适合"按条件去重 + 数据可查"的场景——它把每次预约都存成了完整记录，后面接查询、统计都不用再设计数据结构。

5.4 Agent 组装

tools = [rag_search, search_desks, create_visit_booking]
system_prompt = f"""你是共享办公助手，仅简体中文，严格执行：
1. 用户询问押金、门禁、访客、宠物、退租、续费等制度问题，必须调用rag_search_tool检索手册，禁止编造；
2. 用户查询工位，调用search_desks工具；
3. 用户预约参观，调用create_visit_booking，需要会员ID、工位ID、参观日期；
4. 预约前校验同一会员同一天不可重复预约，工位必须可租；
业务需求必须调用对应工具，不是业务要求直接回答用户。
"""
checkpointer = InMemorySaver()
chant = create_agent(
model=llm, # qwen3.7-plus
tools=tools,
system_prompt=system_prompt,
checkpointer=checkpointer, # 会话记忆
)

system prompt 的写法值得学习：把"什么情况调什么工具"用编号列清楚，工具职责边界一目了然，模型路由错误的概率大幅下降。这是 Agent 工程里性价比最高的一环——先把 prompt 写好，比堆中间件更见效。

5.5 对外接口

@llm2_router.post("/chat", summary="与智能助手进行对话")
async def chat(req: DuiHua):
config = {"configurable": {"thread_id": req.session_id}} # 会话隔离
input_msg = {"messages": [HumanMessage(content=req.user_query)]}
res = await chant.ainvoke(input_msg, config=config)
return {
"code": 200,
"msg": "对话执行成功",
"data": res["messages"][–1].content,
}

前端只传 session_id + user_query，剩下全交给 Agent：

用户: "押金要交多少？"
→ rag_search("押金") → 检索手册 → "入驻时缴纳押金为月租金的1倍，退租无损坏无欠费全额退还"

用户: "静音区有空位吗？"
→ search_desks(zone="静音区") → 返回空闲工位列表

用户: "帮我约明天参观 3 号工位，我是 2 号会员"
→ create_visit_booking(member_id=2, desk_id=3, visit_date="2026-08-12") → 返回预约编号