如何利用Claude API构建一个可靠的人工智能助手

大型语言模型能够回答问题、总结文档、编写代码，还能与外部系统进行交互。但构建一个可靠的人工智能应用，仅仅发送指令并显示响应是远远不够的。

一个可用于实际生产的应用程序必须能够管理对话历史记录、提供相关的上下文信息、安全地使用各种工具、处理不同类型的响应，并判断生成的内容是否有用。

在本教程中，我们将构建ShopHelper——这个为虚构在线商店设计的客户支持助手。完成制作后，ShopHelper将能够做到以下几件事：

• 以统一的语气回答常见问题

• 记住顾客之前说过的话

• 通过调用代码中的函数来查询订单状态

• 安全地处理Claude给出的多段式响应

• 利用工作流程来处理客户支持请求

• 评估修改提示语后效果是否有所改善

每个章节都会介绍一个具体的功能，因此你可以在自己的编辑器中逐步跟随这些步骤进行学习。

目录

• 先决条件

• 如何设置项目并保护API密钥的安全

• 如何发送第一个请求

• 如何管理对话历史记录

• 如何制定界限清晰的提示语

• 如何使用系统提示语

• 如何添加工具

• 如何处理与工具使用相关的响应

• Claude的响应可能包含多个部分

• 工作流程与智能助手的区别

• 链式处理、并行化、路由机制及评估优化工具

• 如何评估提示语的质量

• 总结

先决条件

你需要具备以下条件：

• 基本的Python知识

• Python 3.9或更高版本

• Anthropic的API密钥

• 对函数和JSON格式有基本了解

如何设置项目并保护API密钥的安全

首先创建一个虚拟环境，然后安装Anthropic的Python SDK：

python -m venv .venv
source .venv/bin/activate
pip install anthropic python-dotenv

在 Windows 上：

.venv\Scripts\activate

创建一个 .env 文件：

ANTHROPIC_API_KEY=你的_api_key_here

API 密钥是一种机密凭证。切勿将其放入浏览器的 JavaScript 代码、移动应用程序的代码中，也不得将其包含在客户端配置文件中。同时，也千万不要将它提交到任何版本控制仓库中：

echo ".env" >> .gitignore

如果你后来添加了网页接口，那么请将这个 API 密钥保存在你的后端服务器上：

浏览器 → 你的后端服务器 → Claude API

创建 app.py 文件：

import os

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

MODEL = "claude-sonnet-5"

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

load_dotenv() 会从 .env 文件中读取配置信息。变量 MODEL 表示你只需要在一处修改模型名称即可。在运行示例代码之前，请确保你的账户具有使用该模型的权限。

如何发送第一个请求

response = client.messages.create(
    model=MODEL,
    max_tokens=500,
    messages=[
        {
            "role": "user",
            "content": "用简单的语言解释什么是 API。”
        }
    ],
)

answer = "".join(
    block.text
    for block in response.content
    if block.type == "text"
)

print(answer)

一个请求包含三个重要的部分：

• model 用于指定负责处理该请求的 Claude 模型。不同模型在功能、速度和成本方面可能存在差异。

• max_tokens 限制了 Claude 可以生成的最大文本长度。设置较小的值可以减少延迟，但可能会导致 Claude 在未完成回答之前就停止运行。

• messages 包含了对话内容。每条消息都包含 role 和 content；其中，role 通常为 user 或 assistant。

例如，一个一次性请求只包含一条用户发送的消息；而多轮对话则会包含用户之前发送的所有消息以及助手的回复。

Claude 会返回 response.content，这是一个由文本块组成的列表。常见的文本块类型包括：

在这个示例中，我们并没有直接假设 response.content[0] 总是文本格式；而是实际收集了所有的文本块内容。

你可以通过以下代码查看请求的详细使用信息，以便进行监控：

print(response.usage.input_tokens)
print(response_usage.output_tokens)

如何管理对话记录

