vLLM Function Call 全链路深度解析

resolved_chat_template = load_chat_template(args.chat_template)
# → cached_lru_cache(_load_chat_template)
# → 读文件 / 内置模板 / 字面量

def adjust_request(self, request):
    json_schema = get_json_schema_from_tools(tool_choice, tools)
    if json_schema is not None:
        request.structured_outputs = StructuredOutputsParams(json=json_schema)
    return request

# preprocess_chat() 里
request = reasoning_parser.adjust_request(request)   # 先改 reasoning 参数
request = tool_parser.adjust_request(request)         # 再加工具约束

async def generate(self, prompt, sampling_params, request_id, ...):
    q = await self.add_request(request_id, prompt, params, ...)
    # output_handler 后台 task 从 EngineCore 拉结果塞进队列
    while not finished:
        out = q.get_nowait() or await q.get()
        finished = out.finished
        yield out

async def add_request(self, ...):
    # 1. input_processor.process_inputs(prompt) → EngineCoreRequest
    request = self.input_processor.process_inputs(request_id, prompt, params, ...)
    # 2. output_processor.add_request(request)  → 登记 output 处理
    self.output_processor.add_request(request, ...)
    # 3. engine_core.add_request_async(request) → 发送到 EngineCore 进程
    await self.engine_core.add_request_async(request)

async for res in result_generator:
    # 等待完整输出

# 收集完整文本后
tool_call_info = tool_parser.extract_tool_calls(model_output, request)
# 一行提取，不需要状态

async for res in result_generator:          # 每个 step 迭代一次
    delta_text = output.text                # 本轮新增文本
    current_text = previous_text + delta_text

    # 按模型路径分流
    if is_mistral_grammar_path:
        result = tool_parser.extract_maybe_reasoning_and_tool_streaming(...)
    elif tool_choice_auto:
        result = tool_parser.extract_tool_calls_streaming(
            previous_text, current_text)

self.tool_call_start_token = "<tool_call>"
self.tool_call_end_token   = "</tool_call>"
self.tool_call_prefix      = "<function="
self.function_end_token    = "</function>"
self.parameter_prefix      = "<parameter="
self.parameter_end_token   = "</parameter>"

def extract_tool_calls(self, model_output, request):
    # 1. 快速过滤
    if self.tool_call_prefix not in model_output:
        return ExtractedToolCallInformation(tools_called=False, content=model_output)
    # 2. 正则找 <tool_call>...</tool_call>
    function_calls = self._get_function_calls(model_output)
    # 3. 逐个解析 XML → ToolCall
    tool_calls = [self._parse_xml_function_call(fc, request.tools) for fc in function_calls]
    # 4. 提取 tool_call 前面的内容
    content = model_output[:找到第一个 <tool_call> 的位置]
    return ExtractedToolCallInformation(tools_called=True, tool_calls=tool_calls, content=content)

is_tool_call_started: bool      # 是否已开始 tool call
in_function: bool               # 是否在 <function=...> 内部
in_param: bool                  # 是否在 <parameter=...> 内部
current_tool_index: int         # 当前处理第几个 tool call
header_sent: bool               # 函数名 header 是否已发
param_count: int                # 当前函数已处理完的参数个数
accumulated_params: dict        # 累积的参数名→值

def _get_tool_schema_from_tool(tool):
    name, params = _extract_tool_info(tool)
    return {
        "properties": {
            "name": {"type": "string", "enum": [name]},
            "parameters": params,
        },
        "required": ["name", "parameters"],
    }

def extract_tool_calls_streaming(self, previous_text, current_text, delta_text, ...):
    # 阶段 1: 无状态完整解析 current_text
    full_state = self._parse_full_text(current_text)
    # 阶段 2: 对比缓存的上次状态 → 计算 delta
    delta = self._diff_state(self._cached_state, full_state)
    # 阶段 3: 更新缓存
    self._cached_state = full_state
    return self._delta_to_message(delta)