AI Agent 实战:让大模型真正干活(ReAct + 工具调用)

🎯 本节目标

学完这一节,你能:

判断客户场景该用 Agent 还是确定性工作流
搭一个询价邮件自动处理 Agent(蚂蚁 FDE 访谈原型)
加 Human-in-the-loop(低置信度转人工)

🍎 生活化类比

RAG = 查资料回答问题的学生。Agent = 能自己查资料、打电话、填表格办事的实习生。

Agent 多了两样东西:

工具调用:能执行动作(发邮件、查数据库、调 API)
自主决策:能判断"这步该用哪个工具""做完没""要不要重来"

📖 概念讲解

Agent vs 工作流:怎么选

关键问题:步骤能不能提前写死?

能提前写死步骤?	用什么
能(固定流程)	工作流(Dify Workflow / LangGraph 确定图)
不能(AI 要自己判断下一步)	Agent(ReAct / Function Calling 循环)

FDE 经验:90% 的企业场景该用工作流,Agent 只用于真正需要灵活判断的。盲目上 Agent = 不可控 + 贵。

ReAct 循环

Agent 的核心是 Reason + Act 循环:

思考(Reason):我现在该干嘛?
行动(Act):调一个工具
观察(Observe):看工具返回什么
重复,直到完成

🛠️ 动手实操:询价邮件 Agent(蚂蚁原型)

场景:收到询价邮件 → 抽取产品/数量 → 查库存 → 报价 → 低置信度转人工。

用模块 3 的 LangGraph,完整代码见实战项目目录。核心结构:

# 节点
extract_node   # 抽取邮件信息
inventory_node # 查库存(API 模拟)
quote_node     # 生成报价
human_node     # 低置信度转人工审批

# 条件边
confidence > 0.8 → 自动报价
confidence < 0.8 → 转人工 → 审批后报价

✅ 这就是蚂蚁 FDE 访谈里"询价邮件自动处理"的最小生产版。

⚠️ 常见错误

问题	原因	解决
Agent 乱调工具,成本爆炸	没限制工具调用次数	设 max_iterations,超过就强制返回
Agent 陷入死循环	同一步骤反复执行	加重复检测,连续 3 次相同动作就退出
该用工作流却上了 Agent	以为 Agent 更"高级"	步骤能写死就用工作流(确定+便宜),Agent 只给真需灵活判断的

✅ 自测题

<summary>1. 判断:客户要"按固定 5 步流程处理订单",该用 Agent 还是工作流?</summary> **工作流**。步骤能提前写死,就该用确定性工作流(更稳、更便宜)。Agent 只在"AI 要自己判断下一步"时才用。 </details> <summary>2. 生产环境 Agent 为什么必须加 HITL?</summary> 因为模型会误判。低置信度的决策(如直接扣款、发邮件给客户)必须转人工审批,否则出错无法挽回。HITL 是生产 Agent 的安全网。 </details>

📚 总结

Agent = RAG + 工具调用 + 自主决策。优先用工作流,灵活判断才上 Agent,生产必须加 HITL。

🏋️ 配套自练

CrewAI 入门 — 30k★ 的入门 Agent 框架,<50 行出多 Agent(免费)
LangGraph Multi-Agent 教程 — 官方多 Agent 协作教程(免费)
把本节的"询价邮件 Agent"用 CrewAI 重写一遍,对比和 LangGraph 的差异

🔗 延伸阅读

← 上一节:RAG 进阶 | 下一节:评估方法论 →

动手实验

实验改编自 ai-engineer-notebooks（MIT License）第 05 章 Agent Loop from Scratch（01-agent-loop-from-scratch.ipynb）· github.com/calmrocks/ai-engineer-notebooks 运行方式：Colab 或本地 Jupyter 任选；模型用 Groq 免费 API（OpenAI 兼容），换成任何 OpenAI 兼容端点（含公司内部网关）只需改 base_url 与 key。

实验目标

不用框架，用约 40 行原生 API 调用写出一个能跑的 agent loop：模型请求工具 → 本地执行 → 结果按 tool_call_id 塞回 messages → 循环，直到任务完成或撞到轮数上限。对应本课时讲义「agent = 模型 + 工具 + 循环 + 停止条件」与 ReAct 式「边想边做」的执行轨迹。

前置条件

免费 GROQ_API_KEY（console.groq.com）。
Colab 或本地 Jupyter，仅需 groq 客户端（aien 包自带）。
建议先看懂单轮 tool calling——本实验是它的循环版。

实验步骤

步骤 1：安装与初始化

%pip install -q "git+https://github.com/calmrocks/ai-engineer-notebooks.git@v0.1.0"

from aien import setup
client, MODEL = setup()

