0.3 控制流与决策逻辑

引言：让 Agent 学会"思考"

一个真正的 AI Agent 不仅仅是简单地执行指令——它需要能够做决策、循环处理任务、根据条件选择不同的路径。这就是控制流的作用。

想象一个客服 Agent：

如果用户问候，回复问候语
否则如果用户提问，搜索知识库
否则如果用户投诉，转接人工
循环处理每一条新消息
根据条件决定是否需要调用外部 API

这一切，都依赖于本节要学习的控制流结构。

🎯 小白理解：什么是"控制流"？
控制流 = 程序执行的"路线图"
想象你在玩一个冒险游戏：
你遇到一个岔路口
    │
    ├── 如果选左边 → 遇到宝藏
    │
    └── 如果选右边 → 遇到怪物
                       │
                       ├── 如果打赢了 → 获得经验
                       │
                       └── 如果打输了 → 重新开始
1
2
3
4
5
6
7
8
9
程序也是这样！它不是傻傻地从头到尾执行，而是会：
做判断（if/else）：走哪条路？
重复做事（循环）：刷怪100次
跳出/跳过（break/continue）：遇到 Boss 就逃跑
没有控制流的程序 = 只会直线前进的 NPC 有控制流的程序 = 会思考、会决策的智能 Agent

学习目标

✅ 掌握 if/elif/else 条件语句
✅ 理解 for 循环和 while 循环
✅ 学会使用 break 和 continue 控制循环
✅ 理解控制流在 LangGraph 状态机中的应用
✅ 构建一个具有决策能力的简单 Agent

第一部分：条件语句——Agent 的决策树

🎯 小白理解：if 语句就是"如果...就..."
生活中我们时刻在做判断：
如果下雨了，就带伞
如果饿了，就吃饭
如果困了，就睡觉
在 Python 中，这种判断用 if 语句来表达：
python
if 下雨了:
    带伞
1
2
注意两个关键点：
if 后面跟的是一个条件（结果是 True 或 False）
条件后面要加冒号 :
要执行的代码要缩进（按4个空格或1个Tab）
缩进很重要！ Python 用缩进来判断哪些代码属于 if 语句：
python
if 条件:
    这行属于 if（缩进了）
    这行也属于 if（缩进了）
这行不属于 if（没缩进）
1
2
3
4

if 语句：单一条件判断

python

temperature: float = 0.9

if temperature > 1.0:
    print("警告: 温度参数过高，可能导致输出不稳定")

if-else 语句：二选一

python

api_key: str | None = None

if api_key:
    print("API 密钥已配置，开始初始化 Agent")
else:
    print("错误: 请先配置 API 密钥")

if-elif-else 语句：多条件分支

python

def route_user_intent(intent: str) -> str:
    """
    根据用户意图路由到不同的处理器
    
    这模拟了 LangGraph 中的条件路由 (conditional routing)
    """
    if intent == "greeting":
        return "处理问候"
    elif intent == "question":
        return "调用 RAG 系统"
    elif intent == "complaint":
        return "转接人工客服"
    elif intent == "goodbye":
        return "结束对话"
    else:
        return "未知意图，请求澄清"

# 测试路由逻辑
print(route_user_intent("question"))    # 调用 RAG 系统
print(route_user_intent("complaint"))   # 转接人工客服
print(route_user_intent("unknown"))     # 未知意图，请求澄清

🔗 与 Agent 的联系：LangGraph 的 add_conditional_edges() 本质上就是根据状态值进行 if/elif/else 判断，决定下一个节点。

嵌套条件：处理复杂逻辑

python

def should_retry(
    attempt_count: int,
    max_retries: int,
    error_type: str
) -> bool:
    """
    决定是否应该重试 API 调用
    
    Args:
        attempt_count: 当前尝试次数
        max_retries: 最大重试次数
        error_type: 错误类型
    
    Returns:
        是否应该重试
    """
    if attempt_count < max_retries:
        if error_type == "rate_limit":
            return True  # 速率限制错误，应该重试
        elif error_type == "timeout":
            return True  # 超时错误，应该重试
        elif error_type == "server_error":
            return True  # 服务器错误，应该重试
        else:
            return False  # 其他错误（如认证失败），不应重试
    else:
        return False  # 已达最大重试次数

# 测试
print(should_retry(2, 3, "rate_limit"))    # True
print(should_retry(3, 3, "rate_limit"))    # False（已达上限）
print(should_retry(1, 3, "auth_error"))    # False（不可重试的错误）

三元表达式：简洁的条件赋值

python

