M5 智能体式人工智能 - 客服智能体

正如 Andrew 在讲座中解释的，结合代码执行的规划意味着让大模型编写代码，让代码本身成为计划。 与纯文本或基于 JSON 的计划相比，这种方法表达力更强、也更灵活：代码不仅记录了各个步骤，还可以直接执行它们。

在这个实验中，你将在实践中实现这种设计模式。 我们不会要求大模型以 JSON 格式输出一份计划、再手动执行每一步，而是让它编写 Python 代码，直接体现计划中的多个步骤。通过执行这段代码，我们就可以自动完成复杂的查询。

为了让这一切更具体，我们模拟了一家太阳镜商店，它拥有一份产品库存和一组交易记录（销售、退货、余额更新）。这个例子展示了大模型如何生成代码来查询或更新记录，体现出这种模式的灵活性。

我们将会：

1. 创建简单的库存和交易数据集。
2. 构建一个描述数据的 schema 区块。
3. 提示大模型把计划编写为 Python 代码（并附有解释每一步的注释）。
4. 在沙箱中执行这段代码，得到答案。

完成本实验后，你将能够：

• 解释为什么让模型编写代码（而不是 JSON 或纯文本计划）能实现更丰富、更灵活的规划。
• 提示大模型生成带有分步注释的 Python 代码，既记录计划，又执行计划。
• 在沙箱中安全地运行生成的代码，并解读结果。

这展示了*以代码作为行动（Code as Action）*的方式，是如何胜过脆弱的工具链和基于 JSON 的规划方法的。

# ==== Imports ====
from __future__ import annotations
import json
from dotenv import load_dotenv
from openai import OpenAI
import re, io, sys, traceback, json
from typing import Any, Dict, Optional
from tinydb import Query, where

# Utility modules
import utils      # helper functions for prompting/printing
import inv_utils  # functions for inventory, transactions, schema building, and TinyDB seeding

load_dotenv()
client = OpenAI()

在 inv_utils 模块中，我们提供了如下函数：

• create_inventory() – 构建太阳镜库存。
• create_transactions() – 构建初始交易记录。
• seed_db() – 将库存和交易数据加载到一个基于 JSON 的存储中。
• build_schema_block() – 生成用于提示词中的 schema 描述。
• get_current_balance() 和 next_transaction_id() 等辅助函数——让大模型能够在库存和交易之间进行一致的更新。

我们现在将使用 TinyDB——一个用纯 Python 编写的轻量级文档型数据库——为这个太阳镜商店模拟创建两张小表。 TinyDB 以 JSON 文档的形式存储数据，非常适合小型应用或原型，因为它无需搭建服务器，且能轻松地查询和更新数据。

这两张表分别是：

• inventory_tbl：包含产品详情，例如名称、商品 ID、描述、库存数量和价格。
• transactions_tbl：以一个期初余额开始，之后会记录购买、退货和调整。

你将使用 inv_utils 中的辅助函数生成这些表，然后在下方预览前几行数据。

db, inventory_tbl, transactions_tbl = inv_utils.seed_db()

现在，你可以通过将每张表以格式化的 JSON 打印出来，检查其中的记录：

utils.print_html(json.dumps(inventory_tbl.all(), indent=2), title="Inventory Table")
utils.print_html(json.dumps(transactions_tbl.all(), indent=2), title="Transactions Table")

如上所示，各表的 schema 如下：

在 schema 明确之后，你将构建提示词，指示模型通过编写代码来做规划，然后执行这段代码。正如 Andrew 所强调的，代码即计划：模型在注释中解释每一步，然后加以执行。下方的提示词还会让模型自行判断该请求是只读操作，还是状态变更操作，并强制执行安全的执行方式（无 I/O、无网络、只使用 TinyDB Query、保持变更的一致性）。

