企业官网建设流程全解析

调研 Agent + LangGraph interrupt_before 人工审批（human-in-the-loop），使用SqliteSaver，用来 SQLite 文件持久保存 Agent 的 Graph 状态（State）

演示的是LangGraph 如何在节点之间保存状态、暂停执行，再根据人工输入继续执行

测试结果

输入问题，并不会立刻结束，而是等待用户继续输入（ok，stop，feedback）

状态里保存什么？

状态的结构定义在ReviewState：初始请求、草稿、反馈历史、用户输入、当前动作、流程状态，以及最终报告。每个节点接收当前状态，并返回要更新的字段。

class ReviewState(TypedDict, total=False): 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 层级裁剪做对比”为例：

1. generate_draft根据初始请求生成草稿，写入draft。
2. 图准备执行review_handler时，由interrupt_before=["review_handler"]暂停，等待审批输入。中断配置在这里。
3. continue_review打印草稿、调用input()等待用户输入，然后通过graph.update_state(..., {"review_input": review})把输入写进状态。
4. graph.invoke(None, config)从保存的检查点继续执行；这里的None表示继续已有会话，而不是传入一份全新状态。
5. review_handler读取review_input：
   • ok：设置action="approve"，路由到正式报告节点。
   • stop：设置action="stop"，结束流程。
   • 其他文字：设置action="revise"，并追加到feedback_history。
6. 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 = ( f"Initial request:\n{state['initial_request']}\n\n" f"Previous 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." ), prompt=prompt, ) 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=( f"Original request:\n{state['initial_request']}\n\n" f"Approved 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( checkpointer=checkpointer, interrupt_before=["review_handler"], )

SQLite 在这里做什么？

SqliteSaver保存每个会话的检查点。会话用thread_id区分，因此程序暂停或重启后，可以从同一个thread_id恢复状态。