# 传统写法
tokens_used: int = 1800
max_tokens: int = 2000

if tokens_used < max_tokens:
    status = "正常"
else:
    status = "超限"

# 三元表达式（更简洁）
status: str = "正常" if tokens_used < max_tokens else "超限"

# 在函数调用中使用
def get_model(is_premium_user: bool) -> str:
    return "gpt-4" if is_premium_user else "gpt-3.5-turbo"

print(get_model(True))   # gpt-4
print(get_model(False))  # gpt-3.5-turbo

第二部分：for 循环——批量处理的利器

🎯 小白理解：for 循环就是"对于每一个...做..."
想象你是老师，要给全班同学发作业：
没有循环（累死）：
给第1个学生发作业
给第2个学生发作业
给第3个学生发作业
...（写50遍）
给第50个学生发作业
1
2
3
4
5
有循环（轻松）：
对于班上的每一个学生：
    给这个学生发作业
1
2
Python 的 for 循环就是这个意思：
python
students = ["小明", "小红", "小刚"]

for student in students:
    print(f"给 {student} 发作业")

# 输出：
# 给 小明 发作业
# 给 小红 发作业
# 给 小刚 发作业
1
2
3
4
5
6
7
8
9
for 循环会自动：
从列表中取出第1个元素
执行循环体里的代码
取出第2个元素
执行循环体里的代码
...直到所有元素都处理完

遍历列表

python

# Agent 的消息历史
messages: list[str] = [
    "用户: 你好",
    "Agent: 您好！有什么可以帮助您的？",
    "用户: 天气怎么样？",
    "Agent: 让我为您查询天气信息...",
]

# 遍历并打印所有消息
for message in messages:
    print(message)

# 带索引遍历
for index, message in enumerate(messages):
    print(f"[{index}] {message}")

遍历范围

python

# 批量创建 Agent 实例
for agent_id in range(5):
    print(f"初始化 Agent #{agent_id}")

# 输出:
# 初始化 Agent #0
# 初始化 Agent #1
# 初始化 Agent #2
# 初始化 Agent #3
# 初始化 Agent #4

# 指定起始、结束、步长
for i in range(0, 10, 2):  # 0, 2, 4, 6, 8
    print(f"偶数: {i}")

遍历字典

python

# Agent 配置
config: dict[str, str | float | int] = {
    "model": "gpt-4",
    "temperature": 0.7,
    "max_tokens": 2000,
}

# 遍历键
for key in config.keys():
    print(f"配置项: {key}")

# 遍历值
for value in config.values():
    print(f"值: {value}")

# 同时遍历键和值（最常用）
for key, value in config.items():
    print(f"{key}: {value}")

列表推导式：高效创建列表

python

# 传统方式：创建 1-10 的平方列表
squares: list[int] = []
for i in range(1, 11):
    squares.append(i ** 2)

# 列表推导式（更 Pythonic）
squares: list[int] = [i ** 2 for i in range(1, 11)]
print(squares)  # [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]

# 带条件的列表推导式
even_squares: list[int] = [i ** 2 for i in range(1, 11) if i % 2 == 0]
print(even_squares)  # [4, 16, 36, 64, 100]

# AI 应用：批量格式化消息
raw_messages: list[str] = ["hello", "how are you", "goodbye"]
formatted: list[str] = [f"用户: {msg}" for msg in raw_messages]
print(formatted)
# ['用户: hello', '用户: how are you', '用户: goodbye']

第三部分：while 循环——Agent 的重试机制

🎯 小白理解：while 循环就是"只要...就一直..."
for 循环 vs while 循环：
for 循环：我知道要做几次（比如遍历一个列表）
while 循环：我不知道要做几次，只知道什么时候停（比如游戏主循环、等待用户输入）
生活例子：
while 还没吃饱:
    再吃一口饭

while 还没到目的地:
    继续开车

while 用户还在线:
    等待用户消息
1
2
3
4
5
6
7
8
Python 写法：
python
count = 0
while count < 3:
    print(f"第 {count + 1} 次尝试")
    count = count + 1  # 别忘了更新条件！

# 输出：
# 第 1 次尝试
# 第 2 次尝试
# 第 3 次尝试
1
2
3
4
5
6
7
8
9
⚠️ 小心无限循环！ 如果条件永远为 True，程序会一直跑下去：
python
# ❌ 危险！无限循环！
while True:
    print("我会一直打印...")  # 按 Ctrl+C 强制停止
1
2
3

基本 while 循环

python

# 重试机制的简单实现
max_retries: int = 3
attempt: int = 0