PROMPT = """You are a senior data assistant. PLAN BY WRITING PYTHON CODE USING TINYDB.

Database Schema & Samples (read-only):
{schema_block}

Execution Environment (already imported/provided):
- Variables: db, inventory_tbl, transactions_tbl  # TinyDB Table objects
- Helpers: get_current_balance(tbl) -> float, next_transaction_id(tbl, prefix="TXN") -> str
- Natural language: user_request: str  # the original user message

PLANNING RULES (critical):
- Derive ALL filters/parameters from user_request (shape/keywords, price ranges "under/over/between", stock mentions,
  quantities, buy/return intent). Do NOT hard-code values.
- Build TinyDB queries dynamically with Query(). If a constraint isn't in user_request, don't apply it.
- Be conservative: if intent is ambiguous, do read-only (DRY RUN).

TRANSACTION POLICY (hard):
- Do NOT create aggregated multi-item transactions.
- If the request contains multiple items, create a separate transaction row PER ITEM.
- For each item:
  - compute its own line total (unit_price * qty),
  - insert ONE transaction with that amount,
  - update balance sequentially (balance += line_total),
  - update the item's stock.
- If any requested item lacks sufficient stock, do NOT mutate anything; reply with STATUS="insufficient_stock".

HUMAN RESPONSE REQUIREMENT (hard):
- You MUST set a variable named `answer_text` (type str) with a short, customer-friendly sentence (1–2 lines).
- This sentence is the only user-facing message. No dataframes/JSON, no boilerplate disclaimers.
- If nothing matches, politely say so and offer a nearby alternative (closest style/price) or a next step.

ACTION POLICY:
- If the request clearly asks to change state (buy/purchase/return/restock/adjust):
    ACTION="mutate"; SHOULD_MUTATE=True; perform the change and write a matching transaction row.
  Otherwise:
    ACTION="read"; SHOULD_MUTATE=False; simulate and explain briefly as a dry run (in logs only).

FAILURE & EDGE-CASE HANDLING (must implement):
- Do not capture outer variables in Query.test. Pass them as explicit args.
- Always set a short `answer_text`. Also set a string `STATUS` to one of:
  "success", "no_match", "insufficient_stock", "invalid_request", "unsupported_intent".
- no_match: No items satisfy the filters → suggest the closest in style/price, or invite a different range.
- insufficient_stock: Item found but stock < requested qty → state available qty and offer the max you can fulfill.
- invalid_request: Unable to parse essential info (e.g., quantity for a purchase/return) → ask for the missing piece succinctly.
- unsupported_intent: The action is outside the store's capabilities → provide the nearest supported alternative.
- In all cases, keep the tone helpful and concise (1–2 sentences). Put technical details (e.g., ACTION/DRY RUN) only in stdout logs.

OUTPUT CONTRACT:
- Return ONLY executable Python between these tags (no extra text):
  <execute_python>
  # your python
  </execute_python>

CODE CHECKLIST (follow in code):
1) Parse intent & constraints from user_request (regex ok).
2) Build TinyDB condition incrementally; query inventory_tbl.
3) If mutate: validate stock, update inventory, insert a transaction (new id, amount, balance, timestamp).
4) ALWAYS set:
   - `answer_text` (human sentence, required),
   - `STATUS` (see list above).
   Also print a brief log to stdout, e.g., "LOG: ACTION=read DRY_RUN=True STATUS=no_match".
5) Optional: set `answer_rows` or `answer_json` if useful, but `answer_text` is mandatory.

TONE EXAMPLES (for `answer_text`):
- success: "Yes, we have our Classic sunglasses, a round frame, for $60."
- no_match: "We don't have round frames under $100 in stock right now, but our Moon round frame is available at $120."
- insufficient_stock: "We only have 1 pair of Classic left; I can reserve that for you."
- invalid_request: "I can help with that—how many pairs would you like to purchase?"
- unsupported_intent: "We can't refurbish frames, but I can suggest similar new models."

Constraints:
- Use TinyDB Query for filtering. Standard library imports only if needed.
- Keep code clear and commented with numbered steps.

User request:
{question}
"""

让我们生成即计划本身的代码。

