1.任务目标
构建一个基于状态机的多步骤智能体，使其能够执行循环、分支和暂停操作，系统不该依赖脆弱的 while True 循环。
2.任务目的
在生产环境中，简单的“聊天循环”极易崩溃。实际业务工作流需要确定性的状态管理和故障恢复机制。
3.demo设计
构建一个“智能内容生成与审查”工作流，包含以下几个核心状态节点：
分类节点 (
classifier)：根据输入任务类型进行分支路由。生成节点 (
generator)：撰写或优化初稿。评审节点 (
critic)：检查初稿质量。如果质量不合格且未达到最大循环次数，则循环回生成节点重试。人工审批节点 (
human_approval)：利用 LangGraph 的 Checkpointer 实现持久化与暂停，等待人工确认后才继续。
4.python代码实现
基本的deepseek api配置下就行，主要看节点与状态的跳转
import os
from pathlib import Path
from typing import Literal, TypedDict
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, StateGraph
from langgraph.types import interrupt
from pydantic import BaseModel, Field
load_dotenv(Path(__file__).resolve().parent / ".env")
def _require_env(name: str) -> str:
value = os.getenv(name, "").strip()
if not value:
raise RuntimeError(f"缺少环境变量 {name}，请在 PythonProjects/.env 中配置")
return value
def build_llm() -> ChatOpenAI:
return ChatOpenAI(
model=os.getenv("DEEPSEEK_MODEL", "deepseek-chat").strip() or "deepseek-chat",
api_key=_require_env("DEEPSEEK_API_KEY"),
base_url=_require_env("DEEPSEEK_BASE_URL"),
temperature=0.2,
timeout=60,
max_retries=2,
max_completion_tokens=1200,
)
llm = build_llm()
class CategoryResult(BaseModel):
category: Literal["code", "text"] = Field(description="code 表示代码任务，text 表示文本任务")
class CritiqueResult(BaseModel):
approved: bool = Field(description="草稿是否达到可提交人工终审的质量")
feedback: str = Field(description="未通过时给出具体修改意见；通过时简要说明理由")
classifier_llm = llm.with_structured_output(CategoryResult, method="function_calling")
critic_llm = llm.with_structured_output(CritiqueResult, method="function_calling")
# 1. 定义状态结构 (State)
class AgentState(TypedDict):
task: str
category: str # 分支类型
draft: str # 当前草稿
feedback: str # 评审意见
iteration: int # 循环计数器
approved: bool # 审批状态
# 2. 定义各个节点函数 (Nodes)
def classifier_node(state: AgentState) -> dict:
print(f"--- [节点: 分类] 正在分析任务: {state['task']} ---")
result = classifier_llm.invoke([
SystemMessage(content=(
"你是任务分类器。根据任务内容只返回一个类别："
"code 表示需要编写、修改或解释代码；text 表示文章、说明、总结等非代码内容。"
)),
HumanMessage(content=state["task"]),
])
print(f"--- [节点: 分类] 结果: {result.category} ---")
return {"category": result.category, "iteration": 0}
def generator_node(state: AgentState) -> dict:
iter_count = state.get("iteration", 0) + 1
print(f"--- [节点: 生成] 第 {iter_count} 次生成内容 ---")
feedback = state.get("feedback") or "无"
result = llm.invoke([
SystemMessage(content=(
"你是内容生成器。根据任务类别生成一份简洁草稿。"
"category 为 code 时输出可运行的代码；category 为 text 时输出正文。"
"如果已有评审反馈，按反馈修改。只输出草稿正文。"
)),
HumanMessage(content=(
f"任务: {state['task']}\n"
f"类别: {state.get('category', 'text')}\n"
f"上一版草稿: {state.get('draft') or '无'}\n"
f"评审反馈: {feedback}"
)),
])
return {"draft": result.content, "iteration": iter_count}
def critic_node(state: AgentState) -> dict:
print("--- [节点: 评审] 正在评估草稿质量 ---")
result = critic_llm.invoke([
SystemMessage(content=(
"你是评审员。判断草稿是否完成任务、结构清楚且没有明显错误。"
"达到标准则 approved 为 true；否则为 false，并在 feedback 中指出需要修改的点。"
)),
HumanMessage(content=(
f"任务: {state['task']}\n"
f"类别: {state.get('category', 'text')}\n"
f"草稿:\n{state.get('draft', '')}"
)),
])
print(f"--- [节点: 评审] approved={result.approved} feedback={result.feedback} ---")
return {"feedback": result.feedback, "approved": result.approved}
# 3. 定义条件边与路由逻辑 (Branching & Looping)
def route_after_critic(state: AgentState) -> Literal["generator", "human_review"]:
"""
循环与分支决策逻辑：
- 如果未通过评审，且循环次数小于 3，则循环回 generator
- 否则进入人工审批暂停环节
"""
if not state["approved"] and state["iteration"] < 3:
print(">>> [路由决策] 评估未通过，触发循环：返回生成节点修改。")
return "generator"
else:
print(">>> [路由决策] 评估通过或达到上限，进入人工审批环节。")
return "human_review"
def human_review_node(state: AgentState) -> dict:
"""暂停并等待人工确认。通过后发布，驳回则带着修改意见回到生成节点。"""
print("--- [节点: 人工审核] 等待确认 ---")
decision = interrupt({
"task": state["task"],
"draft": state.get("draft", ""),
"feedback": state.get("feedback", ""),
"iteration": state.get("iteration", 0),
})
approved = bool(decision.get("approved"))
feedback = (decision.get("feedback") or "").strip() or state.get("feedback", "")
print(f"--- [节点: 人工审核] approved={approved} feedback={feedback} ---")
return {"approved": approved, "feedback": feedback}
def route_after_human(state: AgentState) -> Literal["generator", "final_polish"]:
if state["approved"]:
print(">>> [路由决策] 人工审核通过，进入终审发布。")
return "final_polish"
print(">>> [路由决策] 人工审核驳回，返回生成节点修改。")
return "generator"
def final_polish(state: AgentState) -> dict:
print(f"--- [节点: 终审发布] 内容已成功发布: {state['draft']} ---")
return {"approved": True}
# 4. 构建状态图 (StateGraph)
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node("classifier", classifier_node)
workflow.add_node("generator", generator_node)
workflow.add_node("critic", critic_node)
workflow.add_node("human_review", human_review_node)
workflow.add_node("final_polish", final_polish)
# 设置入口
workflow.set_entry_point("classifier")
# 添加普通边
workflow.add_edge("classifier", "generator")
workflow.add_edge("generator", "critic")
# 添加条件边（实现循环与分支跳转）
workflow.add_conditional_edges(
"critic",
route_after_critic,
{
"generator": "generator",
"human_review": "human_review",
}
)
workflow.add_conditional_edges(
"human_review",
route_after_human,
{
"generator": "generator",
"final_polish": "final_polish",
}
)
workflow.add_edge("final_polish", END)
# 5. 配置持久化。人工审核节点内部调用 interrupt() 暂停，等待确认后再继续。
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
测试脚本
from langgraph.types import Command
from main import app
# 初始化线程配置（用于状态持久化追踪）
config = {"configurable": {"thread_id": "production_task_001"}}
# 初始输入
initial_input = {"task": "编写一个Python快速排序脚本"}
def run_until_pause(graph_input):
"""执行到人工审核暂停点，或直到流程结束。"""
interrupted = False
for event in app.stream(graph_input, config):
if "__interrupt__" in event:
interrupted = True
continue
print(event)
return interrupted
def ask_human(snapshot) -> dict | None:
values = snapshot.values
print("\n========== 人工审核 ==========")
print(f"任务: {values.get('task')}")
print(f"当前轮次: {values.get('iteration')}")
print(f"模型评审: {values.get('feedback')}")
print("---------- 草稿 ----------")
print(values.get("draft"))
print("--------------------------")
while True:
choice = input("请确认：通过输入 y，驳回输入 n，退出输入 q: ").strip().lower()
if choice in {"y", "yes", "通过"}:
return {"approved": True, "feedback": values.get("feedback") or "人工审核通过"}
if choice in {"n", "no", "驳回"}:
feedback = input("请输入修改意见: ").strip()
if not feedback:
print("驳回时需要填写修改意见。")
continue
return {"approved": False, "feedback": feedback}
if choice in {"q", "quit", "退出"}:
return None
print("请输入 y、n 或 q。")
print("=== 工作流开始运行 ===")
pending = run_until_pause(initial_input)
while pending:
decision = ask_human(app.get_state(config))
if decision is None:
print("\n已退出。当前草稿未发布，本次进程结束后暂停状态不会保留。")
break
print("\n=== 已收到人工审核结果，恢复工作流 ===")
pending = run_until_pause(Command(resume=decision))
else:
print("\n=== 工作流结束 ===")
评论区