本篇拆解四个 Agent 里工程复杂度最高的试卷批改 AgentExam。它展现了两个 LangGraph 的核心能力一是asyncio.gather驱动的三轨并行批改规则引擎 / LLM 语义 / 代码评估二是HitLHuman-in-the-Loop——图执行到教师审核节点用interrupt()暂停、等教师确认后Command(resume...)恢复。这是把AI 自动化和人工把关结合起来的教科书级实现。一、整体流程线性图 一个中断点Command resumeSTARTparse_word解析Word答卷load_questions_meta加载试卷元数据run_three_tracks三轨并行批改aggregate_results汇总预批改analyze_weak_points知识薄弱点分析notify_teacher通知教师teacher_reviewinterrupt 暂停apply_teacher_decision合并教师决策publish_results发布给学员END图结构是纯线性链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:questionsstate[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_exceptionsTrue某一轨抛异常不会中断其他轨rawawaitasyncio.gather(_run_objective_track(objective_qs),# 第一轨规则引擎_run_subjective_track(subjective_qs),# 第二轨LLM语义评分_run_code_track(code_qs),# 第三轨LLM代码评估return_exceptionsTrue,)# 逐轨检查失败的轨给空列表不拖后腿objective_resultsraw[0]ifnotisinstance(raw[0],Exception)else[]subjective_resultsraw[1]ifnotisinstance(raw[1],Exception)else[]code_resultsraw[2]ifnotisinstance(raw[2],Exception)else[]return{objective_results:objective_results,subjective_results:subjective_results,code_results:code_results}关键设计return_exceptionsTrue。默认情况下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_correctstudent_normcorrect_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_SIZE3# 每组最多3道题并行防止同时请求太多LLM被限速groups[questions[i:iGROUP_SIZE]foriinrange(0,len(questions),GROUP_SIZE)]all_results[]forgroupingroups:# 组间顺序执行group_resultsawaitasyncio.gather(*[_review_one_subjective(q)forqingroup],# 组内并行return_exceptionsTrue,)forq,resultinzip(group,group_results):ifisinstance(result,Exception):# 单题失败降级0分 needs_reviewTrue强制教师复核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用结构化输出模型返回SubjectiveReviewResultclassSubjectiveReviewResult(BaseModel):question_id:strstudent_answer:strtotal_score:int# 各得分点 earned 的分值之和full_score:intconfidence:float# LLM 置信度 [0,1]0.7 标记需复核point_results:list[ScoringPointResult]# 得分点明细overall_comment:strScoringPointResult是更细的维度每个得分点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 根据错题内容推断知识点。合并去重后生成WeakPointsReportclassWeakPoint(BaseModel):tag:str# 知识点标签如 Spring IOCwrong_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_decisioninterrupt(display_data)logger.info(teacher_review.resumed,actionteacher_decision.get(action))return{teacher_decision:teacher_decision}interrupt()做两件事把display_data暴露给外部教师端可通过接口读取冻结图执行当前 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(resumedecision), config) → 图从断点恢复5.3 恢复Command(resume…)教师确认后API 层用Command恢复图fromlanggraph.typesimportCommand resultawaitgraph.ainvoke(Command(resumedecision),# decision {action: approve/modify, ...}configconfig,# 同一个 thread_id)恢复后interrupt()的返回值就是decision写入state[teacher_decision]图继续往后走。5.4 合并教师决策apply_teacher_decision_node处理两种教师动作asyncdefapply_teacher_decision_node(state:ExamState)-dict:decisionstate.get(teacher_decision,{})actiondecision.get(action,approve)modificationsdecision.get(modifications,[])# 底子 AI 预批改结果教师修改是在上面覆盖all_resultslist(state.get(pre_review_summary,{}).get(by_question,[]))forrinall_results:r[final_score]r.get(score,0)# 默认 final AI 分ifactionmodifyandmodifications:id_to_idx{r[question_id]:idxforidx,rinenumerate(all_results)}formodinmodifications:qidmod.get(question_id)ifqidinid_to_idx:idxid_to_idx[qid]ifnew_scoreinmod:all_results[idx][teacher_score]mod[new_score]all_results[idx][final_score]mod[new_score]ifcommentinmod: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.pyExam 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 没把握的留给人。七、本篇小结知识点实现位置核心技巧三轨并行run_three_tracks_nodegather(return_exceptionsTrue)各轨独立成败客观题零 LLM_run_objective_track规则引擎字符串比对限流保护_run_subjective_track3题一组组内并行、组间串行得分点证据链ScoringPointResult每题给出哪句得分/哪句丢分HitL 暂停teacher_review_nodeinterrupt(display_data)冻结图HitL 恢复API 层Command(resumedecision) 同 thread_id决策合并apply_teacher_decision_node默认AI分 教师覆盖 留痕幂等发布publish_results_node先删后插 状态机流转Exam Agent 最值得学习的不是某个炫技点而是自动化与人工的边界划分客观题规则批、主观题 AI 批、AI 没把握的标教师复核、教师终审留痕、发布幂等——每一层都在回答这个环节 AI 能不能兜住兜不住怎么办。下篇预告下一篇《EduAgent 项目全解析六模拟面试 Agent——五阶段状态机》看最后一个 Agent用InterviewStage枚举定义热身→技术基础→项目深挖→反问收尾→结束五个阶段check_stage节点按轮数和回答质量驱动阶段流转回答质量四档标签excellent/adequate/weak/no_answer驱动追问与换题决策最后生成五维度面试报告。项目源码仅供教学参考欢迎评论区交流讨论。