我们不会要求模型以 JSON 输出一份计划、再用大量微型工具逐步运行它，而是让它编写能承载整个计划的 Python 代码（例如“先筛选这个，再计算那个，再更新这一行”）。函数 generate_llm_code：

1. 从 inventory_tbl 和 transactions_tbl 动态构建 schema，让模型看到真实的字段、类型和示例。
2. 用这份 schema 加上用户的问题格式化提示词。
3. 调用模型，生成一份“计划即代码”的响应——通常是一个 <execute_python>...</execute_python> 代码块，其中包含分步逻辑。
4. 返回完整的响应内容（包括计划和代码）。 我们在这一步不会执行任何代码。

为什么采用这种模式？让我们把 Python/TinyDB 当作模型早已“熟知”的丰富工具箱来使用，这样它就可以直接用代码组合出多步骤的解决方案，而不必依赖一套不断膨胀的专用工具。我们会在后面的步骤中提取并运行这段代码。

# ---------- 1) Code generation ----------
def generate_llm_code(
    prompt: str,
    *,
    inventory_tbl,
    transactions_tbl,
    model: str = "gpt-4.1-mini",
    temperature: float = 0.2,
) -> str:
    """
    Ask the LLM to produce a plan-with-code response.
    Returns the FULL assistant content (including surrounding text and tags).
    The actual code extraction happens later in execute_generated_code.
    """
    schema_block = inv_utils.build_schema_block(inventory_tbl, transactions_tbl)
    prompt = PROMPT.format(schema_block=schema_block, question=prompt)

    resp = client.chat.completions.create(
        model=model,
        temperature=temperature,
        messages=[
            {
                "role": "system",
                "content": "You write safe, well-commented TinyDB code to handle data questions and updates."
            },
            {"role": "user", "content": prompt},
        ],
    )
    content = resp.choices[0].message.content or ""
    
    return content  

我们将使用 Andrew 在讲座中用过的同一条提示词：

提示词： “Do you have any round sunglasses in stock that are under $100?”（你们有库存中价格低于 100 美元的圆框太阳镜吗？）

在生成任何代码之前，让我们先手动检查一下 TinyDB 表，看看是否真的有*圆形（round）*款式（仅按单词匹配），以及它们的价格如何。运行下一个单元格，预览库存内容，并高亮显示符合“round”这个纯单词筛选条件的商品。

Item = Query()                    # Create a Query object to reference fields (e.g., Item.name, Item.description)

# Search the inventory table for documents where either the description OR the name
# contains the word "round" (case-insensitive). The check is done inline:
# - (v or "") ensures we handle None by converting it to an empty string
# - .lower() normalizes case
# - " round " enforces a crude word boundary (won't match "wraparound")
round_sunglasses = inventory_tbl.search(
    (Item.description.test(lambda v: " round " in ((v or "").lower()))) |
    (Item.name.test(        lambda v: " round " in ((v or "").lower())))
)

# Render the results as formatted JSON in the notebook UI
utils.print_html(json.dumps(round_sunglasses, indent=2), title="Inventory Status: Round Sunglasses")

太好了——我们确实有圆框款式。从手动检查来看，库存中有两款圆框太阳镜，但只有一款价格低于 100 美元。因此，满足这个要求的商品是：

{
  "item_id": "SG005",
  "name": "Classic",
  "description": "Classic round profile with minimalist metal frames, offering a timeless and versatile style that fits both casual and formal wear.",
  "quantity_in_stock": 10,
  "price": 60
}

现在，让我们请模型生成一份代码形式的计划来回答 Andrew 的这条提示词（暂时不执行）。

# Andrew's prompt from the lecture
prompt_round = "Do you have any round sunglasses in stock that are under $100?"

# Generate the plan-as-code (FULL content; may include <execute_python> tags)
full_content_round = generate_llm_code(
    prompt_round,
    inventory_tbl=inventory_tbl,
    transactions_tbl=transactions_tbl,
    model="o4-mini",
    temperature=1.0,
)

# Inspect the LLM's plan + code (no execution here)
utils.print_html(full_content_round, title="Plan with Code (Full Response)")

