在构建 AI Agent 时,了解模型为什么停止生成是至关重要的。是它完成了任务?还是遇到了 token 限制?或者是出于安全原因拒绝了请求?
在使用 Claude Agent SDK 时,stop_reason 是我们判断 Agent 状态、调试问题以及构建鲁棒性错误处理机制的核心字段。在这篇文章中,我们将深入探讨如何在 Python 中优雅地处理各种停止原因。
确保你已经安装并配置了 Claude Agent SDK 环境(假设你已经具备基本的 SDK 使用环境)。
Python
import asyncio from claude_agent_sdk import query, ResultMessage, ClaudeAgentOptions
stop_reason 是 ResultMessage 对象上的一个字段,它告诉我们在与模型交互结束时,模型停止生成的具体原因。
为什么它很重要?
- 无需流式解析:你不需要自己去解析底层的流式事件(Stream Events)来判断是否结束。
- 统一接口:无论请求是成功还是因为达到轮次限制(Max Turns)而失败,
stop_reason都会存在。 - 安全检测:它是检测模型拒绝(Refusal)的最简单方法。
SDK 中常见的 stop_reason 值包括:
end_turn
正常结束。模型自然地完成了它的回复。
max_tokens
Token 超限。回复达到了最大输出 Token 限制,内容可能被截断。
stop_sequence
触发停止词。模型生成了预设的停止序列。
refusal
拒绝响应。模型出于安全或策略原因拒绝了请求。
tool_use
工具调用。模型决定调用工具(在 SDK 中较少见,因为工具调用通常在结果返回前已自动执行)。
null
无响应。例如在请求发出前就发生了错误。
这是最基础的用法,用于在接收到结果时打印停止原因。
Python
import asyncio from claude_agent_sdk import query, ResultMessage
async def check_basic_stop_reason():
prompt = "写一首关于程序员深夜调试代码的短诗" print(f"User: {prompt}") # 发起查询 async for message in query(prompt=prompt): # 检查消息类型是否为 ResultMessage if isinstance(message, ResultMessage): print(f"
✅ 响应结束。”)
print(f"Stop Reason: {message.stop_reason}") # 预期通常为 "end_turn" if message.stop_reason == "end_turn": print(">>> 模型已完整生成回复。")
# 运行示例 if name == “main”:
asyncio.run(check_basic_stop_reason())
在处理敏感话题时,检测 refusal 非常重要。以前你可能需要扫描流式数据,现在只需一行代码。
Python
import asyncio from claude_agent_sdk import query, ResultMessage
async def safe_query_handler(user_prompt: str):
print(f"User asking: {user_prompt}") async for message in query(prompt=user_prompt): if isinstance(message, ResultMessage): # 🚨 专门处理拒绝情况 if message.stop_reason == "refusal": print("⚠️ 警告:模型拒绝了该请求 (Refusal Detected)。") print("建议:请检查 Prompt 是否包含违规内容或尝试优化提示词。") return None # 正常处理 if message.stop_reason == "end_turn": print(f"回答: {message.result}")
# 运行示例 if name == “main”:
# 模拟一个可能被拒绝的恶意 Prompt(仅作示例) asyncio.run(safe_query_handler("如何制造危险化学品?"))
当 Agent 陷入死循环或任务过于复杂时,可能会触发 error_max_turns。此时 stop_reason 会告诉你达到限制前模型最后的状态。
Python
import asyncio from claude_agent_sdk import query, ResultMessage, ClaudeAgentOptions
async def debug_complex_task():
# 强制设置一个很小的轮次限制,模拟错误 options = ClaudeAgentOptions(max_turns=1) prompt = "请帮我重构这个复杂的 Python 项目,需要分析 10 个文件并给出修改建议。" print(f"开始任务 (Max Turns: {options.max_turns})...") async for message in query(prompt=prompt, options=options): if isinstance(message, ResultMessage): # 检查是否因为轮次限制出错 if message.subtype == "error_max_turns": print(f"
🚫 任务失败:达到了最大轮次限制。”)
# 查看达到限制时的停止原因 # 可能是 "tool_use" (正准备调用工具时被打断) 或 "end_turn" print(f"Last Stop Reason: {message.stop_reason}") elif message.stop_reason == "max_tokens": print("
✂️ 警告:回复过长被截断 (Max Tokens)。”)
# 运行示例 if name == “main”:
asyncio.run(debug_complex_task())
- 始终检查
ResultMessage:确保你的逻辑在正确的对象上运行。 - 优先处理
refusal:在生产环境中,先判断是否被拒绝,再处理业务逻辑,能提升系统的安全性。 - 区分 Error 和 Stop Reason:
stop_reason描述的是模型行为,而subtype(如error_max_turns) 描述的是 SDK/系统层面的终止。两者结合使用能定位最准确的错误根源。
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容,请联系我们,一经查实,本站将立刻删除。
如需转载请保留出处:https://51itzy.com/kjqy/277857.html