Claude不会自动记住所有的API请求内容。因此，每次发送请求时都应附带相关的对话记录：

messages = [
    {
        "role": "user",
        "content": "你们的退货政策是怎样的？"
    },
    {
        "role": "assistant",
        "content": "商品可以在30天内退回。”
    },
    {
        "role": "user",
        "content": "那我还有多少时间呢？"
    },
]

response = client.messages.create(
    model=MODEL,
    max_tokens=300,
    messages=messages,
)

助手会记录下Claude之前的回答，这样用户提出的最后一个问题就能在上下文中得到正确的理解。

一个简单的聊天功能也可以用来维护对话记录：

def chat(history, user_text):
    history.append({
        "role": "user",
        "content": user_text,
    })

    response = client.messages.create(
        model=MODEL,
        max_tokens=500,
        messages=history,
    )

    reply = "".join(
        block.text
        for block in response.content
        if block.type == "text"
    )

    history.append({
        "role": "assistant",
        "content": reply,
    })

    return reply


history = []

print(chat(history, "你们的退货政策是怎样的？"))
printCHAT(history, "那我还有多少时间呢？"))

每次发送请求时，都会添加新的用户消息，同时将完整的对话记录一起发送出去，并保存Claude的回复以备下一次对话使用。在实际应用中，可以根据客户或会话ID来存储这些对话记录。

如何处理不断增长的对话记录

如果允许无限量的对话记录被保存下来，那么输入数据的规模就会增大，这可能会让Claude难以集中注意力进行处理。因此，可以采取以下两种方法之一：

def trim_history(history, max_messages=10):
    trimmed = history[-max_messages:]

    while trimmed and trimmed[0]["role"] != "user":
        trimmed.pop(0)

    return trimmed

一种方法是只保留最近的几条对话记录；另一种方法则是对较早的对话内容进行总结，同时仍然保留最新的消息：

def summarise_history(history, keep_last=6):
    old = history[:-keep_last]
    recent = history[-keep_last:]

    transcript = "\n".join(
        f"{message['role']}: {message['content']}"
        for message in old
    )

    response = client.messages.create(
        model=MODEL,
        max_tokens=250,
        messages=[{
            "role": "user",
            "content": (
                "请用不到100个词来总结这次对话。请保持顺序并保留未解决的问题。\n\n"
                f"<conversation>{transcript}</conversation>"
            ),
        ],
    )

    summary = "".join(
        block.text
        for block in response.content
        if block.type == "text"
    )

    return summary, recent

应该将对话总结内容作为单独的应用状态进行保存，并在下次请求时将其作为上下文信息一起传递。切勿将总结内容插入到recent之前的用户消息中，因为这样可能会导致连续的用户消息顺序混乱。

在存储或传输敏感信息之前，也应对其进行脱敏处理：

import re

def redact(text):
    return re.sub(
        r"\b(?:\d[ -]?){13,16}\b",
        "[已屏蔽内容]",
        text,
    )

如何为提示信息设定明确的边界结构

这些采用XML风格的标签其实只是普通文本，并非特殊的API指令。它们能够明确地标识出提示信息中的各个部分：

prompt = """
<customer_reviews>
这款产品穿着很舒适，但可选择的颜色种类有限。
顾客们也认为它非常耐用。
</customer_reviews>

<sales_data>
1月：销售了120件商品
2月：销售了150件商品
3月：销售了98件商品
</sales_data>

<task>>
将这些评价与销售数据进行对比分析，
找出其中可能存在的关系，并说明那些尚不确定的地方。
</task>
"""

在这里，<customer_reviews>表示参考内容，<sales_data>表示数据信息，而<task>则表示指令。对于政策说明、用户生成的内容、示例以及输出要求等，也可以使用类似的边界划分方式。

如何使用系统提示语

系统提示语决定了ShopHelper的整体行为模式：

system_prompt = """
您是ShopHelper，一位友好的客户支持助手。

请保持回答简洁明了，
不要随意编造价格、政策或订单细节，
如果缺少相关信息，请及时询问。
"""

在发送对话请求时，需要将系统提示语与其他信息分开传递：

response = client.messages.create(
    model=MODEL,
    max_tokens=500,
    system=system_prompt,
    messages=[
        {"role": "user", "content": "我的订单在哪里？"}
    ],
)

由于客户没有提供订单编号，因此ShopHelper应该询问客户这个信息，而不是自行猜测。

如何添加辅助工具

Claude无法直接访问您的数据库。辅助工具可以为它提供一种结构化的方式，以便从您的应用程序中获取所需信息：

def get_order_status(order_id):
    orders = {
        "ORD-1001": "已发货",
        "ORD-1002": "正在处理中"
    }

    return {
        "order_id": order_id,
        "status": orders.get(order_id, "未找到")
    }

这个函数接受一个订单编号作为参数，然后在数据库中查找相应的信息，并返回结果。在实际应用环境中，这个功能会通过数据库查询来实现。Claude本身并不执行这个函数，而是由您的应用程序来处理的。

可以使用数据结构来描述这个函数的功能：

tools = [
    {
        "name": "get_order_status",
        "description": "获取客户订单的当前状态。",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "例如ORD-1001这样的订单编号"
                }
            }
        },
        "required": ["order_id"]
    }
]

