Skip to content

11. Claude式 LLM 与 Agent 案例分析

版本:v1.0

最后更新:2026-07-09

适用对象:想把 LLMAgent 的区别真正讲清楚,并且希望拿到一组可以直接改成自己项目代码骨架的产品、研发、平台、架构与面试准备同学

1. 为什么要把 LLM 案例和 Agent 案例放在同一篇里

很多团队做项目时最容易犯的第一个错误,不是代码写错,而是:

  • 明明只需要一次结构化判断,却先上了 Agent
  • 明明已经需要工具循环和审批恢复,却还在硬撑单轮 LLM 调用

所以这篇案例分析不打算先讲“怎么写一个很炫的 Agent”,而是先把两类系统并排放在一起看:

形态更适合什么问题典型特征
LLM 工作流节点单轮理解、抽取、分类、改写、排序、评分一次调用为主,外部状态少,输出结构稳定
Agent需要多步推理、工具调用、读写边界、审批与恢复多轮循环、工具面控制、状态续接、可观测与补偿

一句话记忆:

  • LLM 更像“高能力判断节点”
  • Agent 更像“带工具和状态的控制循环”

这也是为什么很多生产系统最后都不是“纯 LLM”或“纯 Agent”,而是:

  • 在工作流里放一个或多个 LLM 节点
  • 只在真正需要时,才进入 Agent loop

2. 这篇案例借用了哪些 Claude 原理

下面这些点,主要参考 Anthropic 在 2026-07-09 可访问的官方文档,包括 Using the Messages APITool use with ClaudeStop reasons and fallbackPrompt cachingExtended thinking

2.1 Messages API 是无状态的,所以状态要靠你自己续

Claude 的 Messages API 并不会自动记住上一次调用。你每次都要主动带上:

  • 历史消息
  • 工具结果
  • 审批结果
  • 需要保留的上下文摘要

这件事的工程含义很大,因为它决定了:

  • 什么时候该保留完整历史
  • 什么时候该压缩成摘要
  • 哪些状态只留在本地运行时,哪些状态真的要喂回模型

2.2 Claude 负责发起 tool_use,真正执行工具的是你的应用

Anthropic 官方工具文档把这一层分得很清楚:

  • client tools 由你的应用执行
  • server tools 由 Anthropic 基础设施执行

这意味着最关键的控制面仍然在你手里:

  • 哪些工具可见
  • 哪些工具只读
  • 哪些工具必须审批
  • 哪些结果允许回灌给模型

2.3 stop_reason 不是附加字段,而是控制循环的主信号

Claude 每次返回都会带 stop_reason。这对工程实现非常关键,因为它决定下一步到底该:

  • 直接把结果交给用户
  • 继续工具循环
  • 续跑
  • 降级
  • 拒绝

最常见的几个分支通常包括:

  • tool_use
  • end_turn
  • max_tokens
  • pause_turn
  • refusal

如果你忽略它,很多“模型怎么突然不对了”的问题,本质上只是:

  • 运行时根本没正确处理停机原因

2.4 稳定前缀应该缓存,而不是每次重算

Anthropic 当前 prompt caching 文档强调,反复出现的大前缀内容很适合缓存,例如:

  • 固定系统提示词
  • 工具定义
  • 长政策文本
  • 大块知识背景

这对案例里的两个系统都很重要,因为真正昂贵的往往不是用户那一句输入,而是:

  • 每轮都重复发送的系统规则和工具面

2.5 写操作最好保持“先读后写、先草稿后审批”

这不是某个厂商专属魔法,而是更稳的工程习惯。

在 Claude 这类 tool use loop 里,这个习惯尤其重要,因为它会直接降低:

  • 误调用写工具
  • 循环中途副作用失控
  • 审批与审计无法对账

所以这篇案例会刻意把第二个 Agent 示例做成:

  1. 先只开放只读工具
  2. 先拿到证据和结论草稿
  3. 审批通过后再开放写工具

3. 先看一个最常见判断:什么时候其实不该上 Agent

如果任务满足下面这些条件,通常先做 LLM 节点 更稳:

  • 输入短
  • 输出结构明确
  • 不依赖多步外部查询
  • 不需要长期状态
  • 不需要真实副作用

