毕设机器人技术解构：从任务调度到高可用部署的完整实践-智慧文博士

毕设机器人技术解构：从任务调度到高可用部署的完整实践

每到毕业季，高校教务群就像春运售票大厅：同一篇格式要求被反复@，凌晨两点还有人问“封面页码到底要不要罗马数字”。去年我们给学院搭了一套“毕设机器人”，把平均响应时间从 40 min 压到 3 s，高峰期扛住 1.2 k QPS。今天把踩过的坑、量过的代码、调过的参数全部摊开，供下一届想“用技术换老师头发”的同学直接抄作业。

1. 高校场景下的典型痛点

咨询洪峰集中爆发：教务通知一出，3 小时内 80% 问题重复，人工答疑占用导师 30% 工作时间。
流程状态不透明：学生不知道“开题→中期→查重→答辩”当前卡在哪一环，反复追问进度。
答案版本碎片化：同一学院不同教研室对“参考文献格式”要求不一致，群文件来回覆盖，学生拿到的答案常常过期。
并发窗口短且陡：答辩前一周 QPS 是平时的 20 倍，传统单体服务直接 502。

2. 技术选型对比：Dialogflow vs. Rasa vs. 自研规则引擎

维度	Dialogflow	Rasa	自研规则引擎
单轮延迟	600-800 ms（含外网）	120 ms（本地 GPU）	5 ms（内存查询）
成本（1k 日活）	550 美元/月	0（开源）+ 2 核 GPU 云主机	1 核 2 G 学生机即可
可控性	黑盒，意图 100 条后不可回滚	可解释，但需标注数据	代码即文档，Git 回滚
中文鲁棒	依赖 Google 分词，专业术语易漂移	需自训 BERT，数据量 2 k+	正则+同义词表，可热更新
离线场景	必须联网	可离线	完全离线

结论：

预算为零、答案集合稳定、需要 100% 可控 → 自研规则引擎是最不坏选择。
把 LLM 当“兜底语义搜索”而非主路径，可兼顾“低成本”与“泛化能力”。

3. 核心架构设计

3.1 基于有限状态机（FSM）的对话管理

把毕设抽象成 7 个状态：Start、Proposal、MidTerm、Paper、CheckSimilarity、Defense、End。
每条消息只触发一次状态迁移，杜绝“重复提交开题报告”类 bug。

迁移条件用“事件”表达，伪代码如下：

class DefenseNode(FSMNode): async def handle(self, ctx: Context) -> str: if ctx.similarity_rate < 20: return "查重未通过，请先修改后再次提交" ctx.state = End return "答辩已完成，恭喜毕业！"

好处：

代码即流程图，产品与教务老师能看懂。
单元测试可直接pytest test_defense_node.py，无需起完整服务。

3.2 异步任务队列解耦

把“查重”这类耗时 5-30 s 的第三方调用拆成两个微服务：

Chatbot-API：只负责状态机与内存缓存，< 50 ms 返回。
Worker：消费 RabbitMQ，执行查重、生成 PDF、发邮件等重任务。

队列设计要点：

使用“业务单号”做路由键，保证同一学生的多条消息顺序消费。
设置 TTL=30 min，超时直接 NACK，前端轮询拿到“超时请重试”提示，避免永久挂起。

4. 完整可复现代码（Python 3.11）

以下示例遵循 Clean Code 原则：单一职责、显式优于隐式、函数长度 < 20 行。

# fsm.py from enum import Enum, auto from dataclasses import dataclass class State(Enum): START = auto() PROPOSAL = auto() MIDTERM = auto() END = auto() @dataclass(slots=True) class Context: user_id: str state: State similarity_rate: float | None = None class Node(Protocol): async def handle(self, ctx: Context) -> str: ... class ProposalNode: async def handle(self, ctx: Context) -> str: if "开题报告" in ctx.text: ctx.state = State.PROPOSAL return "已收到开题报告，3 个工作日内在系统更新状态。" return "请先提交开题报告（模板下载：xxx）" # router.py class FSMMachine: def __init__(self) -> None: self.nodes: dict[State, Node] = { State.START: ProposalNode(), State.PROPOSAL: MidTermNode(), } async def react(self, ctx: Context) -> str: node = self.nodes.get(ctx.state) if not node: return "状态未知，请联系教务老师" return await node.handle(ctx) # api.py from fastapi import FastAPI app = FastAPI() machine = FSMMachine() @app.post("/chat") async def chat(req: ChatRequest): ctx = await redis.get(req.user_id) or Context(user_id=req.user_id, state=State.START) reply = await machine.react(ctx) await redis.set(req.user_id, ctx, ex=3600) return {"reply": reply}

运行uvicorn api:app --workers 4即可拉起服务，内存占用 < 120 MB。

5. 高并发下的稳定性三板斧

幂等性：
利用 RedisSET user_id:msg_hash 1 NX EX 60防止群聊里同一条消息被重复消费。
冷启动延迟：
把状态机节点提前import并@lru_cache编译后的正则；Docker 镜像里加PYTHONPATH预编译至.pyc，容器拉起 1.8 s → 0.4 s。
安全防护：
- 输入过滤：用re.match(r"[\u4e00-\u9fa5\w ,\.]+", text)丢弃表情与乱码，防止 LLM 提示注入。
- 权限校验：教务网 OAuth2 返回的 JWT 带role=student，机器人拒绝“批量下载全校论文”这类越权指令。

6. 生产环境避坑指南

日志追踪：
统一 JSON 格式，trace_id=uuid1|user_id|msg_id，Filebeat 直送 ES；排查时可按trace_id一键拉通前端→队列→Worker 全链路。
状态持久化丢失：
Redis 别忘开appendonly yes，曾经 RDB 异步快照掉电，30 分钟状态归零，学生以为报告被吞。
第三方 API 限流：
查重接口 10 QPS，超了直接封 IP。本地做令牌桶 + 退避重试，代码如下：

from asyncio import sleep from limits import RateLimiter limiter = RateLimiter(key_func=lambda: "check_api", rate="10/second") async def check_similarity(text: str) -> float: await limiter.wait() # 阻塞直到令牌可用 async with aiohttp.ClientSession() as sess: for attempt in range(3): async with sess.post(API, json={"text": text}) as r: if r.status == 429: await sleep(2 ** attempt) continue return await r.json()