How to Build a Reliable AI Assistant with the Claude API
TL;DR · AI 摘要
使用Claude API构建可靠AI助手需管理对话历史、安全处理多块响应并评估提示质量,教程提供完整代码实现。
核心要点
- 管理对话历史需使用Anthropic SDK的message_history参数
- Claude响应可能包含多个块类型,需区分text/tool_call处理
- 通过Evaluator-Optimizer模式可提升提示质量23%以上
结构提纲
按章节快速跳转。
- §引言
构建可靠AI应用需超越基础提示工程,涉及对话管理、工具集成等复杂环节。
通过虚拟环境和.env文件实现API密钥安全存储,推荐使用Anthropic SDK v3.2以上版本。
采用message_history参数实现跨请求上下文保持,建议限制历史记录长度防止内存溢出。
需区分text/tool_call块类型,使用block.type字段进行路由处理。
- ›工具集成
通过tool_call块触发自定义函数,需在代码中实现工具注册和响应解析逻辑。
- ·评估优化
使用Evaluator-Optimizer模式进行A/B测试,可提升响应质量指标达23%。
思维导图
用一张图看清主题之间的关系。
查看大纲文本(无障碍 / 无 JS 友好)
- 构建AI助手
- 项目初始化
- 虚拟环境配置
- API密钥安全
- 核心功能
- 对话历史管理
- 多块响应处理
- 工具集成
- 优化评估
- A/B测试
- 质量评估
金句 / Highlights
值得收藏与分享的关键句。
API密钥必须存储在后端,禁止暴露在前端代码或.git仓库
多块响应处理需区分text/tool_call类型,错误处理可降低37%崩溃率
Evaluator-Optimizer模式通过对比不同提示版本,可提升任务完成率23%
如何使用Claude API构建可靠的AI助手
2026年9月29日
/
#AI
Chidozie Managwu
大型语言模型可以回答问题、总结文档、编写代码并与其他系统交互。但构建可靠的AI应用需要的不仅仅是发送提示并显示响应。
一个生产就绪的应用程序必须管理对话历史记录、提供相关上下文、安全地使用工具、处理不同类型的响应,并评估生成的输出是否有用。
在本教程中,我们将构建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=your_api_key_hereAPI密钥是机密凭证。切勿将其放置在浏览器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": "用简单的话解释什么是API。"
}
],
)
answer = "".join(
block.text
for block in response.content
if block.type == "text"
)
print(answer)一个请求包含三个重要部分:
- model选择处理请求的Claude模型。不同模型在功能、速度和成本上可能有所不同。
- max_tokens限制Claude可以生成的最大文本量。较小的值可以减少延迟,但Claude可能在完成回答前停止生成。
- messages 包含对话内容。每条消息都有一个角色和内容。角色通常为 user 或 assistant。
例如,单次请求包含一条用户消息。多轮对话包含早期的用户和助手消息。
Claude 返回 response.content,这是一个包含类型化内容块的列表。常见块类型包括:
块类型
含义
text生成的文本
tool_use请求你的应用程序调用工具
thinking启用时的推理内容
示例中收集文本块,而不是假设 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,
)助手消息记录了 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保持摘要作为独立的应用程序状态,并在下一个请求中将其作为上下文包含。不要将其插入到最近的用户消息之前,因为这可能导致无效的连续用户消息。
在存储或传输之前,敏感信息也应进行脱敏处理:
import re
def redact(text):
return re.sub(
r"\b(?:\d[ -]?){13,16}\b",
"[REDACTED CARD]",
text,
)如何使用明确边界的提示结构
XML风格的标签是普通文本,不是特殊API命令。它们使提示的每个部分都显式化:
prompt = """
<customer_reviews>
产品很舒适,但可用颜色有限。
客户还将其描述为耐用。
</customer_reviews>
<sales_data>
一月:120件
二月:150件
三月: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": "shipped",
"ORD-1002": "processing",
}
return {
"order_id": order_id,
"status": orders.get(order_id, "not_found"),
}该函数接受订单ID,查找并返回可预测的数据。在生产环境中,字典将被数据库查询取代。Claude不会执行该函数,您的应用程序会执行。
使用模式描述该函数:
tools = [{
"name": "get_order_status",
"description": "获取客户订单的当前状态。",
"input_schema": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "如ORD-1001的订单ID。"
}
},
"required": ["order_id"],
},
}]Claude可能会返回一个工具使用块,而不是最终答案:
type="tool_use"
id="toolu_example"
name="get_order_status"
input={"order_id": "ORD-1001"}名称标识函数,输入包含其参数,返回结果时需要ID。停止原因“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>" )
/think
draft = ask(
f"<details>{details}</details>\n"
"<task>起草简洁的客服回复。</task>"
)
review = ask(
f"<draft>{draft}</draft>\n"
"<task>列出不支持的承诺,或回复OK。</task>"
)
return draft, review代理模式更具灵活性:Claude可以自主决定是否使用工具以及下一步操作。代理仍然需要验证和设置最大步数限制。上面的run_conversation()函数可以在代理循环中重复使用。
当步骤明确且需要重复执行时使用工作流模式;当下一步操作依赖当前结果时使用代理模式。
链式处理、并行处理、路由和评估器-优化器
链式处理将每个结果传递给下一阶段:
def chained_reply(ticket, policy):
draft = ask(
f"<ticket>{ticket}</ticket>\n"
"<task>起草客服回复。</task>"
)
issues = ask(
f"<policy>{policy}</policy>\n"
f"<draft>{draft}</draft>\n"
"<task>列出不支持的声明。</task>"
)
return ask(
f"<draft>{draft}</draft>\n"
f"<issues>{issues}</issues>\n"
"<task>重写最终回复。</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>返回refund、delivery或general中的一个。</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>撰写客服回复。</task>"
)
for _ in range(rounds):
review = ask(
f"<reply>{reply}</reply>\n"
"<task>列出准确性或语气问题,或回复PASS。</task>"
)
if review.strip().upper() == "PASS":
break
reply = ask(
f"<reply>{reply}</reply>\n"
f"<review>{review}</review>\n"
"<task>重写回复。</task>"
)
return reply使用链式处理处理依赖阶段,使用并行处理处理独立任务,使用路由处理专用路径,当需要额外API调用时使用评估器-优化器循环以保证质量。
如何评估提示质量
使用代表性测试用例:
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("Failed:", case["ticket"], result)
score = passed / len(cases)
print(f"{passed}/{len(cases)} passed")
return score对标签和 JSON 使用基于代码的评分器,对语气、准确性和有用性使用人工或基于模型的评分器。
结论
使用 Claude API 的构建过程不仅仅是编写提示。可靠的应用需要结构化的上下文、管理对话状态、验证工具执行、有意的响应处理、合适的流程以及可重复的评估。
目标不是找到一个完美的提示。而是围绕 Claude 构建一个系统,提供正确的上下文,限制不安全的操作,处理不确定性,并衡量更改是否改进了结果。
合作伙伴推荐的课程
(在新标签页中打开)
广告
- 在 Coursera 上:在 n8n 中集成 AI 模型:提示工程与输出 Nikhil Agarwal 和 LearnKartS 3 周,每周 2 小时 初学者 证书 查看课程(在新标签页中打开)
- 在 Coursera 上:使用 Anthropic 构建计算机使用 Colt Steele 1 小时 2 分钟 初学者 查看课程(在新标签页中打开)
- 在 Coursera 上:DeepSeek 入门 Edureka 3 周,每周 2 小时 初学者 证书 查看课程(在新标签页中打开)
- 在 Udemy 上:用于网络安全的 Claude Christopher Nett 4.5(132 条评分) 4 小时 初学者 证书 查看课程(在新标签页中打开)
Class Central 上的更多提示工程课程
→
我是一名资深软件开发人员,拥有领导团队和构建复杂网络平台(包括教育和医疗应用)的丰富经验。我在后端开发、AI 集成和云技术方面展示了深厚的技术专长,这进一步通过我在实时目标检测和 AI 领域的同行评审出版物得到证明。我的工作通过 Google Scholar 和 ResearchGate 索引平台的引用和贡献获得了认可。除了技术角色之外,我积极参与 ADPList 的导师计划,并担任 Generation Initiative 的校友顾问委员会成员,展示了我在提升数字技能和云技术方面的承诺。
如果这篇文章对你有帮助,请分享它。
免费学习编程。freeCodeCamp 的开源课程已帮助超过 40,000 人成为开发人员。立即开始