AI 带人工审批的调研 Agent

调研 Agent LangGraph interrupt_before 人工审批human-in-the-loop使用SqliteSaver用来 SQLite 文件持久保存 Agent 的 Graph 状态State演示的是LangGraph 如何在节点之间保存状态、暂停执行再根据人工输入继续执行测试结果输入问题并不会立刻结束而是等待用户继续输入okstopfeedback状态里保存什么状态的结构定义在ReviewState初始请求、草稿、反馈历史、用户输入、当前动作、流程状态以及最终报告。每个节点接收当前状态并返回要更新的字段。class ReviewState(TypedDict, totalFalse): initial_request: str draft: str feedback_history: list[str] review_input: str action: str status: str final_report: str流程如何转换State 保存数据Node 完成任务Edge 连接步骤Conditional edge 决定分支Checkpoint 记录进度Interrupt 把控制权交还。START│▼generate_draft ← Node生成草稿更新 State│ ← Edge固定路线▼【暂停并保存 Checkpoint】││ 控制台等待输入│ update_state → invoke(None)▼review_handler ← Node解析输入更新 action│▼route_after_review ← Conditional edge选择路线├─ approve → make_final_report → END├─ revise → generate_draft → 再次暂停└─ stop → END以“补充和 Tile 层级裁剪做对比”为例generate_draft根据初始请求生成草稿写入draft。图准备执行review_handler时由interrupt_before[review_handler]暂停等待审批输入。中断配置在这里。continue_review打印草稿、调用input()等待用户输入然后通过graph.update_state(..., {review_input: review})把输入写进状态。graph.invoke(None, config)从保存的检查点继续执行这里的None表示继续已有会话而不是传入一份全新状态。review_handler读取review_inputok设置actionapprove路由到正式报告节点。stop设置actionstop结束流程。其他文字设置actionrevise并追加到feedback_history。route_after_review根据action选择下一节点。反馈会回到generate_draft它会同时读取上一版draft和反馈历史生成新草稿然后再次暂停等待审批def build_graph( settings: Settings, *, checkpointer: SqliteSaver, azure_client: Any | None None, ): client azure_client if azure_client is not None else create_azure_client(settings) def generate_draft(state: ReviewState) - dict[str, Any]: previous_draft state.get(draft, ) feedback state.get(feedback_history, []) prompt ( fInitial request:\n{state[initial_request]}\n\n fPrevious draft:\n{previous_draft or (No previous draft)}\n\n Reviewer feedback to incorporate:\n (\n.join(f- {item} for item in feedback) or (No feedback yet)) \n\nWrite a clear, useful draft that addresses the request and incorporates all reviewer feedback. ) draft _call_responses( client, settings, instructions( You are a careful report-writing assistant. Produce only the requested draft. Do not claim facts that were not provided. ), promptprompt, ) return { draft: draft, review_input: , action: , status: awaiting_review, } def review_handler(state: ReviewState) - dict[str, Any]: review state.get(review_input, ).strip() if not review: raise ValueError(Review input must not be empty.) command review.casefold() if command ok: return {action: approve, review_input: , status: approved} if command stop: return {action: stop, review_input: , status: stopped} return { action: revise, review_input: , feedback_history: [*state.get(feedback_history, []), review], status: awaiting_revision, } def make_final_report(state: ReviewState) - dict[str, str]: report _call_responses( client, settings, instructions( You are a careful report editor. Turn the approved draft into a polished final report. Preserve its meaning and do not add unsupported claims. ), prompt( fOriginal request:\n{state[initial_request]}\n\n fApproved draft:\n{state[draft]} ), ) return {final_report: report, status: completed} def route_after_review(state: ReviewState) - str: return { approve: make_final_report, revise: generate_draft, stop: END, }[state[action]] workflow StateGraph(ReviewState) workflow.add_node(generate_draft, generate_draft) workflow.add_node(review_handler, review_handler) workflow.add_node(make_final_report, make_final_report) workflow.add_edge(START, generate_draft) workflow.add_edge(generate_draft, review_handler) workflow.add_conditional_edges(review_handler, route_after_review) workflow.add_edge(make_final_report, END) return workflow.compile( checkpointercheckpointer, interrupt_before[review_handler], )SQLite 在这里做什么SqliteSaver保存每个会话的检查点。会话用thread_id区分因此程序暂停或重启后可以从同一个thread_id恢复状态。