Qwen3思考标记解析实战：流式处理与Agent场景下的标记剥离方案

这几个月连续做了几个大模型应用项目，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之后的内容全部归入thinking，answer为空。 """ 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之后的内容全部算thinking，answer留空，然后业务侧再决定重试还是提示用户。

同一请求有时有标记有时没有。服务端缓存命中、部署配置、采样参数都可能影响模型是否走思考路径。解法是解析逻辑永远兼容无标记情况，不要在代码里假设“有或无”。

流式首字延迟突然变高。思考模式下，模型会先在prefill阶段算出一大段推理，再开始逐字输出，所以首字比非思考模式慢不少。实测同一个问题，非思考模式可能一两百毫秒出第一个字，思考模式拉长到几秒都不稀奇。解法：UI层提前渲染“正在思考”占位，或者对低难度请求关闭思考。

思考内容太长，影响使用。长思考会放大延迟和token成本。解法是设置合理的max_tokens上限，同时接受“思考被截断”的可能，并在解析层加一个思考长度阈值，超过后强制截断并只保留摘要。

模型回答里出现<|end|>文本。如果最终回答本身包含这串字符，简单的全局正则剥离会误伤。我给的parse_qwen3_response只切第一对标记，后半段的<|end|>会被原样保留在answer里，不会误删。

4.2 排查速查表

4.3 两个额外建议

第一，一定要把thinking单独存一份日志字段。线上出了问题，光看answer很难判断模型是不是理解错了指令，有thinking在，很快就能定位到是提示词问题还是解析问题。

第二，如果你在本地部署Qwen3，确认下你的推理框架是否已经帮你处理过标记。有的框架在输出预处理阶段就自动剥掉了<|start|>和<|end|>，这时候你再写解析逻辑反而会认为“所有输出都没有思考标记”。这不是坏事，但要把这个前提记录下来，避免上线后两套逻辑打架。

我自己在实际项目里的体会是：思考标记处理看起来是个几行代码的小事，但它卡在所有大模型应用的必经之路上。流式场景拆不好，用户看到的是残缺标记；Agent场景拆不好，模型自己推理出的关键信息就丢了；展示场景拆不好，产品观感直接崩。把这套逻辑沉淀成公共组件，后面每个新接入的项目都能少走弯路。

最后再提醒一点：flush()记得在流结束时必须调用，很多同学测试时没走流式、直接一口气传完整文本，解析没问题，一旦换到真实流式就漏掉尾部内容，排查了很久才发现是没flush。如果是接入OpenAI兼容SDK，记得从delta.content取文本，而不是从message.content取，否则拿到的永远是最后一轮的空字符串。