现在，我们将定义那个接收模型生成的计划并安全运行它的函数：

• 它既能接受带有 <execute_python>…</execute_python> 标签的完整大模型响应，也能接受原始的 Python 代码。
• 它会在需要时提取出可执行的代码块。
• 它会在一个受控的命名空间（仅包含 TinyDB 表和安全的辅助函数）中运行这段代码。
• 它会捕获 stdout、错误信息，以及模型设置的答案变量（answer_text、answer_rows、answer_json）。
• 它会渲染执行前后的表格快照，让副作用变得清晰可见。

这就是把代码形式的计划转化为实际操作和一句简明的用户可见答案的“执行器”。

# --- Helper: extract code between <execute_python>...</execute_python> ---
def _extract_execute_block(text: str) -> str:
    """
    Returns the Python code inside <execute_python>...</execute_python>.
    If no tags are found, assumes 'text' is already raw Python code.
    """
    if not text:
        raise RuntimeError("Empty content passed to code executor.")
    m = re.search(r"<execute_python>(.*?)</execute_python>", text, re.DOTALL | re.IGNORECASE)
    return m.group(1).strip() if m else text.strip()


# ---------- 2) Code execution ----------
def execute_generated_code(
    code_or_content: str,
    *,
    db,
    inventory_tbl,
    transactions_tbl,
    user_request: Optional[str] = None,
) -> Dict[str, Any]:
    """
    Execute code in a controlled namespace.
    Accepts either raw Python code OR full content with <execute_python> tags.
    Returns minimal artifacts: stdout, error, and extracted answer.
    """
    # Extract code here (now centralized)
    code = _extract_execute_block(code_or_content)

    SAFE_GLOBALS = {
        "Query": Query,
        "get_current_balance": inv_utils.get_current_balance,
        "next_transaction_id": inv_utils.next_transaction_id,
        "user_request": user_request or "",
    }
    SAFE_LOCALS = {
        "db": db,
        "inventory_tbl": inventory_tbl,
        "transactions_tbl": transactions_tbl,
    }

    # Capture stdout from the executed code
    _stdout_buf, _old_stdout = io.StringIO(), sys.stdout
    sys.stdout = _stdout_buf
    err_text = None
    try:
        exec(code, SAFE_GLOBALS, SAFE_LOCALS)
    except Exception:
        err_text = traceback.format_exc()
    finally:
        sys.stdout = _old_stdout
    printed = _stdout_buf.getvalue().strip()

    # Extract possible answers set by the generated code
    answer = (
        SAFE_LOCALS.get("answer_text")
        or SAFE_LOCALS.get("answer_rows")
        or SAFE_LOCALS.get("answer_json")
    )


    return {
        "code": code,            # <- ya sin etiquetas
        "stdout": printed,
        "error": err_text,
        "answer": answer,
        "transactions_tbl": transactions_tbl.all(),  # For inspection
        "inventory_tbl": inventory_tbl.all(),  # For inspection
    }

你已经检查过货架，确认恰好有一款圆框款式的价格低于 100 美元。现在轮到最有意思的部分了：把模型生成的这份代码形式的计划交给我们的执行器，看它是如何完成这项工作的。执行器会取出 <code><execute_python>...</execute_python></code> 代码块，在一个锁定的沙箱中运行它，然后向你展示所有重要信息——表格中发生了哪些变化（前后对比）、计划打印出的任何日志，以及最终那句面向客户的 answer_text。

# Execute the generated plan for the round-sunglasses question
result = execute_generated_code(
    full_content_round,          # the full LLM response you generated earlier
    db=db,
    inventory_tbl=inventory_tbl,
    transactions_tbl=transactions_tbl,
    user_request=prompt_round, # e.g., "Do you have any round sunglasses in stock that are under $100?"
)

# Peek at exactly what Python the plan executed
utils.print_html(result["answer"], title="Plan Execution · Extracted Answer")

正如你所见，这正是我们之前手动分析所预期的结果。

