如何使用 Claude API 构建可靠的 AI 助手
大语言模型能够回答问题、总结文档、编写代码，并与外部系统交互。但要构建可靠的 AI 应用，仅靠发送提示词并显示回复是远远不够的。
生产级别的应用需要管理对话历史、提供相关上下文、安全地调用工具、处理不同的响应类型，并评估生成的输出是否有用。
在本教程中，我们将构建一个ShopHelper，它是虚构在线商店的客户支持助手。到最后，ShopHelper 将具备以下能力：
以一致的语气回答一般性问题
记住客户之前说过的话
通过调用代码中的函数来查询订单状态
安全地处理 Claude 的多块（multi-block）响应
使用工作流（workflow）处理支持工单
评估提示词修改是否提升了结果
每一节都会增加一个功能模块，方便你在自己的编辑器中跟随操作。
目录
前置准备
你需要准备：
基础的 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=your_api_key_here
API 密钥是秘密凭证。切勿将其放在浏览器 JavaScript、移动应用代码或客户端配置中。也绝不要将其提交到代码仓库：
echo ".env" >> .gitignore
如果后续添加 Web 界面，请确保密钥保存在后端：
Browser → Your backend → 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。角色通常是user或assistant。
比如，单次请求只包含一条 user 消息，而多轮对话则包含之前的 user 和 assistant 消息。
Claude 返回的 response.content 是一个由带类型的内容块组成的列表。常见的内容块包括：
| 内容块类型 | 含义 |
|---|---|
text |
生成的文本 |
tool_use |
要求你的应用调用某个工具 |
thinking |
开启扩展思考时的推理内容 |
示例代码会收集所有 text 内容块，而不是直接假定 response.content[0] 一定是文本。
你可以查看用量信息来做监控：
print(response.usage.input_tokens)
print(response.usage.output_tokens)
如何管理对话历史
Claude 不会自动记住各次独立的 API 请求。你需要在每次请求时把相关的历史记录一起发送：
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?"))
每次调用都会追加新的 user 消息、发送完整的历史记录，并把 Claude 的回复存起来供下一轮使用。在生产环境中，应按客户 ID 或会话 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
请将摘要作为独立的应用状态保存，并在下次请求时作为上下文传入。不要将其插入到 recent 之前作为一条额外的用户消息，因为这会制造出无效的连续用户消息序列。
敏感信息在存储或传输前也应当进行脱敏处理：
import re
def redact(text):
return re.sub(
r"\b(?:\d[ -]?){13,16}\b",
"[REDACTED CARD]",
text,
)
如何用清晰边界构建 Prompt 结构
XML 风格的标签本质上是普通文本，并非特殊的 API 指令。它们能让 Prompt 的各个部分更加明确：
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> 标识指令。对于策略、用户生成内容、示例及输出要求，也请使用类似的边界进行界定。
如何使用 System Prompt
System Prompt 定义了 ShopHelper 的通用行为准则：
system_prompt = """
你是 ShopHelper，一位友善的客服助手。
回答需简洁清晰。
不要虚构价格、政策或订单细节。
如果信息缺失，请主动询问。
"""
在传递时，请将 System Prompt 与会话内容分开：
response = client.messages.create(
model=MODEL,
max_tokens=500,
system=system_prompt,
messages=[
{"role": "user", "content": "我的订单在哪里？"}
],
)
由于顾客未提供订单号，ShopHelper 应当索要订单号，而不是盲目猜测。
如何添加 Tools
Claude 无法直接访问你的数据库。通过 Tools，它可以以结构化的方式向你的应用请求信息：
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"),
}
该函数接收订单 ID，执行查询并返回可预期的数据。在生产环境中，字典会被数据库查询替代。Claude 不会执行该函数，而是由你的应用来执行。
使用 schema 描述该函数：
tools = [{
"name": "get_order_status",
"description": "获取客户订单的当前状态。",
"input_schema": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "订单 ID，例如 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, "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 中，程序会显示文本内容，校验并执行获批的 Tool 请求，屏蔽内部思维链，同时记录未知的块类型。
Workflow 与 Agent
Workflow 遵循预定义的执行序列：
接收工单
↓
提取详情
↓
起草回复
↓
审核回复
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
Agent 的灵活性更高：由 Claude 自行决定何时调用工具以及下一步执行什么操作。不过，Agent 同样需要验证机制和最大步数限制。上述 run_conversation() 函数可以直接复用在 Agent 的循环逻辑中。
如果步骤明确且注重可重复性，选用 Workflow；如果后续动作依赖于当前结果，则选用 Agent。
链式调用、并行化、路由与评估-优化
链式调用将每一步的结果传递给下一阶段：
def chained_reply(ticket, policy):
draft = ask(
f"<ticket>{ticket}</ticket>\n"
"<task>Draft a support reply.</task>"
)
issues = ask(
f"<policy>{policy}</policy>\n"
f"<draft>{draft}</draft>\n"
"<task>List unsupported claims.</task>"
)
return ask(
f"<draft>{draft}</draft>\n"
f"<issues>{issues}</issues>\n"
"<task>Rewrite the final reply.</task>"
)
并行化让独立任务并发执行：
from concurrent.futures import ThreadPoolExecutor
tickets = [
"My headphones arrived broken.",
"I was charged twice.",
"How do I change my address?",
]
def summarise(ticket):
return ask(
f"<ticket>{ticket}</ticket>\n"
"<task>Summarise in one sentence.</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>Summarise today's support themes.</task>"
)
路由是先对请求分类，再选择对应的专用流程：
def route(ticket):
label = ask(
f"<ticket>{ticket}</ticket>\n"
"<task>Return exactly refund, delivery, or general.</task>",
max_tokens=10,
).strip().lower()
return label if label in {"refund", "delivery", "general"} else "general"
评估器-优化器（Evaluator-optimizer）先生成答案，再审查并修订：
def improve_reply(ticket, rounds=2):
reply = ask(
f"<ticket>{ticket}</ticket>\n"
"<task>Write a support reply.</task>"
)
for _ in range(rounds):
review = ask(
f"<reply>{reply}</reply>\n"
"<task>List accuracy or tone problems, or say PASS.</task>"
)
if review.strip().upper() == "PASS":
break
reply = ask(
f"<reply>{reply}</reply>\n"
f"<review>{review}</review>\n"
"<task>Rewrite the reply.</task>"
)
return reply
总结一下：各阶段相互依赖时用链接，工作相互独立时用并行化，需要分流处理时用路由，当质量提升值得多花几次 API 调用时用评估器-优化器循环。
如何评估提示词质量
准备有代表性的测试用例：
test_cases = [
{
"ticket": "I want a refund for broken headphones.",
"expected": "refund",
},
{
"ticket": "Where is ORD-1002?",
"expected": "delivery",
},
{
"ticket": "Do you sell gift cards?",
"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("Failed:", case["ticket"], result)
score = passed / len(cases)
print(f"{passed}/{len(cases)} passed")
return score
针对标签和 JSON 数据，使用代码式评估器。针对语气、准确性和有用性，使用人工或基于模型的评估器。
结语
基于 Claude API 进行开发，远不止编写提示词那么简单。构建可靠的应用需要结构化的上下文、受控的对话状态、经过验证的工具执行、审慎的响应处理、恰当的工作流以及可重复的评估机制。
目标不是寻找完美的提示词，而是围绕 Claude 构建一套系统，该系统能提供正确的上下文，限制不安全操作，妥善处理不确定性，并衡量变更是否提升了结果。