Claude可能会返回一个tool_use块，而不是最终的答案：

type="tool_use"
id="toolu_example"
name="get_order_status"
input={"order_id": "ORD-1001"}

name用于标识该函数，input包含其参数，而id在返回结果时是必需的。当stop_reason为"tool_use"时，意味着你的应用程序应在请求Claude继续处理之前先自行处理这个请求。

如何处理Tool-Use响应

Tool-Use响应就是包含上述tool_use块的响应信息。

在执行操作之前，需要验证工具的名称、参数以及用户的权限：

import re

ORDER_ID_PATTERN = re.compile(r"^ORD-\d{4}$")

def validate_tool_request(name, tool_input, current_user):
    if name != "get_order_status":
        return False, "未知的工具"

    order_id = tool_input.get("order_id")

    if not isinstance(order_id, str):
        return False, "order_id必须是字符串类型"

    if not ORDER_ID_pattern.fullmatch(order_id):
        return False, "订单ID格式无效"

    if order_id not in current_user["order_ids"]:
        return False, "该客户无权访问此订单"

    return True, None

然后，可以通过一个完整的循环来验证并执行这个请求：

def run_conversation(user_text, current_user):
    messages = [{"role": "user", "content": user_text}]

    while True:
        response = client.messages.create(
            model=MODEL,
            max_tokens=500,
            system=system_prompt,
            tools=tools,
            messages=messages,
        )

        if response.stop_reason != "tool_use":
            return "".join(
                block.text
                for block in response.content
                if block.type == "text"
            )

        messages.append({
            "role": "assistant",
            "content": response.content,
        })

        results = []

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

            valid, error = validate_tool_request(
                block.name,
                block.input,
                current_user,
            )

            if valid:
                result = get_order_status(block.input["order_id"])
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result),
                })
            else:
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": error,
                    "is_error": True,
                })

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

tool_use_id这一机制用于将处理结果与原始请求关联起来。应用程序仍然负责授权以及执行相关操作。

Claude的响应可以包含多个内容块

然而，这种假设其实并不十分可靠：

answer = response.content[0].text

这种代码假设第一个内容块一定是文本形式，但实际上应该逐个检查每个内容块：

for block in response.content:
    if block.type == "text":
        print(block.text)
    elif block.type == "tool_use":
        print("验证并执行操作：", block.name)
    elif block.type == "thinking":
        continue
    else:
        print("未知的内容块类型：", block.type)

ShopHelper会显示文本内容，验证并执行那些已被批准的工具请求，不会展示内部处理过程，而对于那些未知类型的 content 块，则会将其记录下来。