while attempt < max_retries:
    print(f"第 {attempt + 1} 次尝试调用 API...")
    attempt += 1
    # 假设 API 调用成功
    success: bool = True
    if success:
        print("API 调用成功！")
        break

实战：Agent 重试逻辑

python

import time
from typing import Optional

def call_llm_with_retry(
    prompt: str,
    max_retries: int = 3,
    retry_delay: float = 1.0
) -> Optional[str]:
    """
    调用 LLM API，失败时自动重试
    
    Args:
        prompt: 提示词
        max_retries: 最大重试次数
        retry_delay: 重试间隔（秒）
    
    Returns:
        LLM 响应，失败返回 None
    """
    attempt: int = 0
    
    while attempt < max_retries:
        try:
            print(f"尝试 #{attempt + 1}...")
            # 模拟 API 调用（实际中会调用 OpenAI/Anthropic API）
            response: str = f"响应: {prompt}"
            return response
            
        except Exception as e:
            attempt += 1
            if attempt < max_retries:
                print(f"错误: {e}. {retry_delay}秒后重试...")
                time.sleep(retry_delay)
            else:
                print(f"达到最大重试次数 ({max_retries})，放弃")
                return None
    
    return None

# 测试
result = call_llm_with_retry("What is AI?")
print(result)

break 和 continue

python

# break: 立即退出循环
for i in range(10):
    if i == 5:
        print("找到目标，停止搜索")
        break
    print(f"检查第 {i} 项")

# continue: 跳过本次迭代，继续下一次
messages: list[str] = ["Hello", "", "World", "", "!"]
for msg in messages:
    if not msg:  # 跳过空消息
        continue
    print(f"处理消息: {msg}")

第四部分：实战案例——构建简单的对话 Agent

让我们综合运用所学知识，构建一个具有决策能力的对话 Agent：

python

"""
简单对话 Agent
演示控制流在 AI Agent 中的应用
"""

from typing import Optional


class SimpleAgent:
    """简单的对话 Agent"""
    
    def __init__(self, name: str, max_turns: int = 10) -> None:
        """
        初始化 Agent
        
        Args:
            name: Agent 名称
            max_turns: 最大对话轮次
        """
        self.name: str = name
        self.max_turns: int = max_turns
        self.conversation_history: list[str] = []
        self.current_turn: int = 0
    
    def process_input(self, user_input: str) -> Optional[str]:
        """
        处理用户输入，返回 Agent 响应
        
        Args:
            user_input: 用户输入
        
        Returns:
            Agent 响应
        """
        # 检查对话轮次限制
        if self.current_turn >= self.max_turns:
            return "已达对话轮次上限，会话结束。"
        
        # 记录用户输入
        self.conversation_history.append(f"用户: {user_input}")
        self.current_turn += 1
        
        # 意图识别（简化版）
        intent: str = self._classify_intent(user_input)
        
        # 根据意图生成响应
        response: str = self._generate_response(intent, user_input)
        
        # 记录 Agent 响应
        self.conversation_history.append(f"Agent: {response}")
        
        return response
    
    def _classify_intent(self, text: str) -> str:
        """
        分类用户意图（简化实现）
        
        实际应用中会使用 LLM 或分类模型
        """
        text_lower: str = text.lower()
        
        if any(word in text_lower for word in ["你好", "hello", "hi"]):
            return "greeting"
        elif any(word in text_lower for word in ["再见", "bye", "goodbye"]):
            return "goodbye"
        elif "天气" in text_lower:
            return "weather_query"
        elif "?" in text or "？" in text or "什么" in text_lower:
            return "question"
        else:
            return "general"
    
    def _generate_response(self, intent: str, user_input: str) -> str:
        """
        根据意图生成响应
        
        这模拟了 LangGraph 的条件路由
        """
        if intent == "greeting":
            return f"您好！我是 {self.name}，很高兴为您服务。"
        
        elif intent == "goodbye":
            return "再见！期待下次为您服务。"
        
        elif intent == "weather_query":
            return "正在为您查询天气信息...（这里会调用天气 API）"
        
        elif intent == "question":
            return f"您问: '{user_input}'。让我思考一下...（这里会调用 LLM）"
        
        else:
            return "我理解了。还有什么我可以帮助您的吗？"
    
    def get_conversation_summary(self) -> str:
        """获取对话摘要"""
        return "\n".join(self.conversation_history)