在上一步中，你只是查询了数据，因此库存和交易记录都没有发生变化。 现在，让我们用“规划即代码”的模式来处理一个退货场景：

请求： “Return 2 Aviator sunglasses I bought last week.”（退掉我上周买的 2 副 Aviator 太阳镜。）

在生成计划之前，让我们先检查一下 Aviator 款式的当前库存。

Item = Query()                    # Create a Query object to reference fields (e.g., Item.name, Item.description)

# Query: fetch all inventory rows whose 'name' is exactly "Aviator".
# Notes:
# - This is a case-sensitive equality check. "aviator" won't match.
# - If you need case-insensitive matching, consider a .test(...) or .matches(...) with re.I.
aviators = inventory_tbl.search(
    (Item.name == "Aviator")
)

# Display the matched documents in a readable JSON panel
utils.print_html(json.dumps(aviators, indent=2), title="Inventory status: Aviator sunglasses before return")

库存确认有一款 Aviator 商品在库——SG001（Aviator）：23 件，单价 80 美元。现在让我们生成一份计划来回答这条提示词：

prompt_aviator = "Return 2 Aviator sunglasses I bought last week."

# Generate the plan-as-code (FULL content; may include <execute_python> tags)
full_content_aviator = generate_llm_code(
    prompt_aviator,
    inventory_tbl=inventory_tbl,
    transactions_tbl=transactions_tbl,
    model="o4-mini",
    temperature=1,
)

# Inspect the LLM's plan + code (no execution here)
utils.print_html(full_content_aviator, title="Plan with Code (Full Response)")

在执行这份计划之前，让我们先检查一下交易记录的当前状态。

utils.print_html(json.dumps(transactions_tbl.all(), indent=2), title="Transactions Table Before Return")

交易日志目前只显示一条记录——即期初余额条目（TXN001），金额为 $500.00，记录于 2025-10-03T09:16:59.628898。

准备好了——运行下方单元格来执行这份计划。

# Execute the generated plan for the round-sunglasses question
result = execute_generated_code(
    full_content_aviator,          # the full LLM response you generated earlier
    db=db,
    inventory_tbl=inventory_tbl,
    transactions_tbl=transactions_tbl,
    user_request=prompt_aviator, # e.g., "Return 2 aviator sunglasses I bought last week."
)

# Peek at exactly what Python the plan executed
utils.print_html(result["answer"], title="Plan Execution · Extracted Answer")

你可以在下方看到，一条针对 Aviator 太阳镜退货的新交易记录已经被插入。

utils.print_html(json.dumps(transactions_tbl.all(), indent=2), title="Transactions Table After Return")

运行下方单元格，你会看到 Aviator 的库存增加到了 25（quantity_in_stock）。

Item = Query()                  

aviators = inventory_tbl.search(
    (Item.name == "Aviator")
)

utils.print_html(json.dumps(aviators, indent=2), title="Inventory status: Aviator sunglasses after return")

你已经搭建好了各个组成部分——schema、提示词、代码生成器和执行器。现在，让我们把它们组合成一个统一的辅助函数：接收一条自然语言请求，生成一份代码形式的计划，安全地执行它，并展示结果（连同执行前后的表格）。

这个智能体会做什么

• 可选地为一次干净的运行重新填充演示数据。
• 生成计划（Python 代码，包裹在 <execute_python>…</execute_python> 中）。
• 在一个受控的命名空间（TinyDB + 辅助函数）中执行该计划。
• 展示一句简明的 answer_text，并渲染执行前后的快照。

