推理输出¶

from openai import OpenAI

# 修改 OpenAI 的 API 密钥和 API 基地址以使用 vLLM 的 API 服务器。
openai_api_key = "EMPTY"
openai_api_base = "http://localhost:8000/v1"

client = OpenAI(
    api_key=openai_api_key,
    base_url=openai_api_base,
)

models = client.models.list()
model = models.data[0].id

# 第一轮
messages = [{"role": "user", "content": "9.11 and 9.8, which is greater?"}]
# 对于 granite，添加：`extra_body={"chat_template_kwargs": {"thinking": True}}`
# 对于 Qwen3 系列，如果您想在推理模式下禁用推理，添加：
# extra_body={"chat_template_kwargs": {"enable_thinking": False}}
response = client.chat.completions.create(model=model, messages=messages)

reasoning = response.choices[0].message.reasoning
content = response.choices[0].message.content

print("reasoning:", reasoning)
print("content:", content)

from openai import OpenAI

# 修改 OpenAI 的 API 密钥和 API 基地址以使用 vLLM 的 API 服务器。
openai_api_key = "EMPTY"
openai_api_base = "http://localhost:8000/v1"

client = OpenAI(
    api_key=openai_api_key,
    base_url=openai_api_base,
)

models = client.models.list()
model = models.data[0].id

messages = [{"role": "user", "content": "9.11 and 9.8, which is greater?"}]
# 对于 granite，添加：`extra_body={"chat_template_kwargs": {"thinking": True}}`
# 对于 Qwen3 系列，如果您想在推理模式下禁用推理，添加：
# extra_body={"chat_template_kwargs": {"enable_thinking": False}}
stream = client.chat.completions.create(
    model=model,
    messages=messages,
    stream=True,
)

print("client: Start streaming chat completions...")
printed_reasoning = False
printed_content = False

for chunk in stream:
    # 安全地从 delta 中提取 reasoning 和 content，
    # 如果属性不存在或为空字符串则默认为 None
    reasoning = (
        getattr(chunk.choices[0].delta, "reasoning", None) or None
    )
    content = getattr(chunk.choices[0].delta, "content", None) or None

    if reasoning is not None:
        if not printed_reasoning:
            printed_reasoning = True
            print("reasoning:", end="", flush=True)
        print(reasoning, end="", flush=True)
    elif content is not None:
        if not printed_content:
            printed_content = True
            print("\ncontent:", end="", flush=True)
        # 提取并打印内容
        print(content, end="", flush=True)

from openai import OpenAI

client = OpenAI(base_url="http://localhost:8000/v1", api_key="dummy")

# 导入所需的包

from vllm.reasoning import ReasoningParser, ReasoningParserManager
from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest
from vllm.entrypoints.openai.engine.protocol import DeltaMessage

# 定义推理解析器并注册到 vllm
# register_module 中的名称列表可用于
# --reasoning-parser 参数。
class ExampleParser(ReasoningParser):
    def __init__(self, tokenizer: TokenizerLike):
        super().__init__(tokenizer)

    def extract_reasoning_streaming(
        self,
        previous_text: str,
        current_text: str,
        delta_text: str,
        previous_token_ids: Sequence[int],
        current_token_ids: Sequence[int],
        delta_token_ids: Sequence[int],
    ) -> DeltaMessage | None:
        """
        应该实现的实例方法，用于从不完整的响应中提取推理内容；
        用于处理推理调用和流式传输。必须是实例方法，因为它需要状态 -
        当前的 token/差异，以及关于之前解析和提取的信息（参见构造函数）
        """

    def extract_reasoning(
        self,
        model_output: str,
        request: ChatCompletionRequest | ResponsesRequest,
    ) -> tuple[str | None, str | None]:
        """
        从完整的模型生成字符串中提取推理内容。

        用于非流式响应，此时在发送给客户端之前我们拥有整个模型响应。

        参数：
        model_output: str
            要从中提取推理内容的模型生成字符串。

        request: ChatCompletionRequest
            用于生成 model_output 的请求对象。

        返回：
        tuple[Optional[str], Optional[str]]
            包含推理内容和内容的元组。
        """
# 注册推理解析器
ReasoningParserManager.register_lazy_module(
    name="example",
    module_path="vllm.reasoning.example_reasoning_parser",
    class_name="ExampleParser",
)

@dataclass
class DeepSeekReasoner(Reasoner):
    """
    DeepSeek R 系列模型的推理器。
    """
    start_token_id: int
    end_token_id: int

    start_token: str = "<tool_call>"
    end_token: str = "</tool_call>"

    @classmethod
    def from_tokenizer(cls, tokenizer: PreTrainedTokenizer) -> Reasoner:
        return cls(
            start_token_id=tokenizer.encode("<tool_call>", add_special_tokens=False)[0],
            end_token_id=tokenizer.encode("</tool_call>", add_special_tokens=False)[0],
        )

    def is_reasoning_end(self, input_ids: list[int]) -> bool:
        return self.end_token_id in input_ids

    def is_reasoning_end_streaming(self, input_ids: list[int], delta_ids: list[int]) -> bool:
        return self.end_token_id in delta_ids
    ...