def main() -> None:
    """主函数：演示 Agent 对话流程"""
    agent = SimpleAgent(name="助手小A", max_turns=5)
    
    # 模拟对话
    test_inputs: list[str] = [
        "你好",
        "今天天气怎么样？",
        "什么是人工智能？",
        "谢谢",
        "再见",
    ]
    
    print(f"=== 开始对话 (最多 {agent.max_turns} 轮) ===\n")
    
    for user_input in test_inputs:
        print(f"用户: {user_input}")
        response = agent.process_input(user_input)
        
        if response:
            print(f"{agent.name}: {response}\n")
        
        # 如果是告别，结束对话
        if "再见" in response or "goodbye" in response.lower():
            break
    
    print("=== 对话历史 ===")
    print(agent.get_conversation_summary())


if __name__ == "__main__":
    main()

运行结果：

=== 开始对话 (最多 5 轮) ===

用户: 你好
助手小A: 您好！我是 助手小A，很高兴为您服务。

用户: 今天天气怎么样？
助手小A: 正在为您查询天气信息...（这里会调用天气 API）

用户: 什么是人工智能？
助手小A: 您问: '什么是人工智能？'。让我思考一下...（这里会调用 LLM）

用户: 谢谢
助手小A: 我理解了。还有什么我可以帮助您的吗？

用户: 再见
助手小A: 再见！期待下次为您服务。

=== 对话历史 ===
用户: 你好
Agent: 您好！我是 助手小A，很高兴为您服务。
用户: 今天天气怎么样？
Agent: 正在为您查询天气信息...（这里会调用天气 API）
用户: 什么是人工智能？
Agent: 您问: '什么是人工智能？'。让我思考一下...（这里会调用 LLM）
用户: 谢谢
Agent: 我理解了。还有什么我可以帮助您的吗？
用户: 再见
Agent: 再见！期待下次为您服务。

控制流与 LangGraph 的联系

在 LangGraph 中，控制流体现在：

1. 条件路由（Conditional Routing）

python

# LangGraph 中的条件边
workflow.add_conditional_edges(
    "agent",
    lambda x: "tools" if x["needs_tool"] else "end",
    {
        "tools": "call_tools",
        "end": END
    }
)

# 等价于
if state["needs_tool"]:
    next_node = "call_tools"
else:
    next_node = END

2. 循环执行（Looping）

python

# LangGraph 的循环
workflow.add_edge("tools", "agent")  # 工具调用后返回 agent

# 等价于
while not task_completed:
    agent_output = call_agent()
    if needs_tool(agent_output):
        tool_result = call_tools()
    else:
        break

本节总结

核心概念

if/elif/else: Agent 的决策分支
for 循环: 批量处理消息、工具调用
while 循环: 重试机制、持续监听
break/continue: 精确控制循环流程
控制流 = Agent 的"大脑"

最佳实践

✅ 使用明确的条件判断，避免复杂嵌套
✅ 循环中始终考虑退出条件，防止无限循环
✅ 使用列表推导式让代码更简洁
✅ 在 while 循环中设置最大迭代次数

下一节：0.4 小结和复习

0.3 控制流与决策逻辑 ​

引言：让 Agent 学会"思考" ​

学习目标 ​

第一部分：条件语句——Agent 的决策树 ​

if 语句：单一条件判断 ​

if-else 语句：二选一 ​

if-elif-else 语句：多条件分支 ​

嵌套条件：处理复杂逻辑 ​

三元表达式：简洁的条件赋值 ​

第二部分：for 循环——批量处理的利器 ​

遍历列表 ​

遍历范围 ​

遍历字典 ​

列表推导式：高效创建列表 ​

第三部分：while 循环——Agent 的重试机制 ​

基本 while 循环 ​

实战：Agent 重试逻辑 ​

break 和 continue ​

第四部分：实战案例——构建简单的对话 Agent ​

控制流与 LangGraph 的联系 ​

1. 条件路由（Conditional Routing） ​

2. 循环执行（Looping） ​

本节总结 ​

核心概念 ​

最佳实践 ​

0.3 控制流与决策逻辑

引言：让 Agent 学会"思考"

学习目标

第一部分：条件语句——Agent 的决策树

if 语句：单一条件判断

if-else 语句：二选一

if-elif-else 语句：多条件分支

嵌套条件：处理复杂逻辑

三元表达式：简洁的条件赋值

第二部分：for 循环——批量处理的利器

遍历列表

遍历范围

遍历字典

列表推导式：高效创建列表

第三部分：while 循环——Agent 的重试机制

基本 while 循环

实战：Agent 重试逻辑

break 和 continue

第四部分：实战案例——构建简单的对话 Agent

控制流与 LangGraph 的联系

1. 条件路由（Conditional Routing）

2. 循环执行（Looping）

本节总结

核心概念

最佳实践