def customer_service_agent(
    question: str,
    *,
    db,
    inventory_tbl,
    transactions_tbl,
    model: str = "o4-mini",
    temperature: float = 1.0,
    reseed: bool = False,
) -> dict:
    """
    End-to-end helper:
      1) (Optional) reseed inventory & transactions
      2) Generate plan-as-code from `question`
      3) Execute in a controlled namespace
      4) Render before/after snapshots and return artifacts

    Returns:
      {
        "full_content": <raw LLM response (may include <execute_python> tags)>,
        "exec": {
            "code": <extracted python>,
            "stdout": <plan logs>,
            "error": <traceback or None>,
            "answer": <answer_text/rows/json>,
            "inventory_after": [...],
            "transactions_after": [...]
        }
      }
    """
    # 0) Optional reseed
    if reseed:
        inv_utils.create_inventory()
        inv_utils.create_transactions()

    # 1) Show the question
    utils.print_html(question, title="User Question")

    # 2) Generate plan-as-code (FULL content)
    full_content = generate_llm_code(
        question,
        inventory_tbl=inventory_tbl,
        transactions_tbl=transactions_tbl,
        model=model,
        temperature=temperature,
    )
    utils.print_html(full_content, title="Plan with Code (Full Response)")

    # 3) Before snapshots
    utils.print_html(json.dumps(inventory_tbl.all(), indent=2), title="Inventory Table · Before")
    utils.print_html(json.dumps(transactions_tbl.all(), indent=2), title="Transactions Table · Before")

    # 4) Execute
    exec_res = execute_generated_code(
        full_content,
        db=db,
        inventory_tbl=inventory_tbl,
        transactions_tbl=transactions_tbl,
        user_request=question,
    )

    # 5) After snapshots + final answer
    utils.print_html(exec_res["answer"], title="Plan Execution · Extracted Answer")
    utils.print_html(json.dumps(inventory_tbl.all(), indent=2), title="Inventory Table · After")
    utils.print_html(json.dumps(transactions_tbl.all(), indent=2), title="Transactions Table · After")

    # 6) Return artifacts
    return {
        "full_content": full_content,
        "exec": {
            "code": exec_res["code"],
            "stdout": exec_res["stdout"],
            "error": exec_res["error"],
            "answer": exec_res["answer"],
            "inventory_after": inventory_tbl.all(),
            "transactions_after": transactions_tbl.all(),
        },
    }

使用 customer_service_agent(...) 这个辅助函数，走完从自然语言请求 → 代码形式的计划 → 安全执行 → 执行前后快照的完整流程。

试试这些提示词：

1. 只读（Andrew 的示例）： “Do you have any round sunglasses in stock that are under $100?”
2. 变更 — 退货： “Return 2 Aviator sunglasses.”
3. 变更 — 购买： “Purchase 3 Wayfarer sunglasses for customer Alice.”
4. 变更 - 购买多件商品： "I want to buy 3 pairs of classic sunglasses and 1 pair of aviator."

prompt = "I want to buy 3 pairs of classic sunglasses and 1 pair of aviator sunglasses."

out = customer_service_agent(
    prompt,
    db=db,
    inventory_tbl=inventory_tbl,
    transactions_tbl=transactions_tbl,
    model="o4-mini",
    temperature=1.0,
    reseed=True,   # set False to keep current state of the inventory and the transactions
)

• 你让代码成为了计划本身。 沿着 Andrew “以代码作为行动”的理念，你让模型编写出把各个步骤（筛选 → 计算 → 更新）串联起来的 Python 代码，然后你只需要运行它。

• 你绕开了脆弱的工具堆砌方式。 你没有堆砌大量微型工具或使用 JSON 计划，而是使用了 Python/TinyDB——为模型提供了一个庞大而熟悉的工具箱，能用一条提示词处理多种查询形态。

• 你让运行过程既安全又可见。 你在一个受控的命名空间中执行代码，捕获日志/错误，并审查执行前后的表格——这样你就始终清楚发生了什么变化，以及为什么会发生。

🎉 <strong>恭喜！</strong>

你刚刚完成了这个实验，并构建了一个<em>智能体式</em>客服工作流。你让模型把代码写成计划，安全地运行它，并使用简单的校验来保持更新的可靠性。当出现失败时，你展示了清晰、人类可读的原因；当一切顺利时，你通过前后快照准确地看到了发生了什么变化。

掌握了这种模式——在代码<em>中</em>进行规划，加上透明的执行过程——你已经准备好设计属于你自己的、感觉自动、安全且易于扩展的工作流了。🚀