步骤 2：工具的「实现」与「定义」两半

模型永远看不到你的 Python 函数，只看到 name + description + JSON schema。先写实现（内存文件系统 + 计算器）：

import json, re
FS = {  # 内存文件系统：安全、可检视、可重置
    "notes/monday.md": "Team lunch: $42.50\nTaxi to client site: $18.00\n",
    "notes/tuesday.md": "Conference tickets, 2 x $150.00 each\n",
    "notes/wednesday.md": "Cloud credits: $75.25\nCoffee: $23.10\n",
}

def list_files():
    return "\n".join(sorted(FS)) if FS else "(no files)"

def read_file(path):
    if path not in FS:
        return f"Error: no file at '{path}'."
    return FS[path]

def write_file(path, content):
    FS[path] = content
    return f"Wrote {len(content)} characters to {path}."

def calculator(expression):  # 白名单校验，杜绝 eval 注入
    if not re.fullmatch(r"[0-9+\-*/(). ]+", expression):
        return "Error: only numbers and + - * / ( ) are allowed."
    return str(eval(expression, {"__builtins__": {}}, {}))

def execute_tool(name, args):
    try:
        return str({"list_files": list_files, "read_file": read_file,
                    "write_file": write_file, "calculator": calculator}[name](**args))
    except Exception as e:
        return f"Error: {e}"

再写模型看到的定义（OpenAI 兼容格式，四个工具都要）：

TOOLS = [
 {"type": "function", "function": {"name": "list_files",
  "description": "List every file path in the workspace, one per line. Use this first if you don't know which files exist.",
  "parameters": {"type": "object", "properties": {}, "required": []}}},
 {"type": "function", "function": {"name": "read_file",
  "description": "Read the full contents of one file. Returns the text, or an error if the path doesn't exist.",
  "parameters": {"type": "object",
   "properties": {"path": {"type": "string", "description": "Exact path from list_files."}},
   "required": ["path"]}}},
 {"type": "function", "function": {"name": "write_file",
  "description": "Create or overwrite a file with the given content. Returns a confirmation.",
  "parameters": {"type": "object",
   "properties": {"path": {"type": "string"}, "content": {"type": "string"}},
   "required": ["path", "content"]}}},
 {"type": "function", "function": {"name": "calculator",
  "description": "Evaluate an arithmetic expression using + - * / and parentheses. Use this for all math instead of computing in your head.",
  "parameters": {"type": "object",
   "properties": {"expression": {"type": "string"}}, "required": ["expression"]}}},
]

步骤 3：循环本身

def run_agent(task, tools, max_turns=10):
    """agent = 模型 + 工具 + while 循环 + 停止条件"""
    messages = [{"role": "user", "content": task}]  # messages 就是 agent 的全部状态
    for turn in range(1, max_turns + 1):
        response = client.chat.completions.create(
            model=MODEL, max_tokens=1024, tools=tools, messages=messages)
        choice, msg = response.choices[0], response.choices[0].message
        if choice.finish_reason != "tool_calls":  # 停止条件 1：模型不再要工具
            return msg.content or ""
        messages.append(msg)                      # 原样追加携带 tool_calls 的回复
        for tc in msg.tool_calls:                 # 执行工具，结果按 id 配对回传
            args = json.loads(tc.function.arguments)
            result = execute_tool(tc.function.name, args)
            print(f"[turn {turn}] {tc.function.name}({tc.function.arguments[:60]}) -> {result[:80]}")
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": result})
    return f"Stopped: hit max_turns={max_turns} before the task finished."  # 停止条件 2

max_turns 不是装饰：每一轮都是付费 API 调用，无界循环就是无界账单。trace 打印就是教学法——你能看着模型一步步规划。

步骤 4：跑一个单轮调用做不完的任务

result = run_agent(
    "Read all the notes files in the workspace, add up every expense using the "
    "calculator, and write a file called summary.md that states the total. "
    "Then tell me the total.", tools=TOOLS)
print("\n=== final answer ===\n" + result)
print(FS.get("summary.md", "(summary.md was not written)"))  # 副作用真实发生了

读 trace：模型自己规划出 list → read → calculate → write 的顺序，没人写死它；「2 x $150.00」必须真的喂给 calculator。跑两次，细节会不同但收敛到同一总额——这种非确定性正是后续 guardrails 要管的东西。

自查问题

为什么说 messages 列表是 agent 的全部状态？「每轮重发全部历史」意味着成本随轮数怎么增长？
把 calculator 的 description 改成 "Does math." 再跑三次，模型还会用它吗？工具定义里最值钱的是什么？
步骤能预先写死的任务（fetch → extract → summarize），为什么该写 pipeline 而不是自主 agent？