★ FEATURED ARTICLE 这几个月连续做了几个大模型应用项目Qwen3思考标记处理几乎是每次接新项目都要重做一遍的活。特别是用Qwen3系列模型做Agent和对话产品时|start|和|end|这两个标记无处不在——我第一次在流式返回里看到它们还以为是解析bug停下来排查了半天才发现是模型自己在思考模式下给推理过程画了一条边界线。这篇文章不聊榜单不聊参数只聊一个实际问题拿到带思考标记的模型输出后怎么稳定、不丢信息地把“思考内容”和“最终结果”拆开并在流式、API透传、Agent上下文三个典型场景里合理处置。适合最近在接Qwen3接口的开发者也适合本地部署Qwen3刚被标记整懵的同学。1. 理解Qwen3思考标记它到底是个什么东西1.1 从隐式思考到显式边界传统大模型的输出每个字符都直接是送给用户的回答。模型内部有没有推理过程有但藏在一层层神经网络权重里外部拿不到。强化学习推理模型这代玩法变了模型被训练成先把推理过程“说出来”再给结论。于是模型输出就变成了一段内部思路加一段最终答复。问题来了——下游系统怎么区分这两段如果只按标点、换行去猜极不可靠。所以有了思考标记。Qwen3在思考模式下把推理过程用|start|和|end|包起来标记之外的才是给用户看的最终回复。这个设计看着简单实际非常关键它相当于模型主动给下游程序划了一条机器可读的边界线让“拆思考”从玄学变成了确定性问题。我理解这套机制像是给快递加了分隔板车厢里哪块是运输单、哪块是货品拆箱的人一眼就能分清。没有分隔板之前只能在纸上写“请把这段话当备注看”模型听不听全看心情。1.2 标记的实际形态一个示例讲清楚Qwen3思考模式下的输出整体长这样|start|用户想让我总结这篇文章我需要先提取核心观点再按逻辑重新组织语言。文章的主要脉络有三个背景、问题、结论……我先列一个草稿再检查是否遗漏。|end|这篇文章的核心要点如下第一……第二……第三……注意几个细节。第一|start|后面可能直接接思考内容没有额外换行第二|end|后面接最终输出同样通常没有换行第三如果模型没有进入思考模式输出里可能压根没有标记直接就是回答。在OpenAI兼容接口里这段完整文本会出现在响应的choices[0].message.content字段中。如果你走流式它会变成几十上百个chunk每个chunk的delta.content里只有几个字符标记被劈成两半属于常规操作。那么关键问题就来了如何用参数控制思考模式通过OpenAI兼容API调用时常见做法是在extra_body里传enable_thinking参数from openai import OpenAI client OpenAI( api_key..., base_url..., ) resp client.chat.completions.create( model你的qwen3模型名, messages[{role: user, content: 用一句话说明量子计算是什么}], extra_body{enable_thinking: False}, ) print(resp.choices[0].message.content)关闭思考后输出通常更简短、首字延迟更低适合翻译、改写、关键词抽取这类不需要复杂推理的任务。开启思考则适合代码生成、数学题、多步规划。这个参数不用每次都传取决于部署端默认配置但代码里显式传一下更可控。1.3 双模式带来的最大坑不确定性接Qwen3最容易被咬的地方不是如何解析标记而是“这次有没有标记”这件事本身不确定。同样是enable_thinking没关的接口同一个问题第一次命中缓存时返回可能直接是最终答案第二次重新计算时又带上了思考标记不同部署配置下甚至同一个系统提示词都会导致模型自行决定要不要思考。结果就是你写的解析逻辑不能假设“一定有标记”或者“一定没标记”必须两种都兼容。我在早期版本里犯过一个错直接按“去掉|start|到|end|之间的部分”写了剥离逻辑结果遇到非思考输出时整段文本被原样返回界面倒是没报错但用户看到的回答里偶尔出现大段碎碎念一度被当成模型幻觉。后来改成先检测标记、再决定怎么处理的写法才稳定下来。2. 思考标记处理的三个高频业务场景2.1 场景一流式输出标记被切成两半对话产品里几乎必用流式。流式返回时服务端把内容拆成一个一个分块推给客户端一个完整的|start|可能被切到三个chunk里chunk 1: |sta chunk 2: rt|用户希望我给出 chunk 3: 一个简洁的总结那么我应该先梳理观点……|end|这是最终总结如果解析代码直接chunk.find(|start|)第一个chunk永远是找不到的因为单看它只是一个字符串截断。要么你把所有chunk先拼接成完整文本再统一解析要么用一个带缓冲的状态机把不完整的尾部留在缓冲里等下一个chunk补全。这两种方式我都实际用过。非流式接口或对首字延迟不敏感的场景直接拼完整再解析最简单代码好维护也方便事后日志排查。流式场景想保持“边收边出”的体验就得用状态机方案后面的第三章会给可复用的实现。2.2 场景二API透传与前端展示思考内容给不给人看接到接口之后下一个问题是思考内容怎么办。好多第一次接入的同学会直接把thinking拼到UI上让用户看到一大段“我应该先……然后……等等这么不对……换一种思路”效果通常不太行。原因有三点。一是观感思考过程里有大量自我纠正、口头禅和半截句子用户看了会觉得模型“说话颠三倒四”二是效率思考内容往往比最终回答还长原样展示会拉长阅读成本三是信息边界思考里可能包含模型对用户指令的推测、对上下文不足的抱怨这些内容并不适合让终端用户看到。我的做法是分场景处理。C端产品默认只展示最终回答最多在等待阶段用“正在思考”动画过渡高级设置里给一个“查看思考过程”的折叠框B端或调试面板则完整保留思考内容因为排查问题时思考过程是最快定位失败原因的入口。另外一个已经被验证有效的折中方案是不展示原始思考文本而是由另一层逻辑把思考压缩成几个要点比如“正在分析文章结构…正在检查遗漏…”观感和信息量都能兼顾。2.3 场景三Agent上下文思考内容要不要回传做Agent项目时思考标记处理的意义比普通对话更重。Qwen3在思考模式下经常是这样先输出一段内部推理再在回答里给出行动方案或调用工具。完整链路是——|start|用户要找昨天提到的那个报告我先在记忆里检索一下看看报告存在哪里……找到了在项目文档目录下。|end|我找到了报告它位于项目文档目录下需要我打开给你看吗如果只把“最终回答”传给下一轮模型就丢了它在thinking里推理出来的关键线索如果原样把thinking传给下一轮又会把标记和推理文本混进对话历史时间一长上下文里全是碎碎念。正确的做法是把thinking截取出来作为一条内部推理记录单独存放可以和工具结果、业务日志一起组成状态再在下一次决策时以适当的形式拼进上下文。注意不要自己在system prompt或历史消息里手动编|start|标记试图“诱导”模型思考——模型有自己的标记格式约定用户编造的标记往往不会生效反而扰乱注意力。你只要把上一轮的thinking和answer都保留需要时以自然语言形式说“你之前的推理过程如下……”即可。3. 动手实现两套可以直接落地的解析方案3.1 方案一完整文本正则解析简单可靠如果你用的是非流式接口或者已经在别处完成了流式拼接那么解析交给一个函数就够了。核心思路是先找|start|再在它之后找第一个|end|找到就切开找不到就说明思考内容不完整按截断处理。import re START_TAG |start| END_TAG |end| def parse_qwen3_response(content: str): 解析Qwen3响应返回 (thinking, answer)。 无思考标记时thinking为空字符串answer为完整内容。 有开始标记但没有结束标记时认为思考被截断 将start之后的内容全部归入thinkinganswer为空。 if not content: return , start_pos content.find(START_TAG) if start_pos -1: return , content end_pos content.find(END_TAG, start_pos len(START_TAG)) if end_pos -1: thinking content[start_pos len(START_TAG):] return thinking, thinking content[start_pos len(START_TAG):end_pos] answer content[:start_pos] content[end_pos len(END_TAG):] return thinking, answer为什么从start_pos len(START_TAG)开始找end因为end标记必须出现在start之后跳过start本身可以避免思考内容里恰好以|start|开头这种极端情况造成误判。为什么剥离时把content[:start_pos]并入answer而不是直接丢弃因为极端情况下模型可能在输出开头多打了几个空格或杂散字符把这些归入回答部分比直接丢掉更安全。3.2 方案二流式解析状态机兼容任意分块流式场景想做到“边收边拆”需要用有状态的对象。下面这个Qwen3StreamParser我实测下来可以作为通用组件直接复制就能用。class Qwen3StreamParser: Qwen3思考标记的流式解析器兼容标记被拆到多个chunk的情况。 用法 parser Qwen3StreamParser() for chunk in stream: delta chunk.choices[0].delta.content if delta: parser.feed(delta) parser.flush() print(parser.thinking) print(parser.answer) START_TAG |start| END_TAG |end| def __init__(self): self.buffer self.state answer # answer - thinking - answer self.thinking_parts [] self.answer_parts [] self.has_think_tag False def feed(self, delta: str): self.buffer delta self._process() def _process(self): while self.buffer: if self.state answer: pos self.buffer.find(self.START_TAG) if pos -1: # 没找到完整标记。缓冲尾部有可能是标记前缀先留几个字符等下一轮。 keep_len min(len(self.START_TAG) - 1, len(self.buffer)) out_len len(self.buffer) - keep_len if out_len 0: self.answer_parts.append(self.buffer[:out_len]) self.buffer self.buffer[out_len:] else: break else: if pos 0: self.answer_parts.append(self.buffer[:pos]) self.buffer self.buffer[pos len(self.START_TAG):] self.state thinking self.has_think_tag True elif self.state thinking: pos self.buffer.find(self.END_TAG) if pos -1: keep_len min(len(self.END_TAG) - 1, len(self.buffer)) out_len len(self.buffer) - keep_len if out_len 0: self.thinking_parts.append(self.buffer[:out_len]) self.buffer self.buffer[out_len:] else: break else: if pos 0: self.thinking_parts.append(self.buffer[:pos]) self.buffer self.buffer[pos len(self.END_TAG):] self.state answer else: # 理论上不会走到这里兜底而已 self.answer_parts.append(self.buffer) self.buffer def flush(self): if self.buffer: if self.state thinking: self.thinking_parts.append(self.buffer) else: self.answer_parts.append(self.buffer) self.buffer property def thinking(self) - str: return .join(self.thinking_parts) property def answer(self) - str: return .join(self.answer_parts)这个状态机的核心是“保守输出留尾巴等下一轮”。无论当前有无标记每次只把缓冲区里确定不会构成标记前缀的部分送出去剩下几个字符等下一个chunk来补全。这样哪怕|start|被切成|sta和rt|最终也能正确识别成完整标记不会把半个标签当普通文本输出。在answer状态下如果模型输出里压根没有思考标记所有文本都会自然汇入answer_parts解析结果等价于“thinking为空answer为全文”。3.3 落地细节测试用例如下这里分享一组我放在单测里的用例覆盖了三种常见情况。def test_no_think_tag(): parser Qwen3StreamParser() for c in 直接返回结果不包含思考标记: parser.feed(c) parser.flush() assert parser.thinking assert parser.answer 直接返回结果不包含思考标记 def test_full_think(): parser Qwen3StreamParser() for chunk in [|start|, 这是思考, |end|, 这是答案]: parser.feed(chunk) parser.flush() assert parser.has_think_tag assert parser.thinking 这是思考 assert parser.answer 这是答案 def test_split_tag(): parser Qwen3StreamParser() chunks [|sta, rt|思考|, end|最终回答] for chunk in chunks: parser.feed(chunk) parser.flush() assert parser.has_think_tag assert parser.thinking 思考 assert parser.answer 最终回答 def test_truncated(): # 思考被截断只有start没有end parser Qwen3StreamParser() for chunk in [|start|, 思考到一半]: parser.feed(chunk) parser.flush() assert parser.thinking 思考到一半 assert parser.answer 这几个测试跑通基本可以放心接各种来源的Qwen3输出。要注意的是代码里假设了每条消息只出现一对思考标记实际业务中我还没见过模型输出多对start/end的情况如果将来遇到状态机方案也只会处理第一对剩余文本会原样进入answer不会崩。4. 实战踩坑常见问题与排查速查表4.1 五个高频问题逐个说解法有start没有end。最常见的原因有两个一是max_tokens不够思考还没完就被硬截断二是模型在思考中途遇到停止条件提前结束。解法我上面已经写进解析函数了——end找不到时start之后的内容全部算thinkinganswer留空然后业务侧再决定重试还是提示用户。同一请求有时有标记有时没有。服务端缓存命中、部署配置、采样参数都可能影响模型是否走思考路径。解法是解析逻辑永远兼容无标记情况不要在代码里假设“有或无”。流式首字延迟突然变高。思考模式下模型会先在prefill阶段算出一大段推理再开始逐字输出所以首字比非思考模式慢不少。实测同一个问题非思考模式可能一两百毫秒出第一个字思考模式拉长到几秒都不稀奇。解法UI层提前渲染“正在思考”占位或者对低难度请求关闭思考。思考内容太长影响使用。长思考会放大延迟和token成本。解法是设置合理的max_tokens上限同时接受“思考被截断”的可能并在解析层加一个思考长度阈值超过后强制截断并只保留摘要。模型回答里出现|end|文本。如果最终回答本身包含这串字符简单的全局正则剥离会误伤。我给的parse_qwen3_response只切第一对标记后半段的|end|会被原样保留在answer里不会误删。4.2 排查速查表症状可能原因处理建议输出中有start但找不到同一输入有时有标记有时没有缓存命中或服务端策略解析逻辑兼容无标记返回流式首字延迟明显增加思考模式prefill阶段长提早渲染占位或关闭思考模式思考内容比答案长好几倍任务复杂度过高或提示词诱导过度思考限制max_tokens提示“简洁思考”直接展示thinking被用户吐槽思考内容不适合直接读C端隐藏或折叠B端保留用于排查标记被切成多个chunk导致漏解析流式按块find必然踩坑用带缓冲的状态机方案4.3 两个额外建议第一一定要把thinking单独存一份日志字段。线上出了问题光看answer很难判断模型是不是理解错了指令有thinking在很快就能定位到是提示词问题还是解析问题。第二如果你在本地部署Qwen3确认下你的推理框架是否已经帮你处理过标记。有的框架在输出预处理阶段就自动剥掉了|start|和|end|这时候你再写解析逻辑反而会认为“所有输出都没有思考标记”。这不是坏事但要把这个前提记录下来避免上线后两套逻辑打架。我自己在实际项目里的体会是思考标记处理看起来是个几行代码的小事但它卡在所有大模型应用的必经之路上。流式场景拆不好用户看到的是残缺标记Agent场景拆不好模型自己推理出的关键信息就丢了展示场景拆不好产品观感直接崩。把这套逻辑沉淀成公共组件后面每个新接入的项目都能少走弯路。最后再提醒一点flush()记得在流结束时必须调用很多同学测试时没走流式、直接一口气传完整文本解析没问题一旦换到真实流式就漏掉尾部内容排查了很久才发现是没flush。如果是接入OpenAI兼容SDK记得从delta.content取文本而不是从message.content取否则拿到的永远是最后一轮的空字符串。
阅读完成 · 觉得有帮助？