典型例子:

  • 工单分类
  • 工单优先级判断
  • 实体抽取
  • 结构化摘要
  • 风险标签判断

如果任务满足下面这些条件,才更像 Agent

  • 需要多步查证
  • 需要调用多个工具
  • 需要把工具结果继续送回模型
  • 需要审批或人工接管
  • 需要长任务恢复或回放

典型例子:

  • 事故分诊
  • 工具编排
  • 多系统巡检
  • 审批型自动化
  • 带外部写操作的运维助手

4. 案例一:先用 LLM 解决“客服工单分流”

4.1 业务背景

假设我们要做一个客服工单入口,用户输入可能像这样:

我昨晚充值成功了,但是会员还是没到账,订单号是 A-1024,已经等了 6 个小时。

这个场景其实非常适合先做 LLM 节点,而不是一上来做 Agent。原因很简单:

  • 目标是做结构化判断
  • 没有必须的多步工具循环
  • 不需要让模型自己决定去查多少个系统
  • 我们更关心输出结构稳定,而不是“自主性”

4.2 目标输出

我们希望模型稳定输出:

  • category
  • priority
  • needs_human
  • sentiment
  • order_id
  • next_action

这里最稳的做法之一,是直接借 Claude 的 tool use 机制,把“结构化结果”做成一个只用于承载输出的客户端工具。

4.3 为什么这里用“结构化 tool output”比自然语言更稳

如果你只让模型自由回答,常见问题会是:

  • 字段名飘
  • 枚举值漂
  • 解释太长,工作流层不好消费
  • 下游系统不得不二次解析自然语言

而把输出做成一个固定工具,例如 emit_ticket_analysis,好处会很直接:

  • 字段更稳定
  • 枚举更稳定
  • 下游可以直接消费 JSON
  • 回归测试更容易写

4.4 完整 Python 示例

下面这段代码的目标很单纯:

  • 用 Claude 把自然语言工单转成结构化分流对象
  • 通过 tool_use 拿到稳定字段
  • 不引入额外 Agent loop
python
import os
from typing import Any, Dict

from anthropic import Anthropic

client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

EMIT_TICKET_ANALYSIS_TOOL: Dict[str, Any] = {
    "name": "emit_ticket_analysis",
    "description": (
        "Return a structured ticket analysis for customer-support routing. "
        "Always call this tool exactly once instead of replying with free-form text."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "category": {
                "type": "string",
                "enum": [
                    "billing",
                    "account",
                    "delivery",
                    "refund",
                    "abuse",
                    "other",
                ],
            },
            "priority": {
                "type": "string",
                "enum": ["low", "medium", "high", "urgent"],
            },
            "needs_human": {"type": "boolean"},
            "sentiment": {
                "type": "string",
                "enum": ["calm", "frustrated", "angry"],
            },
            "order_id": {
                "type": ["string", "null"],
                "description": "The extracted order id when present.",
            },
            "next_action": {
                "type": "string",
                "description": "Short routing advice for the downstream workflow.",
            },
        },
        "required": [
            "category",
            "priority",
            "needs_human",
            "sentiment",
            "order_id",
            "next_action",
        ],
        "additionalProperties": False,
    },
}

SYSTEM_PROMPT = """
You are a customer-support triage model.
Classify the request, extract the order id when present, and provide the next routing action.
Use the emit_ticket_analysis tool exactly once.
Do not answer in free-form text.
""".strip()


def extract_single_tool_input(response: Any, expected_name: str) -> Dict[str, Any]:
    if response.stop_reason != "tool_use":
        raise RuntimeError(
            f"Expected stop_reason=tool_use, got {response.stop_reason!r}"
        )

    tool_blocks = [block for block in response.content if block.type == "tool_use"]
    if len(tool_blocks) != 1:
        raise RuntimeError(f"Expected exactly one tool call, got {len(tool_blocks)}")

    tool_block = tool_blocks[0]
    if tool_block.name != expected_name:
        raise RuntimeError(
            f"Expected tool {expected_name!r}, got {tool_block.name!r}"
        )

    return tool_block.input


