AB
AB
AiBoss站

全部分类

教程

Claude

教程

Claude 客服助手构建教程：从对话历史到工具调用与工作流

把 Claude 接入一个可用的客服助手，需要的不只是发一条提示词再打印回复。本文以 ShopHelper 为例，逐步搭建对话历史管理、系统提示词、工具调用循环、多块响应处理与工作流编排，并说明如何评估提示词改动的效果。

大语言模型可以回答问题、总结文档、写代码，也能与外部系统交互。但要把 Claude 做成一个能上线的应用，只发一条提示词再把回复显示出来是不够的。一个可用的应用需要管理对话历史、提供相关上下文、安全地使用工具、处理不同类型的响应块，并且能判断生成的内容是否真的有用。本文以 ShopHelper 为例——一个虚构网店的客服助手——把这些环节逐个搭起来。如果你只是想先快速试一下 Claude 的对话能力，可以先用 Claude 熟悉交互方式，再回到本文看工程化的部分。

跟着本文走完，ShopHelper 会具备这些能力：用一致的语气回答一般性问题；记住客户之前说过的话；通过调用你自己代码里的函数查询订单状态；安全地处理 Claude 返回的多块响应；用工作流处理客服工单；以及评估提示词改动是否真的让结果变好。每一节只加一个部件，可以在自己的编辑器里同步操作。

准备工作

开始之前需要具备这些条件：

  • 基本的 Python 知识
  • Python 3.9 或更高版本
  • 一个 Anthropic API 密钥
  • 对函数和 JSON 有基本了解

先创建虚拟环境并安装 Anthropic Python SDK 与 dotenv：

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

在 Windows 上激活虚拟环境的命令不同：

.venv\Scripts\activate

然后创建 .env 文件，把密钥放进去：

ANTHROPIC_API_KEY=your_api_key_here

API 密钥是机密凭据。不要把它放进浏览器 JavaScript、移动端应用代码或任何客户端配置里，也不要提交到代码仓库：

echo ".env" >> .gitignore

如果之后要加网页界面，密钥必须留在后端，调用链路是：浏览器 → 你的后端 → 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": "Explain what an API is in simple terms."
        }
    ],
)

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。一次性的请求只包含一条 user 消息；多轮对话则包含更早的 user 与 assistant 消息。

Claude 返回的 response.content 是一个带类型的块列表，常见类型包括：

块类型含义
text生成的文本
tool_use请求你的应用去调用某个工具
thinking启用后的推理内容

上面的示例是遍历所有块、只取 text，而不是假设 response.content[0] 一定是文本。这个习惯在后面处理工具调用时会省很多事。

需要监控用量时，可以读取 usage 信息：

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

第二步：管理对话历史

Claude 不会自动记住两次独立请求之间的内容。要让对话连贯，每次请求都要把相关的历史一起发过去：

messages = [
    {"role": "user", "content": "What is your returns policy?"},
    {"role": "assistant", "content": "Items can be returned within 30 days."},
    {"role": "user", "content": "How long do I have?"},
]

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

中间那条 assistant 消息记录了 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, "What is your returns policy?"))
print(chat(history, "How long do I have?"))

每次调用都会追加新的用户消息、发送完整历史，并把 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": (
                "Summarise this conversation in under 100 words. "
                "Keep order numbers and unresolved issues.\n\n"
                f"<conversation>{transcript}</conversation>"
            ),
        }],
    )

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

摘要要作为独立的应用状态保存，在下次请求时作为上下文带上。不要把它当成一条额外的 user 消息插在 recent 之前，那样会产生连续两条 user 消息，属于无效结构。

敏感信息在存储或传输之前应当先脱敏：

import re

def redact(text):
    return re.sub(
        r"\b(?:\d[ -]?){13,16}\b",
        "[REDACTED CARD]",
        text,
    )

第四步：用清晰的边界组织提示词

XML 风格的标签只是普通文本，不是 API 的特殊指令。它的作用是让提示词里每一部分的身份变得明确：

prompt = """
<customer_reviews>
The product is comfortable, but the available colours are limited.
Customers also describe it as durable.
</customer_reviews>

<sales_data>
January: 120 units
February: 150 units
March: 98 units
</sales_data>

<task>
Compare the reviews with the sales data.
Identify possible relationships and state uncertainty.
</task>
"""

这里 <customer_reviews> 标出参考资料，<sales_data> 标出数据，<task> 标出指令。政策条款、用户生成内容、示例和输出要求都可以用同样的方式划出边界。

第五步：设置系统提示词

系统提示词定义 ShopHelper 的整体行为：

system_prompt = """
You are ShopHelper, a friendly customer-support assistant.
Keep answers concise and clear.
Do not invent prices, policies, or order details.
If information is missing, ask for it.
"""

它和对话内容是分开传入的：

response = client.messages.create(
    model=MODEL,
    max_tokens=500,
    system=system_prompt,
    messages=[
        {"role": "user", "content": "Where is my order?"}
    ],
)

因为客户没有提供订单号，ShopHelper 应当主动索要，而不是编一个出来。

第六步：添加工具

Claude 无法直接访问你的数据库。工具的作用是给它一个结构化的方式，向你的应用请求信息：