工作流程与智能代理的区别

工作流程是按照预先定义的顺序来执行的：

接收工单
↓
提取相关信息
↓
起草回复内容
↓
审核回复内容

def ask(prompt, max_tokens=500):
    response = client.messages.create(
        model=MODEL,
        max_tokens=max_tokens,
        messages=[{"role": "user", "content": prompt}],
    )

    return "".join(
        block.text
        for block in response.content
        if block.type == "text"
    )


def handle_ticket_workflow(ticket):
    details = ask(
        f"<ticket>{ticket}</ticket>\n"
        "<task>>提取问题描述及期望的结果。起草简洁的支持回复内容。列出无法支持的请求项，或直接选择“同意”。"
    )

    return draft, review

智能代理则具有更大的灵活性：Claude会自行决定是否使用某种工具，以及接下来应该执行什么操作。不过，智能代理仍然需要经过验证，并且其操作步骤的数量也是有限制的。上面提到的run_conversation()函数可以在智能代理的循环中被重复使用。

当处理步骤已经明确、且重复性非常重要时，应使用工作流程；而当下一步行动需要根据当前的结果来决定时，则应该使用智能代理。

链式处理、并行计算、路由机制以及评估优化工具

链式处理是指将每个处理结果传递到下一阶段进行处理：

def chained_reply(ticket, policy):
    draft = ask(
        f"<ticket>{ticket}</ticket>\n"
        "<task>>起草支持回复内容。列出无法支持的请求项。"
    )

    return ask(
        f"<draft>{draft}</draft>\n"
        f"<issues>{issues}</issues>\n"
        "<task>>重新编写最终回复内容。

并行处理是指同时运行独立的任务：

from concurrent.futures import ThreadPoolExecutor

tickets = [
    "我的耳机收到时已经损坏了。",
    "我被收取了两次费用。",
    "如何更改我的地址？",
]

def summarise(ticket):
    return ask(
        f"<ticket>{ticket}</ticket>\n"
        "<task>>用一句话总结这个问题。<./task>",
        max_tokens=100,
    )

with ThreadPoolExecutor(max_workers=3) as pool:
    summaries = list(pool.map(summarise, tickets))

digest = ask(
    "<summaries>\n"
    + "\n".join(summaries)
    + "\n</summaries>\n"
    "<task>>总结今天的客服主题。<./task>"
)

def route(ticket):
    label = ask(
        f"<ticket>{ticket}</ticket>\n"
        "<task>>请准确选择“退款”、“配送”或“其他一般性咨询”。<./task>"
        max_tokens=10,
    ).strip().lower()

    return label if label in {"refund", "delivery", "general"} else "general"

def improve_reply(ticket, rounds=2):
    reply = ask(
        f"<ticket>{ticket}</ticket>\n"
        "<task>>请撰写客服回复。"
    )

    for _ in range(rounds):
        review = ask(
            f"<reply>{reply}</reply>\n"
            "<task>>请指出回复中的错误或不足，或者直接标记为“通过”。<./task>"
        )

        if review.strip().upper() == "PASS":
            break

        reply = ask(
            f"<reply>{reply}</reply>\n"
            f"<review>>{review}</review>\n"
            "<task>>请重新撰写回复。"
        )

    return reply

test_cases = [
    {
        "ticket": "我的耳机损坏了，需要退款。",
        "expected": "refund",
    },
    {
        "ticket": "ORD-1002订单在哪里？",
        "expected": "delivery",
    },
    {
        "ticket": "你们有销售礼品卡吗？",
        "expected": "general",
    },
]

def evaluate(route_fn, cases):
    passed = 0

    for case in cases:
        result = route_fn(case["ticket"])

        if result == case["expected":
            passed += 1
        else:
            print("失败：", case["ticket"], "实际结果：", result)

    score = passed / len(cases)
    print(f"{passed}/{len(cases)}个测试用例通过")
    return score