def analyze_ticket(user_text: str) -> Dict[str, Any]:
    response = client.messages.create(
        model="your-claude-model",
        max_tokens=800,
        system=SYSTEM_PROMPT,
        tools=[EMIT_TICKET_ANALYSIS_TOOL],
        messages=[{"role": "user", "content": user_text}],
    )
    return extract_single_tool_input(response, "emit_ticket_analysis")


if __name__ == "__main__":
    ticket = (
        "我昨晚充值成功了,但是会员还是没到账,订单号是 A-1024,"
        "已经等了 6 个小时,现在很着急。"
    )
    result = analyze_ticket(ticket)
    print(result)

4.5 这段代码真正值得学的不是 SDK,而是 4 个控制点

  1. 把任务收敛成一次结构化判断,而不是默认进入 Agent loop。
  2. 用固定 tool schema 约束输出,而不是让自然语言承担系统契约。
  3. 显式检查 stop_reason,不要假设模型“一定会按你想的方式停下来”。
  4. 让下游工作流直接消费结构化对象,而不是再做一轮文本解析。

4.6 什么时候该把这个 LLM 节点升级成 Agent

如果后面业务要求变成:

  • 分类之前要先查用户历史订单
  • 要结合多个系统做真假投诉判断
  • 需要自动起草补偿方案
  • 涉及真实退款、发消息或改单

那你就已经开始进入 Agent 区间了。


5. 案例二:再用 Agent 解决“生产事故分诊”

5.1 业务背景

现在看另一个更像 Agent 的问题:

支付服务告警升高,最近 30 分钟成功率下降,请帮我先做事故分诊,并在确认后创建事故工单。

这个场景和上面的工单分类完全不一样,因为它天然需要:

  • 多步查证
  • 多个工具
  • 把工具结果继续送回模型
  • 先只读调查,再审批,再写入系统

这就是典型的 Agent 问题。

5.2 这个 Agent 为什么适合按 Claude 风格来实现

因为 Claude 的 tool use loop 天然适合表达下面这个控制流:

  1. 模型先读任务。
  2. 如果需要证据,返回 tool_use
  3. 应用执行工具,并把 tool_result 回传。
  4. 模型继续判断。
  5. 如果只是读操作,继续循环。
  6. 如果准备写操作,先走审批,再开放写工具。

这个模式最大的好处不是“聪明”,而是:

  • 运行边界非常清楚

5.3 设计原则

这个案例故意采用 4 个收缩动作:

  1. 第一阶段只开放只读工具。
  2. 先让模型产出结论和建单草稿依据。
  3. 审批通过以后,第二阶段才开放写工具。
  4. 工具执行层和模型层明确分离。

5.4 完整 Python 示例

下面这段代码重点展示:

  • Claude 工具循环
  • stop_reason 分支
  • 工具结果回填
  • 审批后再开放写工具
python
import json
import os
from typing import Any, Dict, List, Tuple

from anthropic import Anthropic

client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

BASE_SYSTEM_PROMPT = """
You are an incident-triage agent.
Use read-only tools first.
Do not create incident tickets until an approval id is explicitly provided.
Base conclusions on concrete tool evidence, not guesses.
Return concise summaries that operators can review quickly.
""".strip()

READ_ONLY_TOOLS: List[Dict[str, Any]] = [
    {
        "name": "read_recent_alerts",
        "description": "Read recent alerts for a service within the last N minutes.",
        "input_schema": {
            "type": "object",
            "properties": {
                "service": {"type": "string"},
                "window_minutes": {"type": "integer"},
            },
            "required": ["service", "window_minutes"],
            "additionalProperties": False,
        },
    },
    {
        "name": "search_runbook",
        "description": "Search the incident runbook for a service and symptom.",
        "input_schema": {
            "type": "object",
            "properties": {
                "service": {"type": "string"},
                "symptom": {"type": "string"},
            },
            "required": ["service", "symptom"],
            "additionalProperties": False,
        },
    },
]

WRITE_TOOLS: List[Dict[str, Any]] = [
    {
        "name": "create_incident_ticket",
        "description": "Create an incident ticket after human approval is granted.",
        "input_schema": {
            "type": "object",
            "properties": {
                "title": {"type": "string"},
                "severity": {
                    "type": "string",
                    "enum": ["sev1", "sev2", "sev3"],
                },
                "summary": {"type": "string"},
                "approval_id": {"type": "string"},
            },
            "required": ["title", "severity", "summary", "approval_id"],
            "additionalProperties": False,
        },
    }
]