def get_order_status(order_id):
    orders = {
        "ORD-1001": "shipped",
        "ORD-1002": "processing",
    }
    return {
        "order_id": order_id,
        "status": orders.get(order_id, "not_found"),
    }

这个函数接收订单号、做一次查找、返回结构可预期的数据。生产环境里这个字典会换成数据库查询。

关键点在于：Claude 不会执行这个函数，执行的是你的应用。你需要用 schema 把函数描述给它：

tools = [{
    "name": "get_order_status",
    "description": "Get the current status of a customer order.",
    "input_schema": {
        "type": "object",
        "properties": {
            "order_id": {
                "type": "string",
                "description": "An order ID such as 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 继续。

第七步：处理工具调用响应

在执行之前，要校验工具名、参数以及用户权限：

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, "Unknown tool"

    order_id = tool_input.get("order_id")
    if not isinstance(order_id, str):
        return False, "order_id must be a string"
    if not ORDER_ID_PATTERN.fullmatch(order_id):
        return False, "Invalid order ID format"
    if order_id not in current_user["order_ids"]:
        return False, "The customer cannot access this order"

    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("Validate and execute:", block.name)
    elif block.type == "thinking":
        continue
    else:
        print("Unhandled block type:", block.type)

对应到 ShopHelper 的行为：显示文本、校验并执行通过审核的工具请求、不展示内部推理内容、把未知块类型记录下来。

第九步：工作流与智能体

工作流按预先定义的顺序执行：接收工单 → 提取细节 → 起草回复 → 复核回复。

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>Extract the problem and desired outcome.</task>"
    )
    draft = ask(
        f"<details>{details}</details>\n"
        "<task>Draft a concise support reply.</task>"
    )
    review = ask(
        f"<draft>{draft}</draft>\n"
        "<task>List unsupported promises, or say OK.</task>"
    )
    return draft, review

智能体则更灵活：由 Claude 自己决定是否使用工具、下一步做什么。但智能体同样需要校验，并且要设置最大步数上限，避免陷入无限循环。

一个完整示例

把上面的部件拼起来，就是一个能跑通的最小客服助手。它接收用户输入，必要时调用订单查询工具，最后返回文本回复。

import os
import re
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

MODEL = "claude-sonnet-5"
client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

system_prompt = """
You are ShopHelper, a friendly customer-support assistant.
Keep answers concise and clear.
Do not invent prices, policies, or order details.
If information is missing, ask for it.
"""

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

def get_order_status(order_id):
    orders = {
        "ORD-1001": "shipped",
        "ORD-1002": "processing",
    }
    return {
        "order_id": order_id,
        "status": orders.get(order_id, "not_found"),
    }

tools = [{
    "name": "get_order_status",
    "description": "Get the current status of a customer order.",
    "input_schema": {
        "type": "object",
        "properties": {
            "order_id": {
                "type": "string",
                "description": "An order ID such as ORD-1001."
            }
        },
        "required": ["order_id"],
    },
}]

def validate_tool_request(name, tool_input, current_user):
    if name != "get_order_status":
        return False, "Unknown tool"
    order_id = tool_input.get("order_id")
    if not isinstance(order_id, str):
        return False, "order_id must be a string"
    if not ORDER_ID_PATTERN.fullmatch(order_id):
        return False, "Invalid order ID format"
    if order_id not in current_user["order_ids"]:
        return False, "The customer cannot access this order"
    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})

if __name__ == "__main__":
    user = {"order_ids": ["ORD-1001"]}
    print(run_conversation("Where is my order ORD-1001?", user))
    print(run_conversation("Where is my order ORD-9999?", user))

第一句会触发工具调用并返回已发货状态；第二句的订单号格式合法但不属于该用户，校验会拦下来，把错误信息作为 tool_result 回传，由 Claude 组织成对用户友好的说明。

注意事项

  • API 密钥属于机密凭据，不能出现在浏览器端、移动端或任何客户端配置中，也不要提交进仓库。加了网页界面之后，密钥必须留在后端。
  • Claude 不会自动记住两次独立请求之间的内容，历史必须由应用显式携带。
  • 历史无限增长会推高输入规模，也可能让模型难以聚焦。裁剪或摘要时要注意：摘要不能作为额外的 user 消息插在最近消息之前，否则会产生连续两条 user 消息的无效结构。
  • 工具函数由你的应用执行，Claude 只提出请求。执行前必须校验工具名、参数类型、参数格式和用户权限。
  • 不要假设 response.content[0] 一定是文本块，响应可能包含 tool_use、thinking 等多种块类型。
  • 使用智能体模式时，除了校验之外还要设置最大步数上限。
  • 模型标识符、可用范围、配额与价格会变化，运行前请确认所用模型对你的账号可用，并以官网当前信息为准。

如何评估提示词质量

提示词改动之后，需要判断结果是否真的变好，而不是凭感觉。可行的做法是准备一组有代表性的输入，覆盖常见问题、边界情况和已知的失败案例，对改动前后的输出做对比。评估时关注的不只是回答是否流畅，还包括：是否遵守了系统提示词里的约束、是否在信息缺失时主动追问、工具调用是否被正确触发和校验、以及多块响应是否被完整处理。把这些检查固化成可重复运行的用例，提示词迭代才有依据。

返回