Tutorial 5: 循环与迭代
循环的重要性
在 AI Agent 中，循环是实现以下功能的关键：
迭代优化: 反复改进内容直到满意
重试机制: 失败后自动重试
对话循环: 持续的人机交互
自我纠错: Agent 检查并修正自己的输出
LangGraph 中的循环
LangGraph 原生支持循环，通过条件边实现：
┌─────────┐
│ START │
└────┬────┘
│
▼
┌─────────┐ ┌─────────┐
│ Process │────►│ Evaluate│
└─────────┘ └────┬────┘
▲ │
│ ┌──────────┴──────────┐
│ │ │
│ ▼ ▼
│ 不满意 满意
│ │ │
└────┘ ▼
┌─────────┐
│ END │
└─────────┘
基本循环模式
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
content: str
iteration: int
max_iterations: int
is_satisfied: bool
def process(state: State) -> dict:
"""处理节点"""
content = state["content"]
iteration = state.get("iteration", 0) + 1
# 模拟处理
improved_content = f"{content} [改进 #{iteration}]"
return {
"content": improved_content,
"iteration": iteration
}
def evaluate(state: State) -> dict:
"""评估节点"""
iteration = state["iteration"]
max_iter = state.get("max_iterations", 3)
# 模拟评估：达到最大迭代次数则满意
is_satisfied = iteration >= max_iter
return {"is_satisfied": is_satisfied}
def should_continue(state: State) -> Literal["continue", "end"]:
"""路由函数"""
if state["is_satisfied"]:
return "end"
return "continue"
# 构建图
graph = StateGraph(State)
graph.add_node("process", process)
graph.add_node("evaluate", evaluate)
graph.add_edge(START, "process")
graph.add_edge("process", "evaluate")
graph.add_conditional_edges(
"evaluate",
should_continue,
{
"continue": "process", # 循环回到 process
"end": END
}
)
app = graph.compile()
result = app.invoke({
"content": "初始内容",
"iteration": 0,
"max_iterations": 3,
"is_satisfied": False
})
print(f"最终内容: {result['content']}")
print(f"迭代次数: {result['iteration']}")
防止无限循环
最大迭代次数限制
def should_continue(state: State) -> str:
max_iterations = state.get("max_iterations", 5)
current = state.get("iteration", 0)
if current >= max_iterations:
return "end" # 强制结束
if state["is_satisfied"]:
return "end"
return "continue"
使用图的 recursion_limit
# 编译时设置递归限制
app = graph.compile()
# 运行时设置
result = app.invoke(
initial_state,
{"recursion_limit": 10} # 最多执行10步
)
实战：内容迭代优化系统
from typing import TypedDict, List, Literal, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
# ========== 状态定义 ==========
class ContentState(TypedDict):
topic: str
platform: str
current_content: str
quality_score: int
target_score: int
iteration: int
max_iterations: int
improvement_history: Annotated[List[dict], add]
# ========== 节点定义 ==========
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
def generate_initial(state: ContentState) -> dict:
"""生成初始内容"""
topic = state["topic"]
platform = state["platform"]
response = llm.invoke(f"""
为「{platform}」写一篇关于「{topic}」的短文（300字左右）。
""")
return {
"current_content": response.content,
"iteration": 1
}
def evaluate_content(state: ContentState) -> dict:
"""评估内容质量"""
content = state["current_content"]
response = llm.invoke(f"""
评估以下内容的质量（0-100分）：
{content}
评分标准：
- 内容深度 (40分)
- 结构清晰 (30分)
- 语言表达 (30分)
输出格式：
分数: [数字]
问题: [主要问题，一句话]
""")
result = response.content
# 解析分数
try:
score_line = [l for l in result.split("\n") if "分数" in l][0]
score = int(''.join(filter(str.isdigit, score_line)))
except:
score = 70
# 解析问题
try:
issue_line = [l for l in result.split("\n") if "问题" in l][0]
issue = issue_line.split(":")[-1].strip()
except:
issue = "需要整体优化"
return {
"quality_score": score,
"improvement_history": [{
"iteration": state["iteration"],
"score": score,
"issue": issue
}]
}
def improve_content(state: ContentState) -> dict:
"""改进内容"""
content = state["current_content"]
history = state["improvement_history"]
# 获取最新的问题
latest_issue = history[-1]["issue"] if history else "整体优化"
response = llm.invoke(f"""
请改进以下内容：
原文：
{content}
需要改进的问题：{latest_issue}
要求：
1. 针对问题进行改进
2. 保持原有优点
3. 输出完整的改进后内容
""")
return {
"current_content": response.content,
"iteration": state["iteration"] + 1
}
# ========== 路由函数 ==========
def should_continue(state: ContentState) -> Literal["improve", "done"]:
"""决定是否继续优化"""
score = state["quality_score"]
target = state.get("target_score", 85)
iteration = state["iteration"]
max_iter = state.get("max_iterations", 5)
# 达到目标分数或最大迭代次数
if score >= target:
return "done"
if iteration >= max_iter:
return "done"
return "improve"
# ========== 构建图 ==========
def create_improvement_workflow():
graph = StateGraph(ContentState)
graph.add_node("generate", generate_initial)
graph.add_node("evaluate", evaluate_content)
graph.add_node("improve", improve_content)
# 流程：生成 -> 评估 -> (改进 -> 评估) 循环
graph.add_edge(START, "generate")
graph.add_edge("generate", "evaluate")
graph.add_conditional_edges(
"evaluate",
should_continue,
{
"improve": "improve",
"done": END
}
)
graph.add_edge("improve", "evaluate")
return graph.compile()
# ========== 运行 ==========
workflow = create_improvement_workflow()
# 流式查看每步结果
print("开始内容优化流程...")
print("=" * 50)
for event in workflow.stream({
"topic": "AI编程入门",
"platform": "微信公众号",
"target_score": 85,
"max_iterations": 3,
"iteration": 0,
"improvement_history": []
}):
for node, output in event.items():
if node == "evaluate":
print(f"\n[评估] 第 {output.get('improvement_history', [{}])[-1].get('iteration', '?')} 轮")
print(f" 分数: {output.get('quality_score', 'N/A')}")
if output.get('improvement_history'):
print(f" 问题: {output['improvement_history'][-1].get('issue', 'N/A')}")
elif node == "improve":
print(f"\n[改进] 第 {output.get('iteration', '?')} 轮优化完成")
print("\n" + "=" * 50)
print("优化完成！")
带重试的循环
class RetryState(TypedDict):
task: str
result: str
error: str
retry_count: int
max_retries: int
success: bool
def execute_task(state: RetryState) -> dict:
"""执行任务（可能失败）"""
try:
# 模拟可能失败的操作
result = risky_operation(state["task"])
return {"result": result, "success": True, "error": ""}
except Exception as e:
return {
"error": str(e),
"success": False,
"retry_count": state.get("retry_count", 0) + 1
}
def should_retry(state: RetryState) -> Literal["retry", "success", "fail"]:
"""决定是否重试"""
if state["success"]:
return "success"
if state["retry_count"] >= state.get("max_retries", 3):
return "fail"
return "retry"
graph.add_conditional_edges(
"execute",
should_retry,
{
"retry": "execute", # 重试
"success": "complete",
"fail": "handle_error"
}
)
循环最佳实践
始终设置终止条件
def should_continue(state):
# 多重终止条件
if state["iteration"] >= state["max_iterations"]:
return "end"
if state["is_complete"]:
return "end"
if state.get("error_count", 0) >= 3:
return "end"
return "continue"
记录迭代历史
class State(TypedDict):
history: Annotated[List[dict], add] # 追加模式
def process(state):
return {
"history": [{
"iteration": state["iteration"],
"timestamp": datetime.now().isoformat(),
"action": "processed"
}]
}
提供进度反馈
for event in workflow.stream(initial_state):
for node, output in event.items():
print(f"[{node}] 完成 - 迭代 {output.get('iteration', '?')}")
下一步
在下一个教程中，我们将学习如何实现人工干预（Human-in-the-Loop）。