ALERT_DATA = {
    "payments": [
        {
            "timestamp": "2026-07-09T10:05:00+08:00",
            "signal": "success_rate_drop",
            "value": "91.2%",
        },
        {
            "timestamp": "2026-07-09T10:08:00+08:00",
            "signal": "db_timeout_spike",
            "value": "p95=2.8s",
        },
    ]
}

RUNBOOK_DATA = {
    ("payments", "success_rate_drop"): {
        "likely_causes": [
            "database latency",
            "downstream payment gateway timeout",
        ],
        "recommended_checks": [
            "check recent db latency changes",
            "compare error spikes by gateway",
        ],
        "default_severity": "sev2",
    }
}


def call_claude(messages: List[Dict[str, Any]], tools: List[Dict[str, Any]]) -> Any:
    return client.messages.create(
        model="your-claude-model",
        max_tokens=1400,
        system=BASE_SYSTEM_PROMPT,
        tools=tools,
        messages=messages,
    )


def response_block_to_dict(block: Any) -> Dict[str, Any]:
    if block.type == "text":
        return {"type": "text", "text": block.text}
    if block.type == "tool_use":
        return {
            "type": "tool_use",
            "id": block.id,
            "name": block.name,
            "input": block.input,
        }
    raise ValueError(f"Unsupported content block type: {block.type}")


def extract_text(content_blocks: List[Any]) -> str:
    return "\n".join(block.text for block in content_blocks if block.type == "text").strip()


def run_tool(name: str, tool_input: Dict[str, Any]) -> Dict[str, Any]:
    if name == "read_recent_alerts":
        service = tool_input["service"]
        return {
            "service": service,
            "window_minutes": tool_input["window_minutes"],
            "alerts": ALERT_DATA.get(service, []),
        }

    if name == "search_runbook":
        key = (tool_input["service"], tool_input["symptom"])
        return RUNBOOK_DATA.get(
            key,
            {
                "likely_causes": [],
                "recommended_checks": [],
                "default_severity": "sev3",
            },
        )

    if name == "create_incident_ticket":
        return {
            "ticket_id": "INC-20260709-1042",
            "status": "created",
            "approval_id": tool_input["approval_id"],
            "title": tool_input["title"],
            "severity": tool_input["severity"],
        }

    raise ValueError(f"Unknown tool: {name}")


def require_supported_stop_reason(response: Any) -> None:
    if response.stop_reason in {"tool_use", "end_turn"}:
        return
    if response.stop_reason == "max_tokens":
        raise RuntimeError("Claude hit max_tokens. Resume or increase the limit.")
    if response.stop_reason == "pause_turn":
        raise RuntimeError("Claude paused the turn. Resume logic is required here.")
    if response.stop_reason == "refusal":
        raise RuntimeError("Claude refused the request. Check policy and fallback flow.")
    raise RuntimeError(f"Unexpected stop_reason: {response.stop_reason!r}")


def tool_loop(
    messages: List[Dict[str, Any]],
    tools: List[Dict[str, Any]],
) -> Tuple[Any, List[Dict[str, Any]]]:
    while True:
        response = call_claude(messages, tools)
        require_supported_stop_reason(response)

        assistant_message = {
            "role": "assistant",
            "content": [response_block_to_dict(block) for block in response.content],
        }
        messages.append(assistant_message)

        tool_result_blocks: List[Dict[str, Any]] = []
        saw_tool_call = False

        for block in response.content:
            if block.type != "tool_use":
                continue

            saw_tool_call = True
            result = run_tool(block.name, block.input)
            tool_result_blocks.append(
                {
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(result, ensure_ascii=False),
                }
            )

        if not saw_tool_call:
            return response, messages

        messages.append({"role": "user", "content": tool_result_blocks})


def request_human_approval(summary: str) -> Dict[str, Any]:
    print("=== TRIAGE SUMMARY FOR REVIEW ===")
    print(summary)
    approved = input("Approve incident creation? [y/N]: ").strip().lower() == "y"
    return {
        "approved": approved,
        "approval_id": "APR-20260709-0001" if approved else None,
    }


