企业官网建设流程全解析

本篇拆解四个 Agent 里工程复杂度最高的试卷批改 Agent（Exam）。它展现了两个 LangGraph 的核心能力：一是asyncio.gather驱动的三轨并行批改（规则引擎 / LLM 语义 / 代码评估），二是HitL（Human-in-the-Loop）——图执行到教师审核节点用interrupt()暂停、等教师确认后Command(resume=...)恢复。这是把"AI 自动化"和"人工把关"结合起来的教科书级实现。

一、整体流程：线性图 + 一个中断点

图结构是纯线性链（add_edge一条线连到底），唯一的"智能"在teacher_review节点——它调用interrupt()冻结整个图，把控制权交给教师。

二、Word 解析与题目合并

2.1 同步解析 + 线程池

parse_word_node用python-docx解析学员作答的 Word 文件（同样走run_in_executor线程池桥接），提取每道题的题号、题型、学员答案。有个贴心细节——中文数字转阿拉伯数字的辅助函数：

def_chinese_to_int(s:str)->int:""""一"→1，"三"→3 ... 支持"十/百"组合

因为 Word 里学员可能写"三"也可能写"3"，解析时要归一化。

2.2 DB 合并元数据

load_questions_meta_node从 PostgreSQL 的questions表读取每道题的满分、得分点（scoring_points）、知识点标签，和 Word 解析出的题目合并成parsed_questions——这就是三轨批改的输入。

三、三轨并行批改（核心）

run_three_tracks_node把题目按题型分成三堆，用asyncio.gather同时跑三个协程：

asyncdefrun_three_tracks_node(state:ExamState)->dict:questions=state["parsed_questions"]# 按题型分成三堆objective_qs=[qforqinquestionsifq["question_type"]in("single_choice","multi_choice","judge")]subjective_qs=[qforqinquestionsifq["question_type"]=="short_answer"]code_qs=[qforqinquestionsifq["question_type"]=="code"]# 三轨并行启动# return_exceptions=True：某一轨抛异常不会中断其他轨raw=awaitasyncio.gather(_run_objective_track(objective_qs),# 第一轨：规则引擎_run_subjective_track(subjective_qs),# 第二轨：LLM语义评分_run_code_track(code_qs),# 第三轨：LLM代码评估return_exceptions=True,)# 逐轨检查：失败的轨给空列表，不拖后腿objective_results=raw[0]ifnotisinstance(raw[0],Exception)else[]subjective_results=raw[1]ifnotisinstance(raw[1],Exception)else[]code_results=raw[2]ifnotisinstance(raw[2],Exception)else[]return{"objective_results":objective_results,"subjective_results":subjective_results,"code_results":code_results}

关键设计：return_exceptions=True。默认情况下gather遇到第一个异常就整体取消，三轨会互相拖累；设成True后异常作为返回值保留，各轨独立成败——某轨炸了，其他轨照常出结果。

3.1 第一轨：客观题（规则引擎，零 LLM 成本）

asyncdef_run_objective_track(questions):results=[]forqinquestions:# 字符串归一化后精确比对：A == Astudent_norm=_normalize_answer(q["student_answer"])correct_norm=_normalize_answer(q["correct_answer"])is_correct=student_norm==correct_norm results.append({"question_id":q["question_id"],"question_no":q["question_no"],"is_correct":is_correct,"score":q["full_score"]ifis_correctelse0,"full_score":q["full_score"],})returnresults

客观题（单选/多选/判断）答案是确定的，根本不需要 LLM——规则引擎字符串比对，又快又准又省 token。这是"能用规则就不用模型"的典范。

3.2 第二轨：简答题（LLM 语义评分，3 题一组）

简答题需要语义理解，用结构化输出 LLM 按得分点评分。但这里有个限流保护：N 道题不是一次全并行，而是每 3 题一组、组内并行、组间串行：

asyncdef_run_subjective_track(questions):GROUP_SIZE=3# 每组最多3道题并行，防止同时请求太多LLM被限速groups=[questions[i:i+GROUP_SIZE]foriinrange(0,len(questions),GROUP_SIZE)]all_results=[]forgroupingroups:# 组间顺序执行group_results=awaitasyncio.gather(*[_review_one_subjective(q)forqingroup],# 组内并行return_exceptions=True,)forq,resultinzip(group,group_results):ifisinstance(result,Exception):# 单题失败降级：0分 + needs_review=True，强制教师复核all_results.append({...,"score":0,"needs_review":True,"ai_feedback":"AI评分失败，已标记需教师人工批改"})else:all_results.append(result)returnall_results

为什么要限流？一份试卷可能有几十道简答题，如果全部同时发 LLM 请求，很容易触发 DeepSeek API 的速率限制。3 题一组是延迟和并发之间的平衡点——这是真实生产环境才会考虑的问题。

单题批改_review_one_subjective用结构化输出模型，返回SubjectiveReviewResult：

classSubjectiveReviewResult(BaseModel):question_id:strstudent_answer:strtotal_score:int# 各得分点 earned 的分值之和full_score:intconfidence:float# LLM 置信度 [0,1]，<0.7 标记需复核point_results:list[ScoringPointResult]# 得分点明细overall_comment:str

ScoringPointResult是更细的维度：每个得分点（point_id/point_desc/point_score/earned/evidence/missing）——AI 不只给总分，还给出"哪句话得了分、哪句话丢了分"的证据链，这是教师复核和学员复盘的基础。

3.3 第三轨：代码题（LLM 综合评估）

代码题走"Think + 结构化评分"双阶段（和简历的 Think Tool 同款模式）：先让 LLM 通读代码和题目要求做自由推理（找问题、评估思路），再带着推理结论做结构化评分，输出分数、问题和改进建议。

四、汇总与薄弱点分析

4.1 aggregate_results：生成预批改报告

三轨结果合并后按题号排序，计算总分、客观题正确数、简答题平均置信度、需复核题数，生成pre_review_summary——这份汇总就是稍后展示给教师审核的核心数据。

4.2 analyze_weak_points：知识薄弱点分析

薄弱点分析用"规则 + LLM"双路合并：

• 有知识点标签的题：规则聚合——同一个knowledge_tag下错了几题、总共几题，直接算出来；
• 无标签的题：LLM 根据错题内容推断知识点。

合并去重后生成WeakPointsReport：

classWeakPoint(BaseModel):tag:str# 知识点标签（如 "Spring IOC"）wrong_count:int# 该知识点下做错的题数total_count:int# 该知识点下总共的题数question_nos:list# 涉及的题目序号suggestion:str# 复习建议

这份报告让学生一眼看到"我哪里薄弱、该复习什么"——这是 AI 批改相比人工批改的增值点。

五、HitL 核心：interrupt / resume 全流程

5.1 暂停：interrupt(display_data)

asyncdefteacher_review_node(state:ExamState)->dict:display_data={"submission_id":state["submission_id"],"student_id":state["student_id"],"pre_review_summary":state.get("pre_review_summary",{}),"weak_points":state.get("weak_points",[]),"weak_points_summary":state.get("weak_points_summary",""),"message":"请检查AI预批改结果和知识薄弱点分析，确认无误后点击发布。",}# ★ interrupt() 类似 input()：图在此暂停执行，等待教师确认teacher_decision=interrupt(display_data)logger.info("teacher_review.resumed",action=teacher_decision.get("action"))return{"teacher_decision":teacher_decision}

interrupt()做两件事：

1. 把display_data暴露给外部（教师端可通过接口读取）；
2. 冻结图执行，当前 State 自动保存到 MemorySaver checkpoint。

5.2 教师端交互时序

图执行到 teacher_review 节点 → interrupt() 暂停，state 存进 checkpoint ↓ 前端轮询 /exam/status → 发现 waiting_review ↓ 前端调 /exam/review/{session_id} → 后端从 checkpoint 读 state ↓ 教师看到 AI 预批改结果 + 薄弱点分析 ↓ 教师提交决策（approve 或 modify） ↓ 后端调 graph.ainvoke(Command(resume=decision), config) → 图从断点恢复

5.3 恢复：Command(resume=…)

教师确认后，API 层用Command恢复图：

fromlanggraph.typesimportCommand result=awaitgraph.ainvoke(Command(resume=decision),# decision = {"action": "approve"/"modify", ...}config=config,# 同一个 thread_id)

恢复后，interrupt()的返回值就是decision，写入state["teacher_decision"]，图继续往后走。

5.4 合并教师决策

apply_teacher_decision_node处理两种教师动作：

asyncdefapply_teacher_decision_node(state:ExamState)->dict:decision=state.get("teacher_decision",{})action=decision.get("action","approve")modifications=decision.get("modifications",[])# 底子 = AI 预批改结果，教师修改是在上面覆盖all_results=list(state.get("pre_review_summary",{}).get("by_question",[]))forrinall_results:r["final_score"]=r.get("score",0)# 默认 final = AI 分ifaction=="modify"andmodifications:id_to_idx={r["question_id"]:idxforidx,rinenumerate(all_results)}formodinmodifications:qid=mod.get("question_id")ifqidinid_to_idx:idx=id_to_idx[qid]if"new_score"inmod:all_results[idx]["teacher_score"]=mod["new_score"]all_results[idx]["final_score"]=mod["new_score"]if"comment"inmod:all_results[idx]["teacher_comment"]=mod["comment"]all_results[idx]["reviewed_by"]=decision.get("teacher_id","")return{"final_results":all_results}

approve就直接用 AI 分数；modify按question_id逐条覆盖教师给的新分/新评语，并记录reviewed_by（谁审的）。这个"默认 AI 分 + 教师覆盖"的合并模型，是 HitL 的通用范式——AI 给初稿，人做终审，所有修改留痕。

5.5 发布

publish_results_node写两张表：exam_reviews（每道题批改明细，先删后插保证幂等——教师重复点发布不会产生重复记录）和exam_submissions（状态更新为published+ 薄弱点 JSON）。

六、降级策略：批改也要兜底

配合第三篇讲的retry.py，Exam Agent 有两个专属降级策略：

# 代码批改降级：评分服务不可用 → 标记需教师人工复核asyncdef_exam_code_fallback(cls,error,func,args,kwargs)->dict:return{**state,"fallback_used":True,"needs_teacher_review":True,"fallback_note":"代码评分服务暂时不可用，已标记为需教师人工复核。"}# 简答题批改降级：同样标记教师复核asyncdef_exam_subjective_fallback(cls,error,func,args,kwargs)->dict:return{**state,"fallback_used":True,"needs_teacher_review":True,...}

降级的本质是"把不确定性转移给人"——AI 拿不准就标needs_teacher_review，让教师兜底，而不是硬给一个可能错误的分数。这正好和 HitL 机制呼应：AI 能批的 AI 批，AI 没把握的留给人。

七、本篇小结

Exam Agent 最值得学习的不是某个炫技点，而是"自动化与人工的边界划分"：客观题规则批、主观题 AI 批、AI 没把握的标教师复核、教师终审留痕、发布幂等——每一层都在回答"这个环节 AI 能不能兜住，兜不住怎么办"。

下篇预告

下一篇《EduAgent 项目全解析（六）：模拟面试 Agent——五阶段状态机》，看最后一个 Agent：用InterviewStage枚举定义热身→技术基础→项目深挖→反问收尾→结束五个阶段，check_stage节点按轮数和回答质量驱动阶段流转，回答质量四档标签（excellent/adequate/weak/no_answer）驱动追问与换题决策，最后生成五维度面试报告。

项目源码仅供教学参考，欢迎评论区交流讨论。