def triage_incident(task: str) -> Dict[str, Any]:
    messages: List[Dict[str, Any]] = [{"role": "user", "content": task}]

    analysis_response, messages = tool_loop(messages, READ_ONLY_TOOLS)
    analysis_summary = extract_text(analysis_response.content)

    approval = request_human_approval(analysis_summary)
    if not approval["approved"]:
        return {
            "status": "cancelled",
            "analysis_summary": analysis_summary,
            "reason": "human_rejected",
        }

    messages.append(
        {
            "role": "user",
            "content": (
                "Human approval granted. "
                f"approval_id={approval['approval_id']}. "
                "You may now create the incident ticket if still necessary."
            ),
        }
    )

    final_response, messages = tool_loop(
        messages,
        READ_ONLY_TOOLS + WRITE_TOOLS,
    )

    return {
        "status": "completed",
        "analysis_summary": analysis_summary,
        "final_answer": extract_text(final_response.content),
    }


if __name__ == "__main__":
    task = (
        "支付服务最近 30 分钟成功率下降,请先调查,"
        "如果证据充分,再在审批通过后创建事故工单。"
    )
    print(triage_incident(task))

5.5 这段 Agent 代码最关键的 6 个工程点

  1. 第一阶段只开放只读工具,而不是默认把写工具暴露给模型。
  2. 工具执行完全在应用层,模型只负责发起 tool_use
  3. 每轮都显式处理 stop_reason,而不是只看文本内容。
  4. 原始 assistant 内容和 tool_result 都被保留回消息历史,符合 Claude 的无状态续跑方式。
  5. 审批是应用层控制点,不是让模型自己说“我已经批准了”。
  6. 写工具只在审批通过后加入 tool surface,这比只靠 prompt 警告更稳。

5.6 如果把这个 Agent 再做完整,下一步该补什么

真实上线时,建议继续补:

  • trace_id
  • tool latency
  • idempotency_key
  • approval snapshot
  • artifact storage
  • resume token
  • eval cases

否则系统很容易停在“能跑通 Demo”,但一上线就开始:

  • 工具循环过长
  • 审批恢复失败
  • 重试产生重复副作用
  • 事故后没法回放

6. 两个案例放在一起,真正要建立什么判断力

6.1 不要把“会调用模型”误认为“已经做了 Agent”

案例一只是:

  • 一个结构化 LLM 节点

它很有价值,但它不是 Agent。

6.2 不要把“用了工具”就默认升级成 Agent 平台

如果只有一次工具调用,而且没有多步循环、状态续接和审批边界,它可能只是:

  • 带工具的 LLM 工作流节点

6.3 真正的 Agent,核心不是工具数量,而是控制循环

判断重点通常是:

  • 会不会多轮续跑
  • 会不会根据工具结果再决定下一步
  • 会不会带来读写边界和审批问题
  • 会不会需要恢复、补偿和观测

7. 用 Claude 思路做项目时,最值得保留的几个习惯

  1. 永远把 stop_reason 当成运行时协议的一部分。
  2. 永远把 tool_use 和真实执行环境分开。
  3. 尽量把稳定大前缀做 prompt caching。
  4. 能先做 LLM 节点 的,不要一上来就重型 Agent。
  5. 对高风险写工具,优先做“只读调查 -> 草稿 -> 审批 -> 写入”。
  6. 工具结果先结构化、再清洗、再回灌,不要原样整包塞回上下文。

8. 如果你要拿这页去做面试或项目汇报,最好这样讲

一个比较有说服力的表达方式通常是:

  1. 先解释为什么第一个问题只需要 LLM,不需要 Agent。
  2. 再解释为什么第二个问题必须进入 Agent loop。
  3. 然后讲清楚 Claude 风格实现里最关键的三件事:
    • tool_use / tool_result
    • stop_reason
    • 审批前后的 tool surface 收缩
  4. 最后补一句你打算如何做评测、恢复和观测。

这样别人更容易判断你不是只会:

  • 背概念
  • 接 SDK
  • 拼几个 demo

而是真的知道:

  • 什么时候不该上 Agent
  • 什么时候该上
  • 上了以后怎么收边界

9. 推荐联读


10. 参考资料

以下资料已按 2026-07-09 核对到当前正式入口: