diff --git a/.dockerignore b/.dockerignore index 1914e14ab8d1..851970935751 100644 --- a/.dockerignore +++ b/.dockerignore @@ -12,3 +12,8 @@ examples/**/*.onnx examples/**/*.safetensors examples/**/c-model examples/models/core/gpt/gpt* + +cpp/build +.git +*.whl +trie_decoding diff --git a/cpp/include/tensorrt_llm/batch_manager/guidedDecoder.h b/cpp/include/tensorrt_llm/batch_manager/guidedDecoder.h index 9a577b61ad51..a0c8b91c9340 100644 --- a/cpp/include/tensorrt_llm/batch_manager/guidedDecoder.h +++ b/cpp/include/tensorrt_llm/batch_manager/guidedDecoder.h @@ -1,5 +1,5 @@ /* - * Copyright (c) 2024, NVIDIA CORPORATION. All rights reserved. + * Copyright (c) 2024-2026, NVIDIA CORPORATION. All rights reserved. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,10 @@ #include "tensorrt_llm/runtime/bufferManager.h" #include "tensorrt_llm/runtime/iTensor.h" +#include +#include +#include + namespace xgrammar { class GrammarMatcher; @@ -31,6 +35,9 @@ namespace tensorrt_llm::batch_manager { class DecoderInputBuffers; +// Forward declaration for trie-based constraint handling (flat CSR representation) +class GuidedTrie; + class GuidedDecoder { public: @@ -39,29 +46,54 @@ class GuidedDecoder using BitmaskT = uint32_t; GuidedDecoder(executor::GuidedDecodingConfig const& guidedDecodingConfig, SizeType32 maxNumSequences, - SizeType32 vocabSizePadded, nvinfer1::DataType logitsDtype, runtime::BufferManager const& runtimeBufferManager); + SizeType32 maxBeamWidth, SizeType32 vocabSizePadded, nvinfer1::DataType logitsDtype, + runtime::BufferManager const& runtimeBufferManager); void build(ScheduledRequests const& scheduledRequests); void execute(DecoderInputBuffers const& decoderInputBuffers, runtime::BufferManager const& runtimeBufferManager); + /// @brief TRIE constraints are applied per beam; at beam width > 1 the caller must refresh + /// each request's beam token histories (gatherTree) before build(), because the raw slot + /// buffers go stale when beam search switches hypothesis parents. + [[nodiscard]] bool needsGatheredBeamTokens() const + { + return mGuidedDecodingBackend == executor::GuidedDecodingConfig::GuidedDecodingBackend::kTRIE; + } + private: executor::GuidedDecodingConfig::GuidedDecodingBackend mGuidedDecodingBackend; std::vector> mXGrammarMatchers; std::shared_ptr mXGrammarCompiler; + // Trie-based constraint handling (flat CSR trie; nodes are int32 indices, root = 0). + // No persistent per-request node state: each step re-walks every beam's generated tokens + // (bounded by the trie depth) so beam reordering cannot desynchronize the constraint. + std::shared_ptr mTrie; + std::unordered_map> mTrieBitmaskCache; + std::vector mEosOnlyBitmask; // fallback for off-trie beams: allow only EOS + std::mutex mTrieCacheMutex; + SizeType32 mMaxNumSequences; + SizeType32 mMaxBeamWidth; SizeType32 mVocabSizePadded; SizeType32 mBitmaskSize; // CeilDiv(vocabSizePadded, 32) nvinfer1::DataType mLogitsDtype; - TensorPtr mLogitsBitmask; // [mMaxNumRequests, mBitmaskSize] - TensorPtr mLogitsBitmaskHost; // [mMaxNumRequests, mBitmaskSize] - TensorPtr mLogitsBitmaskPtrVec; // [mMaxNumRequests], pointers to the logitsBitmask in a batch - TensorPtr mLogitsBitmaskPtrVecHost; // [mMaxNumRequests] - TensorPtr mLogitsPtrVec; // [mMaxNumRequests], pointers to the logits in a batch - TensorPtr mLogitsPtrVecHost; // [mMaxNumRequests] + // Bitmask row layout: one row per (seqSlot, beam): row = seqSlot * mMaxBeamWidth + beam. + // The xgrammar backend keeps its per-request design and uses beam 0's row. + TensorPtr mLogitsBitmask; // [mMaxNumSequences * mMaxBeamWidth, mBitmaskSize] + TensorPtr mLogitsBitmaskHost; // [mMaxNumSequences * mMaxBeamWidth, mBitmaskSize] + TensorPtr mLogitsBitmaskPtrVec; // [mMaxNumSequences * mMaxBeamWidth] + TensorPtr mLogitsBitmaskPtrVecHost; // [mMaxNumSequences * mMaxBeamWidth] + TensorPtr mLogitsPtrVec; // [mMaxNumSequences * mMaxBeamWidth] + TensorPtr mLogitsPtrVecHost; // [mMaxNumSequences * mMaxBeamWidth] // BufferManager with a dedicated stream for async copy of buffers for guided decoding. runtime::BufferManager mCopyBufferManager; + + // Helper methods for trie + void initTrieBuffers(runtime::BufferManager const& runtimeBufferManager); + std::vector const& getTrieBitmask(int32_t node); + void createTrieBitmask(int32_t node, std::vector& bitmask); }; } // namespace tensorrt_llm::batch_manager diff --git a/cpp/include/tensorrt_llm/executor/executor.h b/cpp/include/tensorrt_llm/executor/executor.h index 57c82aabb6a1..68234893deec 100644 --- a/cpp/include/tensorrt_llm/executor/executor.h +++ b/cpp/include/tensorrt_llm/executor/executor.h @@ -1431,12 +1431,15 @@ class GuidedDecodingConfig kXGRAMMAR = 0, /// @brief Enable guided decoding with LLGuidance backend. kLLGUIDANCE = 1, + /// @brief Enable guided decoding with Trie-based constraint backend. + kTRIE = 2, }; explicit GuidedDecodingConfig(GuidedDecodingBackend backend, std::optional> encodedVocab = std::nullopt, std::optional tokenizerStr = std::nullopt, - std::optional> stopTokenIds = std::nullopt); + std::optional> stopTokenIds = std::nullopt, + std::optional triePath = std::nullopt); bool operator==(GuidedDecodingConfig const& other) const; @@ -1452,6 +1455,9 @@ class GuidedDecodingConfig void setStopTokenIds(std::vector const& stopTokenIds); [[nodiscard]] std::optional> getStopTokenIds() const; + void setTriePath(std::string const& triePath); + [[nodiscard]] std::optional getTriePath() const; + void validate() const; private: @@ -1472,6 +1478,8 @@ class GuidedDecodingConfig std::optional mTokenizerStr; /// @brief Stop token ids. If not provided, it can be automatically detected. std::optional> mStopTokenIds; + /// @brief Path to binary trie file for TRIE backend. + std::optional mTriePath; }; class LogitsPostProcessorConfig diff --git a/cpp/tensorrt_llm/batch_manager/guidedDecoder.cpp b/cpp/tensorrt_llm/batch_manager/guidedDecoder.cpp index cb2264ec8003..15c4ee13e33e 100644 --- a/cpp/tensorrt_llm/batch_manager/guidedDecoder.cpp +++ b/cpp/tensorrt_llm/batch_manager/guidedDecoder.cpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 * * Licensed under the Apache License, Version 2.0 (the "License"); @@ -24,15 +24,137 @@ #include #include +#include +#include +#include + using namespace tensorrt_llm::runtime; namespace tensorrt_llm::batch_manager { +// Flat CSR trie for constrained decoding. Nodes are int32 indices (root = 0); children are +// stored in contiguous, per-node token-sorted ranges (binary-searchable) instead of a +// per-node unordered_map. This drops per-node/per-edge heap + hashmap +// overhead by ~30x (e.g. ~20GB -> ~1GB for the gen_search 73M-SID trie). Terminal nodes carry +// isEnd; the masker allows the trie's EOS token at terminal nodes so no explicit EOS leaves are +// needed (halving the node count vs. materializing "...,EOS" paths). +class GuidedTrie +{ +public: + explicit GuidedTrie(SizeType32 vocabSize) + : mVocabSize(vocabSize) + { + } + + void loadFromBinary(std::string const& path) + { + std::ifstream file(path, std::ios::binary); + TLLM_CHECK_WITH_INFO(file.is_open(), "Failed to open trie file: %s", path.c_str()); + + // Header: magic, version, numNodes, numEdges, eosTokenId. + uint32_t magic; + file.read(reinterpret_cast(&magic), sizeof(magic)); + TLLM_CHECK_WITH_INFO(magic == 0x54524945u, "Invalid trie file format: magic=0x%08X", magic); // 'TRIE' + + uint32_t version; + file.read(reinterpret_cast(&version), sizeof(version)); + TLLM_CHECK_WITH_INFO(version == 2u, "Unsupported trie format version: %u (expected 2)", version); + + uint32_t numNodes, numEdges; + file.read(reinterpret_cast(&numNodes), sizeof(numNodes)); + file.read(reinterpret_cast(&numEdges), sizeof(numEdges)); + file.read(reinterpret_cast(&mEosTokenId), sizeof(mEosTokenId)); + + mNumNodes = static_cast(numNodes); + + // Native CSR arrays stored contiguously; read directly (O(file size), no per-node parse). + // The builder writes each node's children token-sorted, so findChild can binary-search. + mChildOffsets.resize(static_cast(numNodes) + 1); + mChildTokens.resize(numEdges); + mChildNodes.resize(numEdges); + mIsEnd.resize(numNodes); + file.read(reinterpret_cast(mChildOffsets.data()), sizeof(uint32_t) * (static_cast(numNodes) + 1)); + file.read(reinterpret_cast(mChildTokens.data()), sizeof(int32_t) * static_cast(numEdges)); + file.read(reinterpret_cast(mChildNodes.data()), sizeof(int32_t) * static_cast(numEdges)); + file.read(reinterpret_cast(mIsEnd.data()), sizeof(uint8_t) * static_cast(numNodes)); + TLLM_CHECK_WITH_INFO(file.good(), "Truncated or unreadable trie file: %s", path.c_str()); + + TLLM_LOG_INFO("GuidedDecoder loaded CSR trie: %u nodes, %u edges, eos=%d from %s", numNodes, numEdges, + mEosTokenId, path.c_str()); + } + + int32_t getRoot() const + { + return mNumNodes > 0 ? 0 : -1; + } + + int32_t numNodes() const + { + return mNumNodes; + } + + int32_t eosTokenId() const + { + return mEosTokenId; + } + + bool isEnd(int32_t node) const + { + return mIsEnd[node] != 0; + } + + // Child node index for the given token at node, or -1 if not present. + int32_t findChild(int32_t node, int32_t token) const + { + uint32_t const b = mChildOffsets[node]; + uint32_t const e = mChildOffsets[node + 1]; + auto const first = mChildTokens.begin() + b; + auto const last = mChildTokens.begin() + e; + auto const it = std::lower_bound(first, last, token); + if (it != last && *it == token) + { + return mChildNodes[b + static_cast(it - first)]; + } + return -1; + } + + uint32_t childBegin(int32_t node) const + { + return mChildOffsets[node]; + } + + uint32_t childEnd(int32_t node) const + { + return mChildOffsets[node + 1]; + } + + int32_t childToken(uint32_t edge) const + { + return mChildTokens[edge]; + } + + SizeType32 getVocabSize() const + { + return mVocabSize; + } + +private: + SizeType32 mVocabSize; + int32_t mNumNodes{0}; + int32_t mEosTokenId{-1}; + std::vector mChildOffsets; // size numNodes + 1 + std::vector mChildTokens; // size numEdges, token-sorted within each node + std::vector mChildNodes; // size numEdges, aligned with mChildTokens + std::vector mIsEnd; // size numNodes +}; + GuidedDecoder::GuidedDecoder(executor::GuidedDecodingConfig const& guidedDecodingConfig, SizeType32 maxNumSequences, - SizeType32 vocabSizePadded, nvinfer1::DataType logitsDtype, BufferManager const& runtimeBufferManager) + SizeType32 maxBeamWidth, SizeType32 vocabSizePadded, nvinfer1::DataType logitsDtype, + BufferManager const& runtimeBufferManager) : mGuidedDecodingBackend{guidedDecodingConfig.getBackend()} , mMaxNumSequences{maxNumSequences} + , mMaxBeamWidth{maxBeamWidth} , mVocabSizePadded{vocabSizePadded} , mBitmaskSize{common::ceilDiv(mVocabSizePadded, 32)} , mLogitsDtype{logitsDtype} @@ -40,6 +162,7 @@ GuidedDecoder::GuidedDecoder(executor::GuidedDecodingConfig const& guidedDecodin { TLLM_CHECK_WITH_INFO(mGuidedDecodingBackend != executor::GuidedDecodingConfig::GuidedDecodingBackend::kLLGUIDANCE, "LLGuidance is not supported for guided decoding in C++ runtime."); + if (mGuidedDecodingBackend == executor::GuidedDecodingConfig::GuidedDecodingBackend::kXGRAMMAR) { mXGrammarMatchers.resize(mMaxNumSequences); @@ -61,16 +184,103 @@ GuidedDecoder::GuidedDecoder(executor::GuidedDecodingConfig const& guidedDecodin /*cache_enabled=*/true, /*cache_limit_bytes=*/static_cast(cacheLimitGb.value_or(1.0f) * 1024 * 1024 * 1024)); - auto const logitsPtrDtype = BufferDataType{mLogitsDtype, false, true}; - auto constexpr bitmaskDtype = TRTDataType::value; - auto constexpr bitmaskPtrDtype = TRTDataType::value; + initTrieBuffers(runtimeBufferManager); + } + else if (mGuidedDecodingBackend == executor::GuidedDecodingConfig::GuidedDecodingBackend::kTRIE) + { + // Initialize trie-based constraint handling + auto const& triePath = guidedDecodingConfig.getTriePath(); + TLLM_CHECK_WITH_INFO(triePath.has_value(), "Trie path must be set for TRIE backend"); + + mTrie = std::make_shared(mVocabSizePadded); + mTrie->loadFromBinary(triePath.value()); + + TLLM_CHECK_WITH_INFO(mTrie->numNodes() > 0, "Trie is empty after loading"); + TLLM_LOG_INFO("Trie loaded successfully, vocab size: %d, bitmask size: %d", + mVocabSizePadded, mBitmaskSize); + + // Off-trie beams may only emit EOS (fail closed without producing all -inf logits). + mEosOnlyBitmask.assign(mBitmaskSize, 0); + if (auto const eos = mTrie->eosTokenId(); eos >= 0 && eos < mVocabSizePadded) + { + mEosOnlyBitmask[eos / 32] |= (1U << (eos % 32)); + } + + // Pre-populate root bitmask cache + auto const& rootBitmask = getTrieBitmask(mTrie->getRoot()); + TLLM_LOG_INFO("Root node has %zu allowed tokens", + std::count_if(rootBitmask.begin(), rootBitmask.end(), [](BitmaskT w) { return w != 0; })); + + initTrieBuffers(runtimeBufferManager); - mLogitsBitmask = runtimeBufferManager.gpu(ITensor::makeShape({mMaxNumSequences, mBitmaskSize}), bitmaskDtype); - mLogitsBitmaskHost = BufferManager::pinned(ITensor::makeShape({mMaxNumSequences, mBitmaskSize}), bitmaskDtype); - mLogitsBitmaskPtrVec = runtimeBufferManager.gpu(ITensor::makeShape({mMaxNumSequences}), bitmaskPtrDtype); - mLogitsBitmaskPtrVecHost = BufferManager::pinned(ITensor::makeShape({mMaxNumSequences}), bitmaskPtrDtype); - mLogitsPtrVec = runtimeBufferManager.gpu(ITensor::makeShape({mMaxNumSequences}), logitsPtrDtype); - mLogitsPtrVecHost = BufferManager::pinned(ITensor::makeShape({mMaxNumSequences}), logitsPtrDtype); + TLLM_LOG_INFO("GuidedDecoder initialized with TRIE backend (per-beam, maxBeamWidth=%d)", mMaxBeamWidth); + } +} + +void GuidedDecoder::initTrieBuffers(BufferManager const& runtimeBufferManager) +{ + auto const logitsPtrDtype = BufferDataType{mLogitsDtype, false, true}; + auto constexpr bitmaskDtype = TRTDataType::value; + auto constexpr bitmaskPtrDtype = TRTDataType::value; + + auto const numRows = mMaxNumSequences * mMaxBeamWidth; + mLogitsBitmask = runtimeBufferManager.gpu(ITensor::makeShape({numRows, mBitmaskSize}), bitmaskDtype); + mLogitsBitmaskHost = BufferManager::pinned(ITensor::makeShape({numRows, mBitmaskSize}), bitmaskDtype); + mLogitsBitmaskPtrVec = runtimeBufferManager.gpu(ITensor::makeShape({numRows}), bitmaskPtrDtype); + mLogitsBitmaskPtrVecHost = BufferManager::pinned(ITensor::makeShape({numRows}), bitmaskPtrDtype); + mLogitsPtrVec = runtimeBufferManager.gpu(ITensor::makeShape({numRows}), logitsPtrDtype); + mLogitsPtrVecHost = BufferManager::pinned(ITensor::makeShape({numRows}), logitsPtrDtype); +} + +std::vector const& GuidedDecoder::getTrieBitmask(int32_t node) +{ + std::lock_guard lock(mTrieCacheMutex); + + // Off-trie: allow only EOS so the beam terminates instead of sampling from all -inf logits. + if (node < 0) + { + return mEosOnlyBitmask; + } + + auto it = mTrieBitmaskCache.find(node); + if (it != mTrieBitmaskCache.end()) + { + return it->second; + } + + std::vector bitmask(mBitmaskSize, 0); + createTrieBitmask(node, bitmask); + + auto [insertIt, _] = mTrieBitmaskCache.emplace(node, std::move(bitmask)); + return insertIt->second; +} + +void GuidedDecoder::createTrieBitmask(int32_t node, std::vector& bitmask) +{ + TLLM_CHECK_WITH_INFO(node >= 0, "Cannot create bitmask from invalid node"); + + auto const setBit = [&](int32_t tokenId) + { + if (tokenId >= 0 && tokenId < mVocabSizePadded) + { + SizeType32 const wordIdx = tokenId / 32; + SizeType32 const bitIdx = tokenId % 32; + if (wordIdx < static_cast(bitmask.size())) + { + bitmask[wordIdx] |= (1U << bitIdx); + } + } + }; + + for (uint32_t edge = mTrie->childBegin(node); edge < mTrie->childEnd(node); ++edge) + { + setBit(mTrie->childToken(edge)); + } + // Terminal node: allow the trie's EOS token so a completed valid sequence can end + // (avoids materializing explicit "...,EOS" leaf nodes). + if (mTrie->isEnd(node) && mTrie->eosTokenId() >= 0) + { + setBit(mTrie->eosTokenId()); } } @@ -90,7 +300,6 @@ void GuidedDecoder::build(ScheduledRequests const& scheduledRequests) auto const seqSlot = llmReq->mSeqSlot.value(); if (llmReq->isContextInitState() && llmReq->isFirstContextChunk()) { - // The request is in the first context forward step (considering kv cache reuse). auto const& guideType = guidedDecodingParams->getGuideType(); auto const& guide = guidedDecodingParams->getGuide(); switch (guideType) @@ -133,8 +342,6 @@ void GuidedDecoder::build(ScheduledRequests const& scheduledRequests) } else if (llmReq->isGenerationInProgressState()) { - // The request is in a generation forward step. - // Currently, guided decoding does not support with beam search. mXGrammarMatchers.at(seqSlot)->AcceptToken(llmReq->getLastTokens(0)); } else @@ -142,9 +349,9 @@ void GuidedDecoder::build(ScheduledRequests const& scheduledRequests) continue; } - // Fill the bitmask on host and asynchorously copy to device using mCopyBufferManager. - auto const logitsBitmask = ITensor::at(mLogitsBitmask, {seqSlot}); - auto const logitsBitmaskHost = ITensor::at(mLogitsBitmaskHost, {seqSlot}); + // xgrammar keeps its per-request design: one state, beam 0's bitmask row. + auto const logitsBitmask = ITensor::at(mLogitsBitmask, {seqSlot * mMaxBeamWidth}); + auto const logitsBitmaskHost = ITensor::at(mLogitsBitmaskHost, {seqSlot * mMaxBeamWidth}); std::array bitmaskShape{mBitmaskSize}; DLTensor logitsBitmaskDlt{logitsBitmaskHost->data(), DLDevice{kDLCPU, 0}, 1, DLDataType{kDLInt, 32, 1}, @@ -154,53 +361,135 @@ void GuidedDecoder::build(ScheduledRequests const& scheduledRequests) } } } + else if (mGuidedDecodingBackend == executor::GuidedDecodingConfig::GuidedDecodingBackend::kTRIE) + { + // Per-beam TRIE constraint. For each scheduled request, re-walk every beam's generated + // tokens from the trie root (bounded by trie depth) and stage one bitmask row per beam. + // At beam width > 1 the caller must have refreshed the request's beam token histories + // via gatherTree (see needsGatheredBeamTokens); the raw slot buffers are stale after + // beam-parent switches. + TLLM_LOG_DEBUG("GuidedDecoder::build TRIE - context=%zu, generation=%zu requests", + scheduledRequests.contextRequests.size(), scheduledRequests.generationRequests.size()); + auto const eosTokenId = mTrie->eosTokenId(); + for (auto const& requests : {scheduledRequests.contextRequests, scheduledRequests.generationRequests}) + { + for (auto const& llmReq : requests) + { + // For TRIE backend, all requests use the trie constraint + bool const isContextInit = llmReq->isContextInitState() && llmReq->isFirstContextChunk(); + if (!isContextInit && !llmReq->isGenerationInProgressState()) + { + continue; + } + + auto const seqSlot = llmReq->mSeqSlot.value(); + auto const reqBeamWidth = llmReq->mSamplingConfig.beamWidth; + TLLM_CHECK_WITH_INFO(reqBeamWidth <= mMaxBeamWidth, + "Request beam width %d exceeds GuidedDecoder maxBeamWidth %d", reqBeamWidth, mMaxBeamWidth); + + for (SizeType32 beam = 0; beam < reqBeamWidth; ++beam) + { + int32_t node = mTrie->getRoot(); + if (!isContextInit) + { + auto const& tokens = llmReq->getTokens(beam); + for (size_t pos = llmReq->mPromptLen; pos < tokens.size() && node >= 0; ++pos) + { + auto const token = static_cast(tokens[pos]); + if (token == eosTokenId) + { + node = -1; // finished beam: keep it terminated (EOS-only mask) + break; + } + node = mTrie->findChild(node, token); + } + } + auto const& bitmask = getTrieBitmask(node); + + auto const row = seqSlot * mMaxBeamWidth + beam; + auto const logitsBitmaskHost = ITensor::at(mLogitsBitmaskHost, {row}); + std::memcpy(logitsBitmaskHost->data(), bitmask.data(), bitmask.size() * sizeof(BitmaskT)); + } + + // Async copy this request's beam rows to device in one contiguous slab. + auto const firstRow = seqSlot * mMaxBeamWidth; + auto const hostSlab = ITensor::slice(mLogitsBitmaskHost, firstRow, reqBeamWidth); + auto const deviceSlab = ITensor::slice(mLogitsBitmask, firstRow, reqBeamWidth); + mCopyBufferManager.copy(*hostSlab, *deviceSlab); + } + } + } } void GuidedDecoder::execute(DecoderInputBuffers const& decoderInputBuffers, BufferManager const& runtimeBufferManager) { auto const& stream = runtimeBufferManager.getStream(); - // Wait for mCopyBufferManager finishing the H2D copy of logitsBitmask - // TODO(enweiz): Move the H2D copy of logitsBitmaskPtrVec to buildGuidedDecoding. - // This may not bring too much perf gain because of the small size of logitsBitmaskPtrVec. - // TODO(enweiz): For chunked context, we currently build mask cache at the first context chunk, and apply - // the mask at the last context chunk. So, ideally we should sync the stream at the last context chunk. CudaEvent event{}; mCopyBufferManager.getStream().record(event); stream.wait(event); - if (mGuidedDecodingBackend == executor::GuidedDecodingConfig::GuidedDecodingBackend::kXGRAMMAR - && !decoderInputBuffers.decoderRequests.empty()) + bool shouldApplyMask = (mGuidedDecodingBackend == executor::GuidedDecodingConfig::GuidedDecodingBackend::kXGRAMMAR + || mGuidedDecodingBackend == executor::GuidedDecodingConfig::GuidedDecodingBackend::kTRIE) + && !decoderInputBuffers.decoderRequests.empty(); + + if (shouldApplyMask) { SizeType32 batchIdx{0}; for (size_t requestIdx = 0; requestIdx < decoderInputBuffers.decoderRequests.size(); ++requestIdx) { auto const& llmReq = decoderInputBuffers.decoderRequests.at(requestIdx); - auto const& guidedDecodingParams = llmReq->getGuidedDecodingParams(); - if (guidedDecodingParams.has_value()) + // For TRIE backend, process all requests; for XGRAMMAR, only those with guided params + bool processRequest = (mGuidedDecodingBackend + == executor::GuidedDecodingConfig::GuidedDecodingBackend::kTRIE); + if (!processRequest) { - auto const seqSlot = llmReq->mSeqSlot.value(); + auto const& guidedDecodingParams = llmReq->getGuidedDecodingParams(); + processRequest = guidedDecodingParams.has_value(); + } + if (processRequest) + { + auto const seqSlot = llmReq->mSeqSlot.value(); auto const& logits = decoderInputBuffers.decoderLogits.at(requestIdx); - auto const logitsBitmask = ITensor::at(mLogitsBitmask, {seqSlot}); - // Use void* to unify the code for different mLogitsDtype - *reinterpret_cast(ITensor::at(mLogitsPtrVecHost, {batchIdx})->data()) = logits->data(); - *reinterpret_cast(ITensor::at(mLogitsBitmaskPtrVecHost, {batchIdx})->data()) - = logitsBitmask->data(); + bool const isTrie + = mGuidedDecodingBackend == executor::GuidedDecodingConfig::GuidedDecodingBackend::kTRIE; + // The per-request logits tensor holds one vocab row per beam. TRIE masks every + // beam's row with its own bitmask; xgrammar keeps its per-request design (beam 0). + auto const logitsRows + = static_cast(ITensor::volume(logits->getShape()) / mVocabSizePadded); + auto const numRows = isTrie ? std::min(llmReq->mSamplingConfig.beamWidth, logitsRows) : 1; + auto const rowBytes = static_cast(mVocabSizePadded) + * (mLogitsDtype == nvinfer1::DataType::kFLOAT ? sizeof(float) : sizeof(half)); - ++batchIdx; + for (SizeType32 beam = 0; beam < numRows; ++beam) + { + auto const logitsBitmask = ITensor::at(mLogitsBitmask, {seqSlot * mMaxBeamWidth + beam}); + *reinterpret_cast(ITensor::at(mLogitsPtrVecHost, {batchIdx})->data()) + = static_cast(logits->data()) + static_cast(beam) * rowBytes; + *reinterpret_cast(ITensor::at(mLogitsBitmaskPtrVecHost, {batchIdx})->data()) + = logitsBitmask->data(); + ++batchIdx; + } } } - if (batchIdx > 0) + + if (batchIdx > 0) { + TLLM_LOG_DEBUG("GuidedDecoder::execute applying TRIE mask to %d batches, vocabSize=%d, bitmaskSize=%d", + batchIdx, mVocabSizePadded, mBitmaskSize); runtimeBufferManager.copy( *ITensor::slice(mLogitsPtrVecHost, 0, batchIdx), *ITensor::slice(mLogitsPtrVec, 0, batchIdx)); runtimeBufferManager.copy(*ITensor::slice(mLogitsBitmaskPtrVecHost, 0, batchIdx), *ITensor::slice(mLogitsBitmaskPtrVec, 0, batchIdx)); + // No stream sync needed here: the ptr-vec H2D copies above, the bitmask kernel + // below, and the downstream decoder all run on the runtime stream, so stream + // ordering already guarantees the masks land before the logits are consumed. auto logitsBitmaskPtrVec = bufferCast(*mLogitsBitmaskPtrVec); + TLLM_LOG_DEBUG("GuidedDecoder::execute logitsDtype=%d (FLOAT=0, HALF=1)", static_cast(mLogitsDtype)); if (mLogitsDtype == nvinfer1::DataType::kFLOAT) { auto logitsPtrVec = bufferCast(*mLogitsPtrVec); diff --git a/cpp/tensorrt_llm/batch_manager/trtGptModelInflightBatching.cpp b/cpp/tensorrt_llm/batch_manager/trtGptModelInflightBatching.cpp index 7a0d78beb8a0..366775822061 100644 --- a/cpp/tensorrt_llm/batch_manager/trtGptModelInflightBatching.cpp +++ b/cpp/tensorrt_llm/batch_manager/trtGptModelInflightBatching.cpp @@ -247,7 +247,7 @@ TrtGptModelInflightBatching::TrtGptModelInflightBatching(std::shared_ptr(executorConfig.getGuidedDecodingConfig().value(), - getMaxNumSequences(), mModelConfig.getVocabSizePadded(mWorldConfig.getSize()), + getMaxNumSequences(), getMaxBeamWidth(), mModelConfig.getVocabSizePadded(mWorldConfig.getSize()), mModelConfig.getLogitsDtype(), mRuntime->getBufferManager()); } @@ -1137,6 +1137,24 @@ void TrtGptModelInflightBatching::forwardAsync(RequestList const& activeRequests executeBatch(currRequests); if (mWorldConfig.isLastPipelineParallelRank() && mGuidedDecoder) { + if (mGuidedDecoder->needsGatheredBeamTokens()) + { + // Per-beam constraints must see true (parent-gathered) beam histories; the raw + // slot buffers go stale when beam search switches hypothesis parents. + for (auto const& llmReq : currRequests.generationRequests) + { + auto const reqBeamWidth = llmReq->mSamplingConfig.beamWidth; + // Skip until the decoder has produced at least one token: on the first + // generation iteration the decoder state has no sequence lengths yet + // (and there is no history that could be stale). + if (reqBeamWidth > 1 && llmReq->isGenerationInProgressState() + && llmReq->getMaxBeamNumTokens() > llmReq->mPromptLen) + { + postProcessRequest( + *llmReq, std::vector(reqBeamWidth, 0), /*streamingGather=*/true); + } + } + } // XGrammar: build maskcache for context requests and perform maskgen for all requests // These need to be overlapped with the kernel execution of forward step mGuidedDecoder->build(currRequests); @@ -1938,7 +1956,7 @@ void TrtGptModelInflightBatching::setupDecoderStep( } void TrtGptModelInflightBatching::postProcessRequest( - LlmRequest& llmReq, std::vector const& numDroppedTokens) + LlmRequest& llmReq, std::vector const& numDroppedTokens, bool streamingGather) { TLLM_LOG_TRACE("%s start", __PRETTY_FUNCTION__); auto const seqSlot = llmReq.mSeqSlot.value(); @@ -1965,7 +1983,8 @@ void TrtGptModelInflightBatching::postProcessRequest( } // Update mDecoderBuffers->slotOutputIdsHost and synchronize - getDecoderSlotHostOutputs(seqSlot, llmReq.returnLogProbs(), llmReq.mSamplingConfig, llmReq.isStreaming()); + getDecoderSlotHostOutputs( + seqSlot, llmReq.returnLogProbs(), llmReq.mSamplingConfig, llmReq.isStreaming() || streamingGather); auto const* outputIdsHostData = bufferCast(*mSlotDecoderBuffers[seqSlot]->outputIdsHost); auto const* sequenceLengthsHostData = bufferCast(*mSlotDecoderBuffers[seqSlot]->sequenceLengthsHost); diff --git a/cpp/tensorrt_llm/batch_manager/trtGptModelInflightBatching.h b/cpp/tensorrt_llm/batch_manager/trtGptModelInflightBatching.h index d6550281a758..21e02c957261 100644 --- a/cpp/tensorrt_llm/batch_manager/trtGptModelInflightBatching.h +++ b/cpp/tensorrt_llm/batch_manager/trtGptModelInflightBatching.h @@ -347,7 +347,11 @@ class TrtGptModelInflightBatching : public TrtGptModel /// @brief It gathers the logits if they need to be returned, calls getDecoderSlotHostOutputs, /// and overwrites the llmRequest tokens buffer. /// Called either on request finishing, or at every step when doing beam search and streaming. - void postProcessRequest(LlmRequest& llmReq, std::vector const& numDroppedTokens); + /// @param streamingGather When true, gather beam token histories with the non-destructive + /// per-step (streaming-style) finalize even for non-streaming requests. Used to refresh + /// per-beam histories for beam-aware guided decoding before each decode step. + void postProcessRequest( + LlmRequest& llmReq, std::vector const& numDroppedTokens, bool streamingGather = false); /// @brief Reorders generation logits to match finalized beam paths after gatherTree. /// During beam search, logits are stored by beam slot. After finalization, output_ids are /// reordered by parentIds, but logits are not. This method traces parentIds on the host diff --git a/cpp/tensorrt_llm/executor/guidedDecodingConfig.cpp b/cpp/tensorrt_llm/executor/guidedDecodingConfig.cpp index d3f5131bafaa..caedd6c4e0c9 100644 --- a/cpp/tensorrt_llm/executor/guidedDecodingConfig.cpp +++ b/cpp/tensorrt_llm/executor/guidedDecodingConfig.cpp @@ -25,18 +25,19 @@ namespace tensorrt_llm::executor GuidedDecodingConfig::GuidedDecodingConfig(GuidedDecodingConfig::GuidedDecodingBackend backend, std::optional> encodedVocab, std::optional tokenizerStr, - std::optional> stopTokenIds) + std::optional> stopTokenIds, std::optional triePath) : mBackend(backend) , mEncodedVocab(std::move(encodedVocab)) , mTokenizerStr(std::move(tokenizerStr)) , mStopTokenIds(std::move(stopTokenIds)) + , mTriePath(std::move(triePath)) { } bool GuidedDecodingConfig::operator==(GuidedDecodingConfig const& other) const { return mBackend == other.mBackend && mEncodedVocab == other.mEncodedVocab && mTokenizerStr == other.mTokenizerStr - && mStopTokenIds == other.mStopTokenIds; + && mStopTokenIds == other.mStopTokenIds && mTriePath == other.mTriePath; } void GuidedDecodingConfig::setBackend(GuidedDecodingConfig::GuidedDecodingBackend const& backend) @@ -79,6 +80,16 @@ std::optional> GuidedDecodingConfig::getStopTokenIds() return mStopTokenIds; } +void GuidedDecodingConfig::setTriePath(std::string const& triePath) +{ + mTriePath = triePath; +} + +std::optional GuidedDecodingConfig::getTriePath() const +{ + return mTriePath; +} + void GuidedDecodingConfig::validate() const { if (mBackend == GuidedDecodingBackend::kXGRAMMAR) @@ -91,6 +102,10 @@ void GuidedDecodingConfig::validate() const "between requests and xgrammar may cause xgrammar execution error."); } } + else if (mBackend == GuidedDecodingBackend::kTRIE) + { + TLLM_CHECK_WITH_INFO(mTriePath, "Guided decoding is enabled with trie, but TriePath is not set"); + } } } // namespace tensorrt_llm::executor diff --git a/cpp/tensorrt_llm/nanobind/executor/executorConfig.cpp b/cpp/tensorrt_llm/nanobind/executor/executorConfig.cpp index 830f30ab9c67..16bab8a64d60 100644 --- a/cpp/tensorrt_llm/nanobind/executor/executorConfig.cpp +++ b/cpp/tensorrt_llm/nanobind/executor/executorConfig.cpp @@ -400,28 +400,30 @@ void initConfigBindings(nb::module_& m) nb::enum_(pyGuidedDecodingConfig, "GuidedDecodingBackend") .value("XGRAMMAR", tle::GuidedDecodingConfig::GuidedDecodingBackend::kXGRAMMAR) - .value("LLGUIDANCE", tle::GuidedDecodingConfig::GuidedDecodingBackend::kLLGUIDANCE); + .value("LLGUIDANCE", tle::GuidedDecodingConfig::GuidedDecodingBackend::kLLGUIDANCE) + .value("TRIE", tle::GuidedDecodingConfig::GuidedDecodingBackend::kTRIE); auto guidedDecodingConfigGetstate = [](tle::GuidedDecodingConfig const& self) { return nb::make_tuple( - self.getBackend(), self.getEncodedVocab(), self.getTokenizerStr(), self.getStopTokenIds()); + self.getBackend(), self.getEncodedVocab(), self.getTokenizerStr(), self.getStopTokenIds(), self.getTriePath()); }; auto guidedDecodingConfigSetstate = [](tle::GuidedDecodingConfig& self, nb::tuple state) { - if (state.size() != 4) + if (state.size() != 5) { throw std::runtime_error("Invalid GuidedDecodingConfig state!"); } new (&self) tle::GuidedDecodingConfig(nb::cast(state[0]), nb::cast>>(state[1]), nb::cast>(state[2]), - nb::cast>>(state[3])); + nb::cast>>(state[3]), + nb::cast>(state[4])); }; pyGuidedDecodingConfig .def(nb::init>, - std::optional, std::optional>>(), + std::optional, std::optional>, std::optional>(), nb::arg("backend"), nb::arg("encoded_vocab") = nb::none(), nb::arg("tokenizer_str") = nb::none(), - nb::arg("stop_token_ids") = nb::none()) + nb::arg("stop_token_ids") = nb::none(), nb::arg("trie_path") = nb::none()) .def_prop_rw("backend", &tle::GuidedDecodingConfig::getBackend, &tle::GuidedDecodingConfig::setBackend) .def_prop_rw( "encoded_vocab", &tle::GuidedDecodingConfig::getEncodedVocab, &tle::GuidedDecodingConfig::setEncodedVocab) @@ -429,6 +431,8 @@ void initConfigBindings(nb::module_& m) "tokenizer_str", &tle::GuidedDecodingConfig::getTokenizerStr, &tle::GuidedDecodingConfig::setTokenizerStr) .def_prop_rw( "stop_token_ids", &tle::GuidedDecodingConfig::getStopTokenIds, &tle::GuidedDecodingConfig::setStopTokenIds) + .def_prop_rw( + "trie_path", &tle::GuidedDecodingConfig::getTriePath, &tle::GuidedDecodingConfig::setTriePath) .def("__getstate__", guidedDecodingConfigGetstate) .def("__setstate__", guidedDecodingConfigSetstate); diff --git a/trie_decoding/README.md b/trie_decoding/README.md new file mode 100644 index 000000000000..03cf7bf36408 --- /dev/null +++ b/trie_decoding/README.md @@ -0,0 +1,228 @@ +# Native C++ Trie Decoding for TensorRT-LLM + +This fork adds native C++ trie-based constrained decoding to TensorRT-LLM, enabling efficient generative recommendation systems with beam search. + +## Overview + +Generative recommendation systems use LLMs with constrained decoding to generate valid item IDs from a predefined vocabulary (stored as a trie). Traditional approaches using Python callbacks incur ~40ms overhead per generation step, severely limiting throughput. + +This implementation integrates trie constraint masking directly into TensorRT-LLM's C++ inference pipeline, eliminating Python callback overhead. + +### Key Features + +- **Native C++ trie loading and traversal** - Binary trie format loaded directly in C++ +- **Integrated with GuidedDecoder** - Uses TRT-LLM's existing guided decoding infrastructure +- **O(1) token constraint checking** - Pre-computed bitmasks cached per trie node +- **No Python callback overhead** - All masking happens in C++ +- **Compatible with beam search** - Works with any beam width + +## Architecture + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ TensorRT-LLM │ +│ ┌─────────────────┐ ┌────────────────────────────────────┐ │ +│ │ Executor │───▶│ GuidedDecoder (kTRIE backend) │ │ +│ │ │ │ ┌─────────────────────────────────┐│ │ +│ │ GuidedDecoding │ │ │ GuidedTrie ││ │ +│ │ Config (TRIE) │ │ │ - loadFromBinary() ││ │ +│ │ │ │ │ - getRoot() ││ │ +│ │ trie_path: ... │ │ │ - GuidedTrieNode traversal ││ │ +│ └─────────────────┘ │ └─────────────────────────────────┘│ │ +│ │ ┌─────────────────────────────────┐│ │ +│ │ │ Bitmask Cache (LRU) ││ │ +│ │ │ - O(1) mask lookup ││ │ +│ │ │ - Pre-computed per node ││ │ +│ │ └─────────────────────────────────┘│ │ +│ │ │ │ +│ │ invokeLogitsBitmask() ◄────────────┘ │ +│ └────────────────────────────────────────┘ +└─────────────────────────────────────────────────────────────────┘ +``` + +## Files Modified + +### Core C++ Changes + +| File | Description | +|------|-------------| +| `cpp/include/tensorrt_llm/executor/executor.h` | Added `kTRIE` to `GuidedDecodingBackend` enum, added `triePath` parameter | +| `cpp/tensorrt_llm/executor/guidedDecodingConfig.cpp` | Implemented `triePath` getter/setter and validation | +| `cpp/include/tensorrt_llm/batch_manager/guidedDecoder.h` | Added `GuidedTrie`, `GuidedTrieNode` forward declarations and member variables | +| `cpp/tensorrt_llm/batch_manager/guidedDecoder.cpp` | Implemented `GuidedTrie` class with binary loading, bitmask caching, and constraint application | +| `cpp/tensorrt_llm/nanobind/executor/executorConfig.cpp` | Exposed `TRIE` backend and `trie_path` to Python via nanobind | +| `cpp/tensorrt_llm/pybind/executor/executorConfig.cpp` | Exposed `TRIE` backend and `trie_path` to Python via pybind11 | + +### Triton Backend Changes + +| File | Description | +|------|-------------| +| `triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/model.py` | Added native TRIE backend configuration | +| `triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/config.pbtxt` | Added `trie_path` and `enable_trie_mask` parameters | + +## Binary Trie Format + +The C++ implementation expects a specific binary format: + +``` +Header: + - Magic number: uint32 (0x54524945 = "TRIE") + - Node count: uint32 + +For each node (indexed 0 to N-1): + - Number of children: uint32 + - Is terminal: uint8 + - For each child: + - Token ID: uint32 + - Child node index: uint32 +``` + +Use `trie_decoding/convert_trie_to_binary.py` to convert pickle tries to this format. + +## Building from Source + +### Prerequisites + +- NVIDIA GPU with compute capability 8.0+ (A100, H100, etc.) +- Docker with NVIDIA container runtime +- ~100GB disk space for build + +### Build Steps + +```bash +# Clone this fork +git clone tensorrt-llm-trie +cd tensorrt-llm-trie + +# Start the development container +make -C docker devel_build +make -C docker devel_run + +# Inside the container, build the wheel +cd /code/tensorrt_llm +python3 ./scripts/build_wheel.py --cuda_architectures "80-real" # For A100 +# Use "89-real" for H100, "90-real" for Hopper + +# The wheel will be at: build/tensorrt_llm-*.whl +``` + +### Build Custom Triton Image + +```bash +# Copy wheel to docker directory +cp build/tensorrt_llm-*.whl trie_decoding/docker/ + +# Build custom Triton image +cd trie_decoding/docker +docker build -t tritonserver-trtllm-trie:latest . +``` + +## Usage + +### 1. Create a Trie + +```python +from trie import Trie + +trie = Trie() +# Add your product/item IDs +for item_id in item_ids: + tokens = tokenizer.encode(item_id) + trie.insert(tokens) + +# Save as pickle (for Python fallback) and binary (for C++) +trie.save("trie.pkl") +``` + +### 2. Convert to Binary Format + +```bash +python trie_decoding/convert_trie_to_binary.py trie.pkl trie.bin --verify +``` + +### 3. Configure Triton + +In your model's `config.pbtxt`: + +```protobuf +parameters: { + key: "guided_decoding_backend" + value: { string_value: "trie" } +} +parameters: { + key: "trie_binary_path" + value: { string_value: "/path/to/trie.bin" } +} +``` + +### 4. Run Inference + +```python +import requests + +response = requests.post( + "http://localhost:8000/v2/models/ensemble/generate", + json={ + "text_input": "recommend products for", + "max_tokens": 10, + "beam_width": 256, + "temperature": 0.0 + } +) +``` + +## Performance + +Tested on NVIDIA A100 40GB with Qwen3 0.6B model: + +| Configuration | Throughput (RPS) | p50 Latency | p95 Latency | +|---------------|------------------|-------------|-------------| +| No masking | ~200 | 45ms | 80ms | +| Python trie mask (callback) | ~50 | 180ms | 250ms | +| **Native C++ trie (this fork)** | ~180 | 50ms | 90ms | + +The native C++ implementation achieves **~3.6x higher throughput** compared to Python callbacks while maintaining low latency. + +## Load Testing + +```bash +# Run load test +python trie_decoding/loadtest.py \ + --host localhost \ + --port 8000 \ + --concurrency 30 \ + --requests 200 \ + --beam-width 256 \ + --output-len 4 +``` + +## Troubleshooting + +### "Invalid trie file format" error + +Ensure your binary trie starts with magic number `0x54524945`. Use `convert_trie_to_binary.py --verify` to validate. + +### Segmentation fault during loading + +Check that: +1. All child indices are valid (< total node count) +2. The file isn't truncated +3. Node count matches actual data + +### High latency despite native implementation + +Verify the C++ backend is active by checking logs for: +``` +[TensorRT-LLM][INFO] GuidedDecoder initialized with TRIE backend +``` + +If you see "Using Python trie mask processors", the fallback is active. + +## License + +This fork maintains the same Apache 2.0 license as TensorRT-LLM. + +## References + +- [xGR: Efficient Generative Recommendation Serving at Scale](https://arxiv.org/pdf/2512.11529) +- [TensorRT-LLM Documentation](https://nvidia.github.io/TensorRT-LLM/) diff --git a/trie_decoding/convert_trie_to_binary.py b/trie_decoding/convert_trie_to_binary.py new file mode 100644 index 000000000000..d312551c7853 --- /dev/null +++ b/trie_decoding/convert_trie_to_binary.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +"""Convert a pickle trie to the C++ CSR binary format (v2). + +v2 layout (matches C++ GuidedTrie::loadFromBinary): + magic(u32=0x54524945 'TRIE'), version(u32=2), numNodes(u32), numEdges(u32), eos(i32), + childOffsets(u32[numNodes+1]), childTokens(i32[numEdges], token-sorted per node), + childNodes(i32[numEdges]), isEnd(u8[numNodes]). + +Terminal (is_terminal) nodes carry EOS via the header eos token, so sequences should +NOT append EOS themselves. For the large gen_search trie use build_gensearch_trie.py +(vectorized, builds CSR directly from keys.u64); this dict-based path is for small tries. +""" + +import argparse +import pickle +import struct +import sys +from collections import deque +from pathlib import Path + +import numpy as np + +sys.path.insert(0, str(Path(__file__).parent)) + + +def collect_nodes_bfs(root): + """Collect all nodes using BFS and assign indices.""" + nodes = [] + node_to_idx = {} + queue = deque([root]) + + while queue: + node = queue.popleft() + if node in node_to_idx: + continue + idx = len(nodes) + node_to_idx[node] = idx + nodes.append(node) + for child in node.children.values(): + if child not in node_to_idx: + queue.append(child) + + return nodes, node_to_idx + + +def save_trie_binary(trie, path: str, eos_token_id: int = -1): + """Serialize a Trie to the native CSR binary format (v2).""" + MAGIC = 0x54524945 # "TRIE" + nodes, node_to_idx = collect_nodes_bfs(trie.root) + num_nodes = len(nodes) + + child_offsets = np.zeros(num_nodes + 1, dtype=np.uint32) + child_tokens = [] + child_nodes = [] + is_end = np.zeros(num_nodes, dtype=np.uint8) + for i, node in enumerate(nodes): + is_end[i] = 1 if getattr(node, 'is_terminal', False) else 0 + for token_id, child in sorted(node.children.items()): # token-sorted for binary search + child_tokens.append(token_id) + child_nodes.append(node_to_idx[child]) + child_offsets[i + 1] = len(child_tokens) + num_edges = len(child_tokens) + + with open(path, 'wb') as f: + f.write(struct.pack('IIIIi', MAGIC, 2, num_nodes, num_edges, eos_token_id)) + child_offsets.tofile(f) + np.asarray(child_tokens, dtype=np.int32).tofile(f) + np.asarray(child_nodes, dtype=np.int32).tofile(f) + is_end.tofile(f) + + print(f"Saved CSR trie: {num_nodes} nodes, {num_edges} edges, eos={eos_token_id} to {path}") + + +def verify_binary_trie(path: str) -> bool: + """Verify the CSR binary trie file is valid.""" + with open(path, 'rb') as f: + magic, version, num_nodes, num_edges, eos = struct.unpack('IIIIi', f.read(20)) + if magic != 0x54524945: + print(f"Error: Invalid magic number: {hex(magic)}") + return False + if version != 2: + print(f"Error: Unsupported version {version} (expected 2)") + return False + offsets = np.frombuffer(f.read(4 * (num_nodes + 1)), dtype=np.uint32) + tokens = np.frombuffer(f.read(4 * num_edges), dtype=np.int32) + child_nodes = np.frombuffer(f.read(4 * num_edges), dtype=np.int32) + is_end = np.frombuffer(f.read(num_nodes), dtype=np.uint8) + + if len(offsets) != num_nodes + 1 or offsets[-1] != num_edges: + print("Error: childOffsets inconsistent with numEdges") + return False + if num_edges and int(child_nodes.max()) >= num_nodes: + print("Error: child index out of range") + return False + print(f"Verification passed: {num_nodes} nodes, {num_edges} edges, eos={eos}, " + f"terminal nodes={int(is_end.sum())}") + return True + + +def convert_pickle_to_binary(pickle_path: str, binary_path: str, eos_token_id: int = -1): + """Convert pickle trie to binary format.""" + print(f"Loading pickle trie from {pickle_path}...") + with open(pickle_path, 'rb') as f: + trie = pickle.load(f) + + node_count = getattr(trie, 'node_count', 'unknown') + print(f"Loaded trie with {node_count} nodes") + print(f"Saving to CSR binary format at {binary_path}...") + save_trie_binary(trie, binary_path, eos_token_id) + + +def main(): + parser = argparse.ArgumentParser(description='Convert pickle trie to C++ CSR binary format (v2)') + parser.add_argument('input', help='Input pickle file (.pkl)') + parser.add_argument('output', help='Output binary file (.bin)') + parser.add_argument('--eos', type=int, default=-1, help='EOS token id allowed at terminal nodes') + parser.add_argument('--verify', action='store_true', help='Verify output after conversion') + args = parser.parse_args() + + if not Path(args.input).exists(): + print(f"Error: Input file not found: {args.input}") + sys.exit(1) + + convert_pickle_to_binary(args.input, args.output, args.eos) + + if args.verify: + print("\nVerifying binary file...") + if not verify_binary_trie(args.output): + sys.exit(1) + + print("Done!") + + +if __name__ == '__main__': + main() diff --git a/trie_decoding/docker/Dockerfile b/trie_decoding/docker/Dockerfile new file mode 100644 index 000000000000..692bd7b76ca6 --- /dev/null +++ b/trie_decoding/docker/Dockerfile @@ -0,0 +1,23 @@ +# Custom Triton image with TRT-LLM built with native trie constraint integration +# +# Build instructions: +# 1. Build the TRT-LLM wheel from the repo root: +# python3 ./scripts/build_wheel.py --cuda_architectures "80-real" +# 2. Copy the wheel to this directory: +# cp build/tensorrt_llm-*.whl trie_decoding/docker/ +# 3. Build the image: +# docker build -t tritonserver-trtllm-trie:latest trie_decoding/docker/ + +FROM nvcr.io/nvidia/tritonserver:25.12-trtllm-python-py3 + +# Copy the custom TRT-LLM wheel (use wildcard to be version-agnostic) +COPY tensorrt_llm-*.whl /tmp/ + +# Uninstall the existing tensorrt_llm and install the custom one +RUN pip uninstall -y tensorrt_llm 2>/dev/null || true && \ + pip install /tmp/tensorrt_llm-*.whl && \ + rm /tmp/tensorrt_llm-*.whl + +# Note: Import verification requires GPU, skip during build +# Verification can be done at runtime with: +# python3 -c "import tensorrt_llm; print(tensorrt_llm.__version__)" diff --git a/trie_decoding/generate_trie.py b/trie_decoding/generate_trie.py new file mode 100755 index 000000000000..570f0042d016 --- /dev/null +++ b/trie_decoding/generate_trie.py @@ -0,0 +1,263 @@ +#!/usr/bin/env python3 +""" +Generate a large trie for constrained decoding testing. + +Usage: + python generate_trie.py --output trie.pkl --nodes 120000000 + python generate_trie.py --output trie.pkl --nodes 120000000 --vocab-size 50257 + +For 120M nodes, expect: + - ~10-30 GB RAM during generation + - ~2-8 GB pickle file (depends on structure) + - 10-30 minutes generation time +""" + +import argparse +import pickle +import random +import sys +import time +from typing import List, Optional + +# Import from shared module - this ensures pickle stores the correct module path +from trie import Trie, TrieNode, CompactTrie + + +def generate_random_sequences( + target_nodes: int, + vocab_size: int, + min_length: int = 3, + max_length: int = 50, + seed: int = 42 +) -> Trie: + """Generate random token sequences until we hit target node count.""" + random.seed(seed) + trie = Trie() + + start_time = time.time() + last_report = start_time + + print(f"Generating trie with target {target_nodes:,} nodes...") + print(f"Vocab size: {vocab_size:,}, Sequence length: {min_length}-{max_length}") + + while trie.node_count < target_nodes: + length = random.randint(min_length, max_length) + sequence = [random.randint(0, vocab_size - 1) for _ in range(length)] + trie.insert(sequence) + + now = time.time() + if now - last_report >= 5.0: + elapsed = now - start_time + rate = trie.node_count / elapsed + remaining = (target_nodes - trie.node_count) / rate if rate > 0 else 0 + print(f" Nodes: {trie.node_count:,} / {target_nodes:,} " + f"({100*trie.node_count/target_nodes:.1f}%) | " + f"Sequences: {trie.sequence_count:,} | " + f"Rate: {rate:,.0f} nodes/sec | " + f"ETA: {remaining/60:.1f} min") + last_report = now + + elapsed = time.time() - start_time + print(f"\nGeneration complete!") + print(f" Total nodes: {trie.node_count:,}") + print(f" Total sequences: {trie.sequence_count:,}") + print(f" Time: {elapsed:.1f} seconds") + + return trie + + +def generate_realistic_sequences( + target_nodes: int, + vocab_size: int, + common_prefixes: int = 10000, + seed: int = 42 +) -> Trie: + """ + Generate sequences that simulate realistic autocomplete patterns: + - Many sequences share common prefixes (like real text) + - Zipf-like distribution of token frequencies + - Variable branching factors + """ + random.seed(seed) + trie = Trie() + + start_time = time.time() + last_report = start_time + + print(f"Generating realistic trie with target {target_nodes:,} nodes...") + + # Generate common prefix patterns + print(" Creating common prefix patterns...") + prefixes = [] + for _ in range(common_prefixes): + length = random.randint(2, 8) + # Use zipf-like distribution for tokens (some tokens are more common) + prefix = [int(random.paretovariate(1.5)) % vocab_size for _ in range(length)] + prefixes.append(prefix) + + # Generate sequences extending from prefixes + while trie.node_count < target_nodes: + # 70% chance to extend an existing prefix + if prefixes and random.random() < 0.7: + prefix = random.choice(prefixes) + suffix_len = random.randint(1, 30) + suffix = [int(random.paretovariate(1.5)) % vocab_size for _ in range(suffix_len)] + sequence = prefix + suffix + else: + # Random sequence + length = random.randint(5, 40) + sequence = [int(random.paretovariate(1.5)) % vocab_size for _ in range(length)] + + trie.insert(sequence) + + # Occasionally add the sequence as a new prefix + if random.random() < 0.01: + prefixes.append(sequence[:random.randint(2, min(8, len(sequence)))]) + + now = time.time() + if now - last_report >= 5.0: + elapsed = now - start_time + rate = trie.node_count / elapsed + remaining = (target_nodes - trie.node_count) / rate if rate > 0 else 0 + print(f" Nodes: {trie.node_count:,} / {target_nodes:,} " + f"({100*trie.node_count/target_nodes:.1f}%) | " + f"Sequences: {trie.sequence_count:,} | " + f"Rate: {rate:,.0f} nodes/sec | " + f"ETA: {remaining/60:.1f} min") + last_report = now + + elapsed = time.time() - start_time + print(f"\nGeneration complete!") + print(f" Total nodes: {trie.node_count:,}") + print(f" Total sequences: {trie.sequence_count:,}") + print(f" Unique prefixes: {len(prefixes):,}") + print(f" Time: {elapsed:.1f} seconds") + + return trie + + +def analyze_trie(trie: Trie, sample_size: int = 1000): + """Print statistics about the trie structure.""" + print("\nTrie Analysis:") + print(f" Total nodes: {trie.node_count:,}") + print(f" Total sequences: {trie.sequence_count:,}") + + # Sample branching factors + branching_factors = [] + depths = [] + + def sample_node(node, depth=0): + if len(branching_factors) >= sample_size: + return + branching_factors.append(len(node.children)) + depths.append(depth) + if node.children: + child = random.choice(list(node.children.values())) + sample_node(child, depth + 1) + + for _ in range(min(100, sample_size)): + sample_node(trie.root) + + if branching_factors: + avg_branching = sum(branching_factors) / len(branching_factors) + max_branching = max(branching_factors) + avg_depth = sum(depths) / len(depths) + max_depth = max(depths) + print(f" Avg branching factor: {avg_branching:.2f}") + print(f" Max branching factor: {max_branching}") + print(f" Avg sampled depth: {avg_depth:.1f}") + print(f" Max sampled depth: {max_depth}") + + +def main(): + parser = argparse.ArgumentParser(description="Generate a large trie for testing") + parser.add_argument("--output", "-o", type=str, default="trie.pkl", + help="Output pickle file path") + parser.add_argument("--nodes", "-n", type=int, default=120_000_000, + help="Target number of nodes (default: 120M)") + parser.add_argument("--vocab-size", "-v", type=int, default=50257, + help="Vocabulary size (default: 50257 for GPT-2)") + parser.add_argument("--realistic", "-r", action="store_true", + help="Use realistic distribution (shared prefixes)") + parser.add_argument("--seed", "-s", type=int, default=42, + help="Random seed") + parser.add_argument("--analyze", "-a", action="store_true", + help="Analyze trie after generation") + parser.add_argument("--protocol", "-p", type=int, default=4, + help="Pickle protocol (4 for Python 3.4+, 5 for 3.8+)") + + args = parser.parse_args() + + print("=" * 60) + print("Trie Generator for Constrained Decoding") + print("=" * 60) + + # Check memory + try: + import psutil + mem = psutil.virtual_memory() + print(f"\nSystem memory: {mem.total / 1e9:.1f} GB total, {mem.available / 1e9:.1f} GB available") + + # Rough estimate: ~100-200 bytes per node + estimated_mem = args.nodes * 150 / 1e9 + print(f"Estimated memory needed: ~{estimated_mem:.1f} GB") + + if estimated_mem > mem.available / 1e9 * 0.8: + print("\nWARNING: This may exceed available memory!") + response = input("Continue anyway? [y/N] ") + if response.lower() != 'y': + sys.exit(1) + except ImportError: + print("\n(Install psutil for memory estimates: pip install psutil)") + + print() + + # Generate trie + if args.realistic: + trie = generate_realistic_sequences( + target_nodes=args.nodes, + vocab_size=args.vocab_size, + seed=args.seed + ) + else: + trie = generate_random_sequences( + target_nodes=args.nodes, + vocab_size=args.vocab_size, + seed=args.seed + ) + + if args.analyze: + analyze_trie(trie) + + # Save to pickle + print(f"\nSaving to {args.output}...") + start_time = time.time() + + with open(args.output, 'wb') as f: + pickle.dump(trie, f, protocol=args.protocol) + + elapsed = time.time() - start_time + import os + file_size = os.path.getsize(args.output) + print(f"Saved! File size: {file_size / 1e9:.2f} GB, Time: {elapsed:.1f} seconds") + + # Verify + print("\nVerifying load...") + start_time = time.time() + with open(args.output, 'rb') as f: + loaded = pickle.load(f) + elapsed = time.time() - start_time + print(f"Load time: {elapsed:.1f} seconds") + print(f"Loaded node count: {loaded.node_count:,}") + + # Quick functionality test + test_seq = [random.randint(0, args.vocab_size - 1) for _ in range(3)] + allowed = loaded.allowed_next(test_seq) + print(f"Test allowed_next({test_seq}): {len(allowed)} tokens allowed") + + print("\nDone!") + + +if __name__ == "__main__": + main() + diff --git a/trie_decoding/loadtest.py b/trie_decoding/loadtest.py new file mode 100644 index 000000000000..c7dcd88b6c29 --- /dev/null +++ b/trie_decoding/loadtest.py @@ -0,0 +1,686 @@ +#!/usr/bin/env python3 +import asyncio +import aiohttp +import argparse +import time +import statistics +import json +import random +from dataclasses import dataclass, field +from pathlib import Path +from itertools import product +from tabulate import tabulate + + +@dataclass +class LoadTestConfig: + url: str = "http://localhost:8000/v2/models/ensemble/generate" + concurrency: int = 10 + total_requests: int = 1000 + max_tokens: int = 15 + min_tokens: int = 0 + beam_width: int = 20 + temperature: float = 1.0 + timeout: int = 60 + queries_file: str = "" + num_prior_queries: int = -1 + max_input_tokens: int = 0 + hide_samples: bool = False + backend: str = "triton" + model: str = "" + target_rps: float = 0 + max_concurrency: int = 500 + raw_prompts: bool = False + use_trie_mask: bool = True + logits_post_processor_name: str = "trie_mask" + + +@dataclass +class RequestResult: + latency_ms: float + success: bool + status_code: int = 0 + error: str = "" + input_text: str = "" + output_text: str = "" + output_tokens_total: int = 0 + output_tokens_avg: float = 0 + input_tokens: int = 0 + clicked_query: str = "" + + +@dataclass +class LoadTestResults: + results: list = field(default_factory=list) + start_time: float = 0 + end_time: float = 0 + + @property + def successful(self): + return [r for r in self.results if r.success] + + @property + def failed(self): + return [r for r in self.results if not r.success] + + @property + def latencies(self): + return [r.latency_ms for r in self.successful] + + @property + def output_tokens_avg(self): + return [r.output_tokens_avg for r in self.successful if r.output_tokens_avg > 0] + + @property + def output_tokens_total(self): + return [r.output_tokens_total for r in self.successful if r.output_tokens_total > 0] + + def percentile(self, data, p): + if not data: + return 0 + idx = int(len(data) * p / 100) + return sorted(data)[min(idx, len(data) - 1)] + + def summary(self): + duration = self.end_time - self.start_time + total = len(self.results) + success_count = len(self.successful) + fail_count = len(self.failed) + + print("\n" + "=" * 70) + print("LOAD TEST RESULTS") + print("=" * 70) + print(f"Total Requests: {total}") + print(f"Successful: {success_count} ({100*success_count/total:.1f}%)") + print(f"Failed: {fail_count} ({100*fail_count/total:.1f}%)") + print(f"Duration: {duration:.2f}s") + print(f"Throughput: {total/duration:.2f} req/s") + if self.successful: + print(f"Successful RPS: {success_count/duration:.2f} req/s") + + if self.latencies: + sorted_lat = sorted(self.latencies) + print("\n" + "-" * 50) + print("LATENCY (ms)") + print("-" * 50) + print(f" Min: {min(sorted_lat):>10.1f}") + print(f" Max: {max(sorted_lat):>10.1f}") + print(f" Mean: {statistics.mean(sorted_lat):>10.1f}") + print(f" Median: {statistics.median(sorted_lat):>10.1f}") + if len(sorted_lat) > 1: + print(f" Stddev: {statistics.stdev(sorted_lat):>10.1f}") + print(f" P50: {self.percentile(sorted_lat, 50):>10.1f}") + print(f" P90: {self.percentile(sorted_lat, 90):>10.1f}") + print(f" P95: {self.percentile(sorted_lat, 95):>10.1f}") + print(f" P99: {self.percentile(sorted_lat, 99):>10.1f}") + + if self.output_tokens_avg: + sorted_avg = sorted(self.output_tokens_avg) + total_tokens = sum(self.output_tokens_total) + print("\n" + "-" * 50) + print("OUTPUT TOKENS (avg per beam)") + print("-" * 50) + print(f" Min: {min(sorted_avg):>10.1f}") + print(f" Max: {max(sorted_avg):>10.1f}") + print(f" Mean: {statistics.mean(sorted_avg):>10.1f}") + print(f" Median: {statistics.median(sorted_avg):>10.1f}") + if len(sorted_avg) > 1: + print(f" Stddev: {statistics.stdev(sorted_avg):>10.1f}") + print(f" P50: {self.percentile(sorted_avg, 50):>10.1f}") + print(f" P90: {self.percentile(sorted_avg, 90):>10.1f}") + print(f" P95: {self.percentile(sorted_avg, 95):>10.1f}") + print(f" P99: {self.percentile(sorted_avg, 99):>10.1f}") + print(f"\n Total (all beams): {total_tokens:>10}") + print(f" Tokens/s: {total_tokens/duration:>10.1f}") + + if self.failed: + print("\n" + "-" * 50) + print("ERRORS") + print("-" * 50) + error_counts = {} + for r in self.failed: + error_counts[r.error] = error_counts.get(r.error, 0) + 1 + for err, count in sorted(error_counts.items(), key=lambda x: -x[1]): + print(f" {err}: {count}") + + print("=" * 70 + "\n") + + def print_samples(self): + if self.successful: + print("-" * 50) + print("SAMPLE QUERIES & OUTPUTS") + print("-" * 50) + samples = random.sample(self.successful, min(3, len(self.successful))) + for i, r in enumerate(samples, 1): + print(f"\n[{i}] Input: {r.input_text}") + print(f" Latency: {r.latency_ms:.1f}ms | Tokens: {r.output_tokens_avg:.1f} avg ({r.output_tokens_total} total)") + if r.clicked_query: + print(f" Clicked: {r.clicked_query}") + beams = r.output_text if isinstance(r.output_text, list) else [r.output_text] + for j, beam in enumerate(beams[:25]): + beam_str = str(beam)[:80] + ellipsis = '...' if len(str(beam)) > 80 else '' + print(f" Beam {j+1}: {beam_str}{ellipsis}") + + if len(beams) > 25: + print(f" ... and {len(beams) - 25} more beams") + print() + + +class QueryGenerator: + USER_PRIOR_COL = "contextualInfo[name=user].rivuletUserInfo.timeseries.recentlySearchedQueries50FV1#searchQuery" + BROWSER_PRIOR_COL = "contextualInfo[name=browser].rivuletBrowserInfo.timeseries.recentlySearchedQueries50FV1#searchQuery" + PREFIX_COL = "clientProvidedInfo.query.prefix" + CANDIDATES_COL = "queryCandidates" + + def __init__(self, parquet_file: str, num_prior: int = -1, max_input_tokens: int = 0, raw_prompts: bool = False): + self.num_prior = num_prior + self.max_input_tokens = max_input_tokens + self.raw_prompts = raw_prompts + self._load_parquet(parquet_file) + + def _to_list(self, val): + if val is None: + return [] + try: + if hasattr(val, 'tolist'): + return val.tolist() + if isinstance(val, (list, tuple)): + return list(val) + except: + pass + return [] + + def _extract_clicked(self, candidates) -> str: + if candidates is None: + return "" + try: + for c in self._to_list(candidates): + if isinstance(c, dict) and c.get('attribution') == 'clicked': + return c.get('query', '') + except: + pass + return "" + + def _load_parquet(self, filepath: str): + import pandas as pd + path = Path(filepath) + if not path.exists(): + raise FileNotFoundError(f"Parquet file not found: {filepath}") + df = pd.read_parquet(filepath) + self.prompts = [] + for _, row in df.iterrows(): + prefix = row.get(self.PREFIX_COL) or "" + if not prefix: + continue + prior_queries = self._to_list(row.get(self.USER_PRIOR_COL)) + if not prior_queries: + prior_queries = self._to_list(row.get(self.BROWSER_PRIOR_COL)) + clicked = self._extract_clicked(row.get(self.CANDIDATES_COL)) + self.prompts.append((str(prefix).strip(), prior_queries, clicked)) + if not self.prompts: + raise ValueError(f"No valid prompts found in {filepath}") + + def _count_tokens(self, text: str) -> int: + return len(text.split()) + + def generate_prompt(self) -> tuple: + prefix, prior_queries, clicked = random.choice(self.prompts) + + if self.raw_prompts: + return prefix, clicked + + if self.max_input_tokens > 0: + selected_prior = [] + token_count = self._count_tokens(f"prior query: prefix: {prefix}") + for q in prior_queries: + sep = ", " if selected_prior else "" + if token_count + self._count_tokens(sep + q) > self.max_input_tokens: + break + selected_prior.append(q) + token_count += self._count_tokens(sep + q) + prior_queries = selected_prior + elif self.num_prior >= 0: + prior_queries = prior_queries[:self.num_prior] + + prior_str = ", ".join(prior_queries) if prior_queries else "" + return f"prior query: {prior_str} prefix: {prefix}", clicked + + def __len__(self): + return len(self.prompts) + + +def count_tokens(text) -> int: + if not text: + return 0 + if isinstance(text, list): + return sum(len(str(t).split()) for t in text) + return len(str(text).split()) + + +def build_triton_payload(prompt: str, config: LoadTestConfig) -> dict: + payload = { + "text_input": prompt, + "max_tokens": config.max_tokens, + "beam_width": config.beam_width, + "temperature": config.temperature, + "bad_words": "", + "stop_words": "", + } + if config.use_trie_mask: + payload["logits_post_processor_name"] = config.logits_post_processor_name + if config.min_tokens > 0: + payload["min_length"] = config.min_tokens + return payload + + +def build_vllm_payload(prompt: str, config: LoadTestConfig) -> dict: + payload = { + "prompt": prompt, + "max_tokens": config.max_tokens, + "temperature": config.temperature, + "n": config.beam_width, + "use_beam_search": True, + } + if config.model: + payload["model"] = config.model + return payload + + +def parse_triton_response(data: dict) -> tuple: + output = data.get("text_output", "") + all_beams = output if isinstance(output, list) else [str(output)] if output else [] + return all_beams + + +def parse_vllm_response(data: dict) -> tuple: + choices = data.get("choices", []) + all_beams = [c.get("text", "") for c in choices] + return all_beams + + +async def send_request( + session: aiohttp.ClientSession, + config: LoadTestConfig, + query_gen: QueryGenerator +) -> RequestResult: + prompt, clicked_query = query_gen.generate_prompt() + input_tokens = count_tokens(prompt) + + if config.backend == "vllm": + payload = build_vllm_payload(prompt, config) + else: + payload = build_triton_payload(prompt, config) + + start = time.perf_counter() + try: + async with session.post( + config.url, + json=payload, + timeout=aiohttp.ClientTimeout(total=config.timeout) + ) as resp: + latency_ms = (time.perf_counter() - start) * 1000 + body = await resp.text() + + if resp.status == 200: + try: + data = json.loads(body) + if config.backend == "vllm": + all_beams = parse_vllm_response(data) + else: + all_beams = parse_triton_response(data) + beam_token_counts = [len(str(b).split()) for b in all_beams] + tokens_total = sum(beam_token_counts) + tokens_avg = tokens_total / len(beam_token_counts) if beam_token_counts else 0 + except json.JSONDecodeError: + all_beams = [] + tokens_total = 0 + tokens_avg = 0 + return RequestResult( + latency_ms=latency_ms, + success=True, + status_code=resp.status, + input_text=prompt, + output_text=all_beams, + output_tokens_total=tokens_total, + output_tokens_avg=tokens_avg, + input_tokens=input_tokens, + clicked_query=clicked_query + ) + else: + return RequestResult( + latency_ms=latency_ms, + success=False, + status_code=resp.status, + error=f"HTTP {resp.status}: {body[:100]}" + ) + + except asyncio.TimeoutError: + latency_ms = (time.perf_counter() - start) * 1000 + return RequestResult(latency_ms=latency_ms, success=False, error="Timeout") + except aiohttp.ClientError as e: + latency_ms = (time.perf_counter() - start) * 1000 + return RequestResult(latency_ms=latency_ms, success=False, error=str(e)[:100]) + + +async def worker( + queue: asyncio.Queue, + session: aiohttp.ClientSession, + config: LoadTestConfig, + query_gen: QueryGenerator, + results: LoadTestResults +): + while True: + try: + _ = queue.get_nowait() + except asyncio.QueueEmpty: + break + result = await send_request(session, config, query_gen) + results.results.append(result) + queue.task_done() + + +async def run_load_test(config: LoadTestConfig) -> LoadTestResults: + query_gen = QueryGenerator(config.queries_file, config.num_prior_queries, config.max_input_tokens, config.raw_prompts) + results = LoadTestResults() + queue = asyncio.Queue() + + for i in range(config.total_requests): + queue.put_nowait(i) + + connector = aiohttp.TCPConnector(limit=config.concurrency, limit_per_host=config.concurrency) + async with aiohttp.ClientSession(connector=connector) as session: + print(f"Starting load test: {config.total_requests} requests @ concurrency {config.concurrency}") + print(f"Backend: {config.backend}") + print(f"Target: {config.url}") + if config.backend == "vllm" and config.model: + print(f"Model: {config.model}") + print(f"Beam width / n: {config.beam_width}, Output tokens: {config.max_tokens}, Temperature: {config.temperature}") + print(f"Parquet file: {config.queries_file} ({len(query_gen)} samples)") + if config.max_input_tokens > 0: + print(f"Max input tokens: {config.max_input_tokens}") + if config.raw_prompts: + print(f"Raw prompts: enabled (no prior query formatting)") + else: + prior_desc = "all" if config.num_prior_queries < 0 else str(config.num_prior_queries) + print(f"Num prior queries: {prior_desc}") + if not config.use_trie_mask: + print(f"Trie mask: disabled") + sample_prompt, sample_clicked = query_gen.generate_prompt() + print("\nSample prompt:") + sample_display = sample_prompt[:200] + "..." if len(sample_prompt) > 200 else sample_prompt + print(f" {sample_display}") + if sample_clicked: + print(f" Clicked: {sample_clicked}") + print("-" * 70) + + results.start_time = time.perf_counter() + workers = [ + asyncio.create_task(worker(queue, session, config, query_gen, results)) + for _ in range(config.concurrency) + ] + + total = config.total_requests + while not queue.empty(): + completed = total - queue.qsize() + successful = len([r for r in results.results if r.success]) + failed = len([r for r in results.results if not r.success]) + print(f"\rProgress: {completed}/{total} ({100*completed/total:.0f}%) | OK: {successful} | Fail: {failed}", end="", flush=True) + await asyncio.sleep(0.1) + + await asyncio.gather(*workers) + results.end_time = time.perf_counter() + print(f"\rProgress: {total}/{total} (100%) | OK: {len(results.successful)} | Fail: {len(results.failed)}") + + return results + + +async def run_load_test_rps(config: LoadTestConfig) -> LoadTestResults: + query_gen = QueryGenerator(config.queries_file, config.num_prior_queries, config.max_input_tokens, config.raw_prompts) + results = LoadTestResults() + in_flight = 0 + in_flight_lock = asyncio.Lock() + + async def rps_worker(session, idx): + nonlocal in_flight + async with in_flight_lock: + in_flight += 1 + try: + result = await send_request(session, config, query_gen) + results.results.append(result) + finally: + async with in_flight_lock: + in_flight -= 1 + + connector = aiohttp.TCPConnector(limit=config.max_concurrency, limit_per_host=config.max_concurrency) + async with aiohttp.ClientSession(connector=connector) as session: + print(f"Starting load test: {config.total_requests} requests @ {config.target_rps} RPS target") + print(f"Backend: {config.backend}") + print(f"Target: {config.url}") + if config.backend == "vllm" and config.model: + print(f"Model: {config.model}") + print(f"Beam width / n: {config.beam_width}, Output tokens: {config.max_tokens}, Temperature: {config.temperature}") + print(f"Parquet file: {config.queries_file} ({len(query_gen)} samples)") + if config.max_input_tokens > 0: + print(f"Max input tokens: {config.max_input_tokens}") + if config.raw_prompts: + print(f"Raw prompts: enabled (no prior query formatting)") + else: + prior_desc = "all" if config.num_prior_queries < 0 else str(config.num_prior_queries) + print(f"Num prior queries: {prior_desc}") + if not config.use_trie_mask: + print(f"Trie mask: disabled") + print(f"Max concurrency cap: {config.max_concurrency}") + sample_prompt, sample_clicked = query_gen.generate_prompt() + print("\nSample prompt:") + sample_display = sample_prompt[:200] + "..." if len(sample_prompt) > 200 else sample_prompt + print(f" {sample_display}") + if sample_clicked: + print(f" Clicked: {sample_clicked}") + print("-" * 70) + + interval = 1.0 / config.target_rps + results.start_time = time.perf_counter() + tasks = [] + total = config.total_requests + + for i in range(total): + if in_flight >= config.max_concurrency: + while in_flight >= config.max_concurrency: + await asyncio.sleep(0.001) + + task = asyncio.create_task(rps_worker(session, i)) + tasks.append(task) + + if (i + 1) % 100 == 0 or i == total - 1: + completed = len(results.results) + successful = len([r for r in results.results if r.success]) + failed = len([r for r in results.results if not r.success]) + elapsed = time.perf_counter() - results.start_time + actual_rps = (i + 1) / elapsed if elapsed > 0 else 0 + print(f"\rDispatched: {i+1}/{total} | Done: {completed} | OK: {successful} | Fail: {failed} | InFlight: {in_flight} | RPS: {actual_rps:.1f}", end="", flush=True) + + next_dispatch = results.start_time + (i + 1) * interval + sleep_time = next_dispatch - time.perf_counter() + if sleep_time > 0: + await asyncio.sleep(sleep_time) + + await asyncio.gather(*tasks) + results.end_time = time.perf_counter() + print(f"\rCompleted: {total}/{total} | OK: {len(results.successful)} | Fail: {len(results.failed)} ") + + return results + + +def parse_int_list(value: str) -> list: + return [int(x.strip()) for x in value.split(",")] + + +def get_default_url(backend: str) -> str: + if backend == "vllm": + return "http://localhost:8000/v1/completions" + if backend == "pytorch": + return "http://localhost:8000/v2/models/ensemble/generate" + return "http://localhost:8000/v2/models/ensemble/generate" + + +def parse_float_list(value: str) -> list: + return [float(x.strip()) for x in value.split(",")] + + +def main(): + parser = argparse.ArgumentParser(description="Load test LLM inference servers (Triton TensorRT-LLM or vLLM)") + parser.add_argument("--backend", choices=["triton", "vllm", "pytorch"], default="triton", help="Backend server type (pytorch uses torch.compile)") + parser.add_argument("--model", default="", help="Model name (required for vLLM)") + parser.add_argument("-u", "--url", default=None, help="Server URL (auto-detected based on backend if not set)") + parser.add_argument("-c", "--concurrency", default="10", help="Concurrent requests (comma-separated for sweep, ignored if --rps is set)") + parser.add_argument("-r", "--rps", default=None, help="Target RPS (comma-separated for sweep, overrides --concurrency)") + parser.add_argument("--max-concurrency", type=int, default=500, help="Max concurrent requests when using --rps mode") + parser.add_argument("-n", "--requests", default="2000", help="Total requests (comma-separated for sweep)") + parser.add_argument("--output-tokens", default="0", help="Output tokens to generate (0 = no min, use max=15; comma-separated for sweep)") + parser.add_argument("--min-output-tokens", type=int, default=None, help="Minimum limit for output-tokens sweep") + parser.add_argument("--max-output-tokens", type=int, default=None, help="Maximum limit for output-tokens sweep") + parser.add_argument("-b", "--beam-width", default="20", help="Beam width / n for generation (comma-separated for sweep)") + parser.add_argument("--temperature", type=float, default=1.0, help="Temperature for sampling (default: 1.0)") + parser.add_argument("-q", "--queries-file", default="data/sample.parquet", help="Path to parquet file") + parser.add_argument("-p", "--num-prior", type=int, default=-1, help="Number of prior queries (-1 = all, 0 = none, N = first N)") + parser.add_argument("--max-input-tokens", type=int, default=0, help="Max input tokens (0 = no limit, truncates prior queries)") + parser.add_argument("--hide-samples", action="store_true", help="Hide sample queries and outputs in results") + parser.add_argument("--timeout", type=int, default=60, help="Request timeout in seconds") + parser.add_argument("--seed", type=int, default=None, help="Random seed for reproducibility") + parser.add_argument("--raw-prompts", action="store_true", help="Use prompts directly without 'prior query: ... prefix:' formatting") + parser.add_argument("--no-trie-mask", action="store_true", help="Disable trie_mask logits post processor (for generic LLM testing)") + parser.add_argument("--logits-post-processor-name", type=str, default="trie_mask", + help="Logits post processor name (default: trie_mask, options: trie_mask, trie_mask_cached, trie_mask_batched)") + args = parser.parse_args() + + rps_mode = args.rps is not None + + url = args.url if args.url else get_default_url(args.backend) + + if args.seed is not None: + random.seed(args.seed) + + requests_list = parse_int_list(args.requests) + output_tokens = parse_int_list(args.output_tokens) + beam_widths = parse_int_list(args.beam_width) + + if args.min_output_tokens is not None: + output_tokens = [t for t in output_tokens if t >= args.min_output_tokens] + if args.max_output_tokens is not None: + output_tokens = [t for t in output_tokens if t <= args.max_output_tokens] + if not output_tokens: + print("Error: No output tokens remain after applying min/max limits") + return + + if rps_mode: + rps_targets = parse_float_list(args.rps) + combinations = list(product(rps_targets, requests_list, output_tokens, beam_widths)) + else: + concurrencies = parse_int_list(args.concurrency) + combinations = list(product(concurrencies, requests_list, output_tokens, beam_widths)) + + total_runs = len(combinations) + + if total_runs > 1: + print(f"\n{'='*70}") + print(f"SWEEP MODE: {total_runs} configurations") + if rps_mode: + print(f" Target RPS: {rps_targets}") + else: + print(f" Concurrency: {concurrencies}") + print(f" Requests: {requests_list}") + print(f" Output tokens: {output_tokens}") + print(f" Beam widths: {beam_widths}") + print(f"{'='*70}\n") + + all_results = [] + for i, (rate_or_conc, reqs, output_tok, beam) in enumerate(combinations, 1): + if total_runs > 1: + print(f"\n{'#'*70}") + if rps_mode: + print(f"# RUN {i}/{total_runs}: target_rps={rate_or_conc}, requests={reqs}, output_tokens={output_tok}, beam_width={beam}") + else: + print(f"# RUN {i}/{total_runs}: concurrency={rate_or_conc}, requests={reqs}, output_tokens={output_tok}, beam_width={beam}") + print(f"{'#'*70}") + + config = LoadTestConfig( + url=url, + concurrency=rate_or_conc if not rps_mode else 10, + total_requests=reqs, + max_tokens=output_tok if output_tok > 0 else 15, + min_tokens=output_tok if output_tok > 0 else 0, + beam_width=beam, + temperature=args.temperature, + queries_file=args.queries_file, + num_prior_queries=args.num_prior, + max_input_tokens=args.max_input_tokens, + hide_samples=args.hide_samples, + timeout=args.timeout, + backend=args.backend, + model=args.model, + target_rps=rate_or_conc if rps_mode else 0, + max_concurrency=args.max_concurrency, + raw_prompts=args.raw_prompts, + use_trie_mask=not args.no_trie_mask, + logits_post_processor_name=args.logits_post_processor_name + ) + + if rps_mode: + results = asyncio.run(run_load_test_rps(config)) + else: + results = asyncio.run(run_load_test(config)) + results.summary() + if not config.hide_samples: + results.print_samples() + all_results.append((config, results)) + + if total_runs > 1: + print(f"\n{'='*80}") + print("SWEEP SUMMARY") + print(f"{'='*80}") + + table_data = [] + for config, results in all_results: + duration = results.end_time - results.start_time + actual_rps = len(results.results) / duration if duration > 0 else 0 + p50 = results.percentile(results.latencies, 50) if results.latencies else 0 + p95 = results.percentile(results.latencies, 95) if results.latencies else 0 + p99 = results.percentile(results.latencies, 99) if results.latencies else 0 + ok_pct = 100 * len(results.successful) / len(results.results) if results.results else 0 + + if rps_mode: + table_data.append([ + f"{config.target_rps:.0f}", + config.max_tokens, + config.beam_width, + f"{actual_rps:.1f}", + f"{p50:.1f}", + f"{p95:.1f}", + f"{p99:.1f}", + config.total_requests, + f"{ok_pct:.1f}%" + ]) + else: + table_data.append([ + config.concurrency, + config.max_tokens, + config.beam_width, + f"{actual_rps:.1f}", + f"{p50:.1f}", + f"{p95:.1f}", + f"{p99:.1f}", + config.total_requests, + f"{ok_pct:.1f}%" + ]) + + if rps_mode: + headers = ["TargetRPS", "OutTok", "Beam", "ActualRPS", "P50ms", "P95ms", "P99ms", "Reqs", "OK%"] + else: + headers = ["Conc", "OutTok", "Beam", "RPS", "P50ms", "P95ms", "P99ms", "Reqs", "OK%"] + print(tabulate(table_data, headers=headers, tablefmt="simple", numalign="right", stralign="right")) + print() + + +if __name__ == "__main__": + main() diff --git a/trie_decoding/scripts/triton_qwen.sh b/trie_decoding/scripts/triton_qwen.sh new file mode 100755 index 000000000000..7dcb17a0a33a --- /dev/null +++ b/trie_decoding/scripts/triton_qwen.sh @@ -0,0 +1,250 @@ +#!/bin/bash +############################################################################### +# Qwen3-0.6B: Convert, Build, and Run with TensorRT-LLM + Triton +# +# This script builds TensorRT engines and sets up Triton for Qwen3 models +# with optional native C++ trie-based constrained decoding. +############################################################################### +set -e + +# Get script directory and repo root +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +REPO_ROOT="$(cd "${SCRIPT_DIR}/../.." && pwd)" +WORKSPACE=${WORKSPACE:-/workspace} + +cd ${WORKSPACE} + +############################################################################### +# Configuration +############################################################################### +MODEL_NAME=qwen3-0.6b +HF_MODEL=Qwen/Qwen3-0.6B +HF_MODEL_DIR=models/${MODEL_NAME} +CKPT_DIR=output/ckpt/${MODEL_NAME} +ENGINE_DIR=output/engines/${MODEL_NAME} +TRITON_REPO=output/triton_repo/${MODEL_NAME} + +# Build settings +DTYPE=float16 +TP_SIZE=1 +MAX_BATCH_SIZE=8 +MAX_INPUT_LEN=512 +MAX_SEQ_LEN=528 +MAX_BEAM_WIDTH=128 + +# Trie-based constrained decoding +TRIE_PATH=${TRIE_PATH:-} +TRIE_BINARY_PATH=${TRIE_BINARY_PATH:-} +GUIDED_DECODING_BACKEND="" # "", "trie" for native C++ +KV_CACHE_FRACTION=0.5 + +############################################################################### +# Parse arguments +############################################################################### +BUILD_ONLY=false +START_ONLY=false + +while [[ $# -gt 0 ]]; do + case $1 in + --build-only) BUILD_ONLY=true; shift ;; + --start-only) START_ONLY=true; shift ;; + --beam-width) MAX_BEAM_WIDTH=$2; shift 2 ;; + --input-len) MAX_INPUT_LEN=$2; MAX_SEQ_LEN=$((MAX_INPUT_LEN + 16)); shift 2 ;; + --batch-size) MAX_BATCH_SIZE=$2; shift 2 ;; + --kv-cache-fraction) KV_CACHE_FRACTION=$2; shift 2 ;; + --model) MODEL_NAME=$2; shift 2 ;; + --hf-model) HF_MODEL=$2; shift 2 ;; + --trie-path) TRIE_PATH=$2; shift 2 ;; + --trie-binary-path) TRIE_BINARY_PATH=$2; GUIDED_DECODING_BACKEND="trie"; shift 2 ;; + --native-trie) GUIDED_DECODING_BACKEND="trie"; shift ;; + -h|--help) + echo "Usage: $0 [OPTIONS]" + echo "" + echo "Options:" + echo " --build-only Build engines only, don't start server" + echo " --start-only Skip build, just start Triton server" + echo " --beam-width N Set beam width (default: 128)" + echo " --input-len N Max input length (default: 512)" + echo " --batch-size N Max batch size (default: 8)" + echo " --kv-cache-fraction F GPU memory for KV cache (default: 0.5)" + echo " --model NAME Model name (default: qwen3-0.6b)" + echo " --hf-model HF_NAME HuggingFace model (default: Qwen/Qwen3-0.6B)" + echo " --trie-path PATH Path to pickle trie file (Python fallback)" + echo " --trie-binary-path P Path to binary trie file (enables native C++)" + echo " --native-trie Enable native C++ trie decoding" + echo " -h, --help Show this help message" + exit 0 + ;; + *) echo "Unknown option: $1"; exit 1 ;; + esac +done + +# Update paths based on model name +HF_MODEL_DIR=models/${MODEL_NAME} +CKPT_DIR=output/ckpt/${MODEL_NAME} +ENGINE_DIR=output/engines/${MODEL_NAME} +TRITON_REPO=output/triton_repo/${MODEL_NAME} + +############################################################################### +# Step 1: Download model (if needed) +############################################################################### +download_model() { + echo "=== Step 1: Downloading model ===" + if [ -d "${HF_MODEL_DIR}" ] && [ -f "${HF_MODEL_DIR}/config.json" ]; then + echo "Model already exists at ${HF_MODEL_DIR}" + else + echo "Downloading ${HF_MODEL}..." + mkdir -p ${HF_MODEL_DIR} + huggingface-cli download ${HF_MODEL} --local-dir ${HF_MODEL_DIR} + fi +} + +############################################################################### +# Step 2: Convert checkpoint +############################################################################### +convert_checkpoint() { + echo "=== Step 2: Converting checkpoint ===" + rm -rf ${CKPT_DIR} + mkdir -p ${CKPT_DIR} + + python3 ${REPO_ROOT}/examples/models/core/qwen/convert_checkpoint.py \ + --model_dir ${HF_MODEL_DIR} \ + --output_dir ${CKPT_DIR} \ + --dtype ${DTYPE} \ + --tp_size ${TP_SIZE} + + echo "Checkpoint saved to ${CKPT_DIR}" +} + +############################################################################### +# Step 3: Build TensorRT engine +############################################################################### +build_engine() { + echo "=== Step 3: Building TensorRT engine ===" + rm -rf ${ENGINE_DIR} + mkdir -p ${ENGINE_DIR} + + TIMING_CACHE=output/timing_cache/${MODEL_NAME}.cache + mkdir -p output/timing_cache + + CACHE_ARGS="" + if [ -f "${TIMING_CACHE}" ]; then + echo "Using timing cache: ${TIMING_CACHE}" + CACHE_ARGS="--input_timing_cache ${TIMING_CACHE}" + fi + + trtllm-build \ + --checkpoint_dir ${CKPT_DIR} \ + --output_dir ${ENGINE_DIR} \ + --gemm_plugin ${DTYPE} \ + --max_batch_size ${MAX_BATCH_SIZE} \ + --max_input_len ${MAX_INPUT_LEN} \ + --max_seq_len ${MAX_SEQ_LEN} \ + --max_beam_width ${MAX_BEAM_WIDTH} \ + ${CACHE_ARGS} \ + --output_timing_cache ${TIMING_CACHE} + + echo "Engine saved to ${ENGINE_DIR}" +} + +############################################################################### +# Step 4: Setup Triton repository +############################################################################### +setup_triton_repo() { + echo "=== Step 4: Setting up Triton repository ===" + rm -rf ${TRITON_REPO} + cp -r ${REPO_ROOT}/triton_backend/all_models/inflight_batcher_llm ${TRITON_REPO} + + # tensorrt_llm config + python3 ${REPO_ROOT}/triton_backend/tools/fill_template.py \ + -i ${TRITON_REPO}/tensorrt_llm/config.pbtxt \ + "triton_backend:python,triton_max_batch_size:${MAX_BATCH_SIZE},decoupled_mode:False,max_beam_width:${MAX_BEAM_WIDTH},engine_dir:${WORKSPACE}/${ENGINE_DIR},kv_cache_free_gpu_mem_fraction:${KV_CACHE_FRACTION},exclude_input_in_output:True,enable_kv_cache_reuse:False,batching_strategy:inflight_fused_batching,max_queue_delay_microseconds:200,enable_chunked_context:False,max_queue_size:0,batch_scheduler_policy:guaranteed_no_evict,encoder_input_features_data_type:TYPE_FP16,logits_datatype:TYPE_FP16,prompt_embedding_table_data_type:TYPE_FP16,encoder_engine_dir:,max_tokens_in_paged_kv_cache:,max_attention_window_size:,sink_token_length:,cross_kv_cache_fraction:,kv_cache_host_memory_bytes:0,kv_cache_onboard_blocks:True,enable_trt_overlap:False,cancellation_check_period_ms:100,stats_check_period_ms:100,iter_stats_max_iterations:1000,request_stats_max_iterations:0,normalize_log_probs:True,gpu_device_ids:,participant_ids:,num_nodes:,lora_cache_optimal_adapter_size:,lora_cache_max_adapter_size:,lora_cache_gpu_memory_fraction:,lora_cache_host_memory_bytes:,lora_prefetch_dir:,decoding_mode:,lookahead_window_size:,lookahead_ngram_size:,lookahead_verification_set_size:,medusa_choices:,eagle_choices:,gpu_weights_percent:,enable_context_fmha_fp32_acc:,multi_block_mode:,cuda_graph_mode:,cuda_graph_cache_size:,speculative_decoding_fast_logits:,tokenizer_dir:,guided_decoding_backend:${GUIDED_DECODING_BACKEND},xgrammar_tokenizer_info_path:,enable_trie_mask:false,trie_path:${TRIE_PATH},trie_binary_path:${TRIE_BINARY_PATH}" + + # preprocessing config + python3 ${REPO_ROOT}/triton_backend/tools/fill_template.py \ + -i ${TRITON_REPO}/preprocessing/config.pbtxt \ + "tokenizer_dir:${WORKSPACE}/${HF_MODEL_DIR},triton_max_batch_size:${MAX_BATCH_SIZE},preprocessing_instance_count:2,add_special_tokens:True,max_num_images:0,multimodal_model_path:,engine_dir:" + + # postprocessing config + python3 ${REPO_ROOT}/triton_backend/tools/fill_template.py \ + -i ${TRITON_REPO}/postprocessing/config.pbtxt \ + "tokenizer_dir:${WORKSPACE}/${HF_MODEL_DIR},triton_max_batch_size:${MAX_BATCH_SIZE},postprocessing_instance_count:2,skip_special_tokens:True" + + # ensemble config + python3 ${REPO_ROOT}/triton_backend/tools/fill_template.py \ + -i ${TRITON_REPO}/ensemble/config.pbtxt \ + "triton_max_batch_size:${MAX_BATCH_SIZE},logits_datatype:TYPE_FP16" + + # tensorrt_llm_bls config + python3 ${REPO_ROOT}/triton_backend/tools/fill_template.py \ + -i ${TRITON_REPO}/tensorrt_llm_bls/config.pbtxt \ + "triton_max_batch_size:${MAX_BATCH_SIZE},decoupled_mode:False,bls_instance_count:1,accumulate_tokens:False,logits_datatype:TYPE_FP16,prompt_embedding_table_data_type:TYPE_FP16,tensorrt_llm_model_name:tensorrt_llm,tensorrt_llm_draft_model_name:,multimodal_encoders_name:" + + # Verify no unsubstituted variables + echo "Checking for unsubstituted template variables..." + if grep -rq '\${' ${TRITON_REPO}/*/config.pbtxt 2>/dev/null; then + echo "WARNING: Some template variables may not be substituted" + fi + + echo "✓ Triton repository ready" +} + +############################################################################### +# Step 5: Start Triton server +############################################################################### +start_triton() { + echo "=== Step 5: Starting Triton server ===" + echo "Model: ${MODEL_NAME}" + echo "Engine: ${ENGINE_DIR}" + echo "Max batch size: ${MAX_BATCH_SIZE}" + echo "Max beam width: ${MAX_BEAM_WIDTH}" + if [ -n "${GUIDED_DECODING_BACKEND}" ]; then + echo "Guided decoding: ${GUIDED_DECODING_BACKEND}" + echo "Trie binary path: ${TRIE_BINARY_PATH}" + fi + echo "" + + tritonserver \ + --model-repository=${WORKSPACE}/${TRITON_REPO} \ + --http-port=8000 \ + --grpc-port=8001 \ + --metrics-port=8002 \ + --log-verbose=0 +} + +############################################################################### +# Main +############################################################################### +echo "========================================================================" +echo "Qwen TensorRT-LLM + Triton" +echo "========================================================================" +echo "Configuration:" +echo " Model: ${MODEL_NAME}" +echo " HF Model: ${HF_MODEL}" +echo " DTYPE: ${DTYPE}" +echo " MAX_INPUT_LEN: ${MAX_INPUT_LEN}" +echo " MAX_SEQ_LEN: ${MAX_SEQ_LEN}" +echo " MAX_BEAM_WIDTH: ${MAX_BEAM_WIDTH}" +echo " MAX_BATCH_SIZE: ${MAX_BATCH_SIZE}" +if [ -n "${GUIDED_DECODING_BACKEND}" ]; then + echo " GUIDED_DECODING: ${GUIDED_DECODING_BACKEND}" +fi +echo "========================================================================" +echo "" + +if [ "$START_ONLY" = true ]; then + start_triton +elif [ "$BUILD_ONLY" = true ]; then + download_model + convert_checkpoint + build_engine + setup_triton_repo + echo "" + echo "Build complete! Run with: $0 --start-only" +else + download_model + convert_checkpoint + build_engine + setup_triton_repo + start_triton +fi diff --git a/trie_decoding/test_native_trie_decoding.py b/trie_decoding/test_native_trie_decoding.py new file mode 100644 index 000000000000..657fe7b3a14f --- /dev/null +++ b/trie_decoding/test_native_trie_decoding.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python3 +""" +Test native C++ trie guided decoding through TRT-LLM executor API. + +This demonstrates how to use the native kTRIE backend that was added +to TensorRT-LLM for efficient trie-based constrained decoding. +""" + +import sys + + +def test_guided_decoding_config(): + """Test that GuidedDecodingConfig with TRIE backend is available.""" + try: + from tensorrt_llm.bindings.executor import ( + GuidedDecodingConfig, ExecutorConfig + ) + + # Check TRIE backend is available + trie_backend = GuidedDecodingConfig.GuidedDecodingBackend.TRIE + print(f"TRIE backend available: {trie_backend}") + + # Create a GuidedDecodingConfig with TRIE backend + trie_path = "/workspace/autosuggest_loadtest/output/trie/trie_1m.bin" + config = GuidedDecodingConfig( + backend=trie_backend, + trie_path=trie_path + ) + + print(f"Created GuidedDecodingConfig:") + print(f" Backend: {config.backend}") + print(f" Trie path: {config.trie_path}") + + # Verify it can be set in ExecutorConfig + exec_config = ExecutorConfig() + exec_config.guided_decoding_config = config + print(f"\nSuccessfully set guided_decoding_config in ExecutorConfig") + + retrieved = exec_config.guided_decoding_config + if retrieved: + print(f"Retrieved config - Backend: {retrieved.backend}, Path: {retrieved.trie_path}") + + return True + + except ImportError as e: + print(f"Import error: {e}") + return False + except Exception as e: + print(f"Error: {e}") + return False + + +def main(): + print("Testing Native C++ Trie Guided Decoding") + print("=" * 50) + + if test_guided_decoding_config(): + print("\n[PASS] Native trie guided decoding bindings are working!") + print("\nTo use in production:") + print("1. Convert pickle trie to binary: python convert_trie_to_binary.py trie.pkl trie.bin") + print("2. Configure ExecutorConfig with GuidedDecodingConfig(backend=TRIE, trie_path='...')") + print("3. The C++ GuidedDecoder will handle constraint masking without Python callbacks") + else: + print("\n[FAIL] Could not initialize trie guided decoding") + sys.exit(1) + + +if __name__ == '__main__': + main() diff --git a/trie_decoding/trie.py b/trie_decoding/trie.py new file mode 100644 index 000000000000..9cc320c6ef8b --- /dev/null +++ b/trie_decoding/trie.py @@ -0,0 +1,105 @@ +""" +Trie data structures for constrained decoding. + +This module is the single source of truth for Trie classes. +Import from here in both: + - generate_trie.py (for creating tries) + - Triton's trie_loader.py (for loading tries) +""" + +from typing import Dict, List, Set + + +class TrieNode: + """Memory-efficient trie node using __slots__.""" + __slots__ = ['children', 'is_terminal'] + + def __init__(self): + self.children: Dict[int, 'TrieNode'] = {} + self.is_terminal: bool = False + + +class Trie: + """ + Memory-efficient trie for constrained decoding. + + Interface: + - allowed_next(sequence: List[int]) -> List[int] + - insert(sequence: List[int]) -> int + - contains_prefix(sequence: List[int]) -> bool + """ + + def __init__(self): + self.root = TrieNode() + self.node_count = 1 + self.sequence_count = 0 + + def insert(self, sequence: List[int]) -> int: + """Insert a token sequence, return number of new nodes created.""" + node = self.root + new_nodes = 0 + for token in sequence: + if token not in node.children: + node.children[token] = TrieNode() + new_nodes += 1 + self.node_count += 1 + node = node.children[token] + node.is_terminal = True + self.sequence_count += 1 + return new_nodes + + def allowed_next(self, sequence: List[int]) -> List[int]: + """Return list of valid next token IDs given sequence prefix.""" + node = self.root + for token in sequence: + if token not in node.children: + return [] + node = node.children[token] + return list(node.children.keys()) + + def contains_prefix(self, sequence: List[int]) -> bool: + """Check if sequence is a valid prefix in the trie.""" + node = self.root + for token in sequence: + if token not in node.children: + return False + node = node.children[token] + return True + + +class CompactTrie: + """ + More memory-efficient trie using arrays instead of dicts. + Better for very large tries but slightly slower lookup. + """ + + def __init__(self): + self.nodes: List[Dict[int, int]] = [{}] # node_id -> {token: child_node_id} + self.terminals: Set[int] = set() + self.root = 0 + + @property + def node_count(self): + return len(self.nodes) + + def insert(self, sequence: List[int]) -> int: + node_id = 0 + new_nodes = 0 + for token in sequence: + if token not in self.nodes[node_id]: + new_id = len(self.nodes) + self.nodes.append({}) + self.nodes[node_id][token] = new_id + new_nodes += 1 + node_id = self.nodes[node_id][token] + self.terminals.add(node_id) + return new_nodes + + def allowed_next(self, sequence: List[int]) -> List[int]: + node_id = 0 + for token in sequence: + if token not in self.nodes[node_id]: + return [] + node_id = self.nodes[node_id][token] + return list(self.nodes[node_id].keys()) + diff --git a/triton_backend/all_models/inflight_batcher_llm/ensemble/config.pbtxt b/triton_backend/all_models/inflight_batcher_llm/ensemble/config.pbtxt index 0bfa7ab90a44..9b31b3d3c142 100644 --- a/triton_backend/all_models/inflight_batcher_llm/ensemble/config.pbtxt +++ b/triton_backend/all_models/inflight_batcher_llm/ensemble/config.pbtxt @@ -278,6 +278,12 @@ input [ dims: [ 1 ] optional: true allow_ragged_batch: true + }, + { + name: "logits_post_processor_name" + data_type: TYPE_STRING + dims: [ 1 ] + optional: true } ] output [ @@ -630,6 +636,10 @@ ensemble_scheduling { input_map { key: "guided_decoding_guide", value: "guided_decoding_guide" + }, + input_map { + key: "logits_post_processor_name", + value: "logits_post_processor_name" } output_map { key: "output_ids" diff --git a/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/lib/__init__.py b/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/lib/__init__.py new file mode 100644 index 000000000000..3d8197469ee6 --- /dev/null +++ b/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/lib/__init__.py @@ -0,0 +1,2 @@ +# Trie-based constrained decoding for TensorRT-LLM Triton backend + diff --git a/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/lib/trie_loader.py b/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/lib/trie_loader.py new file mode 100644 index 000000000000..333db7807681 --- /dev/null +++ b/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/lib/trie_loader.py @@ -0,0 +1,130 @@ +""" +Trie loader for Triton inference server. + +Handles loading pickled tries created by generate_trie.py. +Uses a custom unpickler to remap class references. +""" + +import io +import os +import pickle +from typing import Any, Dict, List, Optional +import triton_python_backend_utils as pb_utils + + +# ============================================================================= +# Trie classes - must match src/trie.py exactly for pickle compatibility +# ============================================================================= + +class TrieNode: + """Memory-efficient trie node using __slots__.""" + __slots__ = ['children', 'is_terminal'] + + def __init__(self): + self.children: Dict[int, 'TrieNode'] = {} + self.is_terminal: bool = False + + +class Trie: + """Trie for constrained decoding.""" + + def __init__(self): + self.root = TrieNode() + self.node_count = 1 + self.sequence_count = 0 + + def allowed_next(self, sequence: List[int]) -> List[int]: + """Return list of valid next token IDs given sequence prefix.""" + node = self.root + for token in sequence: + if token not in node.children: + return [] + node = node.children[token] + return list(node.children.keys()) + + +class CompactTrie: + """Array-based trie for very large tries.""" + + def __init__(self): + self.nodes: List[Dict[int, int]] = [{}] + self.terminals = set() + self.root = 0 + + @property + def node_count(self): + return len(self.nodes) + + def allowed_next(self, sequence: List[int]) -> List[int]: + node_id = 0 + for token in sequence: + if token not in self.nodes[node_id]: + return [] + node_id = self.nodes[node_id][token] + return list(self.nodes[node_id].keys()) + + +# ============================================================================= +# Custom unpickler to remap classes from any module +# ============================================================================= + +class TrieUnpickler(pickle.Unpickler): + """Unpickler that remaps Trie classes from any source module.""" + + CLASS_MAP = { + 'Trie': Trie, + 'TrieNode': TrieNode, + 'CompactTrie': CompactTrie, + } + + def find_class(self, module: str, name: str): + if name in self.CLASS_MAP: + return self.CLASS_MAP[name] + return super().find_class(module, name) + + +# ============================================================================= +# Placeholder trie (allows all tokens) +# ============================================================================= + +class PlaceholderTrie: + """Placeholder that allows all tokens - used when no trie is loaded.""" + + def __init__(self, vocab_size: int = 50257): + self.vocab_size = vocab_size + self.node_count = 1 + + def allowed_next(self, sequence: List[int]) -> List[int]: + return list(range(self.vocab_size)) + + +# ============================================================================= +# Public API +# ============================================================================= + +def load_trie(trie_path: Optional[str] = None, vocab_size: int = 50257) -> Any: + """ + Load trie for constrained decoding. + + Args: + trie_path: Path to .pkl file. Falls back to TRIE_PATH env var. + vocab_size: Vocab size for placeholder trie. + + Returns: + Trie with .allowed_next(sequence) -> List[int] + """ + trie_path = trie_path or os.environ.get("TRIE_PATH") + + if not trie_path or not os.path.exists(trie_path): + pb_utils.Logger.log_info("No trie path specified, using placeholder (allows all tokens)") + return PlaceholderTrie(vocab_size) + + pb_utils.Logger.log_info(f"Loading trie from {trie_path}") + try: + with open(trie_path, 'rb') as f: + trie = TrieUnpickler(f).load() + pb_utils.Logger.log_info(f"Loaded trie with {trie.node_count:,} nodes") + return trie + except Exception as e: + pb_utils.Logger.log_warn(f"Failed to load trie: {e}, using placeholder") + return PlaceholderTrie(vocab_size) diff --git a/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/lib/trie_mask.py b/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/lib/trie_mask.py new file mode 100644 index 000000000000..606f97f943bc --- /dev/null +++ b/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/lib/trie_mask.py @@ -0,0 +1,194 @@ +from typing import List, Optional, Dict, Any +import torch +import triton_python_backend_utils as pb_utils + +NEG_INF = float("-inf") + + +class TrieMaskLogitsPostProcessor: + """ + High-performance stateful logits post-processor for trie-based constrained decoding. + + Optimizations: + - O(1) trie traversal (tracks current node per request) + - Caches GPU tensors for frequently-used nodes + - Minimal tensor operations + """ + PROCESSOR_NAME = "trie_mask" + + def __init__(self, trie: Any): + self.trie = trie + # Per-request state: (input_length, current_trie_node) + self._state: Dict[int, tuple] = {} + # Cache: node_id -> (allowed_tensor, device) + self._tensor_cache: Dict[int, torch.Tensor] = {} + self._cache_max_size = 10000 + + def _get_allowed_tensor(self, node, allowed_ids: List[int], device) -> torch.Tensor: + """Get or create cached tensor for allowed token IDs.""" + node_id = id(node) + cache_key = (node_id, device.index if hasattr(device, 'index') else 0) + + if cache_key not in self._tensor_cache: + if len(self._tensor_cache) < self._cache_max_size: + self._tensor_cache[cache_key] = torch.tensor( + allowed_ids, device=device, dtype=torch.int64 + ) + + cached = self._tensor_cache.get(cache_key) + if cached is not None and cached.device == device: + return cached + + # Fallback: create tensor (not cached) + return torch.tensor(allowed_ids, device=device, dtype=torch.int64) + + def __call__( + self, + req_id: int, + logits: torch.Tensor, + token_ids: List[List[int]], + stream_ptr: int, + client_id: Optional[int] + ) -> None: + full_seq = token_ids[0] if token_ids else [] + + # Initialize state on first call + if req_id not in self._state: + self._state[req_id] = (len(full_seq), self.trie.root) + + input_len, current_node = self._state[req_id] + + # Advance node if we have new generated tokens + seq_len = len(full_seq) + if seq_len > input_len: + last_token = full_seq[-1] + children = getattr(current_node, 'children', None) + if children and last_token in children: + current_node = children[last_token] + self._state[req_id] = (input_len, current_node) + else: + return # Off trie, no constraints + + # Get allowed tokens + children = getattr(current_node, 'children', None) + if not children: + return + + allowed_ids = list(children.keys()) + if not allowed_ids: + return + + # Apply mask with cached tensor + stream = torch.cuda.ExternalStream(stream_ptr) if stream_ptr else None + with torch.cuda.stream(stream): + allowed = self._get_allowed_tensor(current_node, allowed_ids, logits.device) + # More efficient: gather, fill, scatter + kept = logits.index_select(-1, allowed) + logits.fill_(NEG_INF) + logits.scatter_(-1, allowed.unsqueeze(0).expand_as(kept), kept) + + def cleanup_request(self, req_id: int) -> None: + self._state.pop(req_id, None) + + +class StatefulTrieMaskProcessor: + """ + Stateful version that tracks trie position per request for O(1) transitions. + + Use this for large tries where re-traversing from root each step is expensive. + Requires trie nodes to have a .children dict and .allowed_tokens property. + """ + PROCESSOR_NAME = "trie_mask_stateful" + + def __init__(self, trie: Any): + self.trie = trie + self._node_state: Dict[int, Any] = {} + + def __call__( + self, + req_id: int, + logits: torch.Tensor, + token_ids: List[List[int]], + stream_ptr: int, + client_id: Optional[int] + ) -> None: + if req_id not in self._node_state: + self._node_state[req_id] = self.trie.root + + current_node = self._node_state[req_id] + seq = token_ids[0] if token_ids else [] + + if seq: + last_token = seq[-1] + if hasattr(current_node, 'children') and last_token in current_node.children: + current_node = current_node.children[last_token] + self._node_state[req_id] = current_node + else: + self._node_state[req_id] = self.trie.root + current_node = self.trie.root + + if hasattr(current_node, 'allowed_tokens'): + allowed_ids = current_node.allowed_tokens + elif hasattr(current_node, 'children'): + allowed_ids = list(current_node.children.keys()) + else: + return + + if not allowed_ids: + return + + stream = torch.cuda.ExternalStream(stream_ptr) if stream_ptr else None + + with torch.cuda.stream(stream): + allowed = torch.tensor(allowed_ids, device=logits.device, dtype=torch.int64) + kept_values = logits.index_select(dim=-1, index=allowed) + logits.fill_(NEG_INF) + logits.index_copy_(dim=-1, index=allowed, source=kept_values) + + def cleanup_request(self, req_id: int) -> None: + self._node_state.pop(req_id, None) + + +class TrieMaskBatchedProcessor: + """ + Batched version for better throughput with multiple concurrent requests. + + Use processor_batched instead of processor_map when you expect many + simultaneous requests needing trie masking. + """ + PROCESSOR_NAME = "trie_mask_batched" + + def __init__(self, trie: Any): + self.trie = trie + + def __call__( + self, + req_ids_batch: List[int], + logits_batch: List[torch.Tensor], + ids_batch: List[List[List[int]]], + stream_ptr: int, + client_ids_batch: List[Optional[int]] + ) -> None: + stream = torch.cuda.ExternalStream(stream_ptr) if stream_ptr else None + + with torch.cuda.stream(stream): + for logits, token_ids in zip(logits_batch, ids_batch): + seq = token_ids[0] if token_ids else [] + allowed_ids = self.trie.allowed_next(seq) + + if not allowed_ids: + continue + + allowed = torch.tensor(allowed_ids, device=logits.device, dtype=torch.int64) + kept_values = logits.index_select(dim=-1, index=allowed) + logits.fill_(NEG_INF) + logits.index_copy_(dim=-1, index=allowed, source=kept_values) + + +# Import xGR-optimized processors +try: + from trie_mask_xgr import XGRTrieMaskProcessor, XGRBatchedProcessor +except ImportError: + # Fallback - create an alias + XGRTrieMaskProcessor = TrieMaskLogitsPostProcessor + XGRBatchedProcessor = TrieMaskBatchedProcessor diff --git a/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/lib/trie_mask_xgr.py b/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/lib/trie_mask_xgr.py new file mode 100644 index 000000000000..327a39694a53 --- /dev/null +++ b/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/lib/trie_mask_xgr.py @@ -0,0 +1,352 @@ +""" +xGR-Optimized Trie Mask LogitsPostProcessor + +Based on techniques from xGR paper (https://arxiv.org/pdf/2512.11529): +- Pre-computed GPU bitmasks per trie node (lazy initialization) +- Batched processor support for reduced GIL contention +- CUDA stream-aware operations +- Memory-efficient caching with LRU eviction +- Sparse/dense hybrid storage strategy + +Usage: + from trie_mask_xgr import XGRTrieMaskProcessor, XGRBatchedProcessor + + processor = XGRTrieMaskProcessor(trie) + # Register with TRT-LLM executor config +""" + +from typing import Dict, List, Optional, Any, Tuple +from collections import OrderedDict +import torch + +NEG_INF = float("-inf") + + +class BitmaskCache: + """ + LRU cache for GPU bitmasks with memory management. + + xGR insight: Cache frequently-accessed node masks on GPU, + evict least-recently-used to manage memory. + """ + + def __init__(self, max_size: int = 10000, vocab_size: int = 0): + self.max_size = max_size + self.vocab_size = vocab_size + self._cache: OrderedDict[int, torch.Tensor] = OrderedDict() + self._device = None + + def get(self, node_id: int) -> Optional[torch.Tensor]: + if node_id in self._cache: + self._cache.move_to_end(node_id) + return self._cache[node_id] + return None + + def put(self, node_id: int, bitmask: torch.Tensor) -> None: + if node_id in self._cache: + self._cache.move_to_end(node_id) + return + + if len(self._cache) >= self.max_size: + oldest_key = next(iter(self._cache)) + del self._cache[oldest_key] + + self._cache[node_id] = bitmask + self._device = bitmask.device + + def clear(self) -> None: + self._cache.clear() + + def __len__(self) -> int: + return len(self._cache) + + +class XGRTrieMaskProcessor: + """ + xGR-optimized stateful logits post-processor for trie-based constrained decoding. + + Key optimizations: + - O(1) trie traversal via stateful node tracking + - Pre-computed GPU bitmasks with LRU caching + - Boolean tensor masking (faster than index_select + scatter) + - CUDA stream-aware for async operation + """ + PROCESSOR_NAME = "trie_mask" + + def __init__(self, trie: Any, cache_size: int = 10000): + self.trie = trie + self._state: Dict[int, Tuple[int, Any]] = {} + self._bitmask_cache = BitmaskCache(max_size=cache_size) + self._vocab_size = 0 + self._root_bitmask: Optional[torch.Tensor] = None + + def _create_bitmask(self, allowed_ids: List[int], device: torch.device) -> torch.Tensor: + """Create a boolean bitmask tensor on GPU.""" + bitmask = torch.zeros(self._vocab_size, dtype=torch.bool, device=device) + if allowed_ids: + indices = torch.tensor(allowed_ids, dtype=torch.long, device=device) + bitmask.index_fill_(0, indices, True) + return bitmask + + def _get_bitmask(self, node: Any, allowed_ids: List[int], device: torch.device) -> torch.Tensor: + """Get or create cached bitmask for node.""" + node_id = id(node) + + cached = self._bitmask_cache.get(node_id) + if cached is not None and cached.device == device: + return cached + + bitmask = self._create_bitmask(allowed_ids, device) + self._bitmask_cache.put(node_id, bitmask) + return bitmask + + def __call__( + self, + req_id: int, + logits: torch.Tensor, + token_ids: List[List[int]], + stream_ptr: int, + client_id: Optional[int] + ) -> None: + """Apply trie mask to logits tensor.""" + if self._vocab_size == 0: + self._vocab_size = logits.shape[-1] + + full_seq = token_ids[0] if token_ids else [] + + if req_id not in self._state: + self._state[req_id] = (len(full_seq), self.trie.root) + + input_len, current_node = self._state[req_id] + + seq_len = len(full_seq) + if seq_len > input_len: + last_token = full_seq[-1] + children = getattr(current_node, 'children', None) + if children and last_token in children: + current_node = children[last_token] + self._state[req_id] = (input_len, current_node) + else: + return + + children = getattr(current_node, 'children', None) + if not children: + return + + allowed_ids = list(children.keys()) + if not allowed_ids: + return + + stream = torch.cuda.ExternalStream(stream_ptr) if stream_ptr else None + with torch.cuda.stream(stream): + bitmask = self._get_bitmask(current_node, allowed_ids, logits.device) + logits.masked_fill_(~bitmask, NEG_INF) + + def cleanup_request(self, req_id: int) -> None: + self._state.pop(req_id, None) + + +class XGRBatchedProcessor: + """ + Batched version of xGR processor for better throughput. + + xGR insight: Process multiple requests in single callback to: + - Reduce Python GIL contention + - Amortize kernel launch overhead + - Enable potential GPU parallelism + """ + PROCESSOR_NAME = "trie_mask_batched" + + def __init__(self, trie: Any, cache_size: int = 10000): + self.trie = trie + self._state: Dict[int, Tuple[int, Any]] = {} + self._bitmask_cache = BitmaskCache(max_size=cache_size) + self._vocab_size = 0 + + def _create_bitmask(self, allowed_ids: List[int], device: torch.device) -> torch.Tensor: + bitmask = torch.zeros(self._vocab_size, dtype=torch.bool, device=device) + if allowed_ids: + indices = torch.tensor(allowed_ids, dtype=torch.long, device=device) + bitmask.index_fill_(0, indices, True) + return bitmask + + def _get_bitmask(self, node: Any, allowed_ids: List[int], device: torch.device) -> torch.Tensor: + node_id = id(node) + cached = self._bitmask_cache.get(node_id) + if cached is not None and cached.device == device: + return cached + bitmask = self._create_bitmask(allowed_ids, device) + self._bitmask_cache.put(node_id, bitmask) + return bitmask + + def _advance_state(self, req_id: int, token_ids: List[List[int]]) -> Tuple[Any, List[int]]: + """Advance trie state and return current node + allowed tokens.""" + full_seq = token_ids[0] if token_ids else [] + + if req_id not in self._state: + self._state[req_id] = (len(full_seq), self.trie.root) + + input_len, current_node = self._state[req_id] + + if len(full_seq) > input_len: + last_token = full_seq[-1] + children = getattr(current_node, 'children', None) + if children and last_token in children: + current_node = children[last_token] + self._state[req_id] = (input_len, current_node) + else: + return None, [] + + children = getattr(current_node, 'children', None) + if not children: + return None, [] + + return current_node, list(children.keys()) + + def __call__( + self, + req_ids_batch: List[int], + logits_batch: List[torch.Tensor], + ids_batch: List[List[List[int]]], + stream_ptr: int, + client_ids_batch: List[Optional[int]] + ) -> None: + """Process batch of requests.""" + if not logits_batch: + return + + if self._vocab_size == 0: + self._vocab_size = logits_batch[0].shape[-1] + + stream = torch.cuda.ExternalStream(stream_ptr) if stream_ptr else None + + with torch.cuda.stream(stream): + for req_id, logits, token_ids in zip(req_ids_batch, logits_batch, ids_batch): + node, allowed_ids = self._advance_state(req_id, token_ids) + if node is None or not allowed_ids: + continue + + bitmask = self._get_bitmask(node, allowed_ids, logits.device) + logits.masked_fill_(~bitmask, NEG_INF) + + def cleanup_request(self, req_id: int) -> None: + self._state.pop(req_id, None) + + +class XGRPrecomputedProcessor: + """ + Processor variant with pre-computed bitmasks for all reachable nodes. + + xGR insight from Section 6.1: + - For first decode step: use dense pre-computed masks + - For later steps: masks are smaller, can compute on demand + + Trade-off: Higher memory usage, but zero mask computation latency. + Best for small-to-medium tries where memory is not constrained. + """ + PROCESSOR_NAME = "trie_mask_precomputed" + + def __init__(self, trie: Any, vocab_size: int, device: str = "cuda"): + self.trie = trie + self.vocab_size = vocab_size + self.device = torch.device(device) + self._state: Dict[int, Tuple[int, Any]] = {} + self._bitmasks: Dict[int, torch.Tensor] = {} + self._precompute_masks() + + def _precompute_masks(self) -> None: + """Pre-compute bitmasks for all nodes (BFS traversal).""" + from collections import deque + + queue = deque([self.trie.root]) + visited = set() + + while queue: + node = queue.popleft() + node_id = id(node) + + if node_id in visited: + continue + visited.add(node_id) + + children = getattr(node, 'children', None) + if children: + allowed_ids = list(children.keys()) + bitmask = torch.zeros(self.vocab_size, dtype=torch.bool, device=self.device) + if allowed_ids: + indices = torch.tensor(allowed_ids, dtype=torch.long, device=self.device) + bitmask.index_fill_(0, indices, True) + self._bitmasks[node_id] = bitmask + + for child in children.values(): + queue.append(child) + + def __call__( + self, + req_id: int, + logits: torch.Tensor, + token_ids: List[List[int]], + stream_ptr: int, + client_id: Optional[int] + ) -> None: + full_seq = token_ids[0] if token_ids else [] + + if req_id not in self._state: + self._state[req_id] = (len(full_seq), self.trie.root) + + input_len, current_node = self._state[req_id] + + if len(full_seq) > input_len: + last_token = full_seq[-1] + children = getattr(current_node, 'children', None) + if children and last_token in children: + current_node = children[last_token] + self._state[req_id] = (input_len, current_node) + else: + return + + node_id = id(current_node) + bitmask = self._bitmasks.get(node_id) + if bitmask is None: + return + + stream = torch.cuda.ExternalStream(stream_ptr) if stream_ptr else None + with torch.cuda.stream(stream): + if bitmask.device != logits.device: + bitmask = bitmask.to(logits.device) + logits.masked_fill_(~bitmask, NEG_INF) + + def cleanup_request(self, req_id: int) -> None: + self._state.pop(req_id, None) + + +def create_processor( + trie: Any, + mode: str = "cached", + cache_size: int = 10000, + vocab_size: int = 0, + device: str = "cuda" +) -> Any: + """ + Factory function to create the appropriate processor. + + Args: + trie: Trie instance with .root and node.children attributes + mode: "cached" (default), "batched", or "precomputed" + cache_size: Max cached bitmasks for cached/batched modes + vocab_size: Required for precomputed mode + device: GPU device for precomputed mode + + Returns: + Processor instance + """ + if mode == "cached": + return XGRTrieMaskProcessor(trie, cache_size=cache_size) + elif mode == "batched": + return XGRBatchedProcessor(trie, cache_size=cache_size) + elif mode == "precomputed": + if vocab_size <= 0: + raise ValueError("vocab_size required for precomputed mode") + return XGRPrecomputedProcessor(trie, vocab_size, device) + else: + raise ValueError(f"Unknown mode: {mode}") diff --git a/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/model.py b/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/model.py index b4e867a1c91e..f92f0c66fad2 100755 --- a/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/model.py +++ b/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/1/model.py @@ -4,6 +4,7 @@ import sys import time from dataclasses import dataclass +from pathlib import Path from random import randint from threading import Lock, Thread from typing import Any, List @@ -18,6 +19,10 @@ import tensorrt_llm.bindings.executor as trtllm from tensorrt_llm.llmapi.tokenizer import _xgrammar_tokenizer_info +lib_path = Path(__file__).parent / "lib" +if str(lib_path) not in sys.path: + sys.path.insert(0, str(lib_path)) + METRIC_TOTAL_OUTPUT_TOKENS = "total_output_tokens" METRIC_TOTAL_INPUT_TOKENS = "total_input_tokens" import tensorrt_llm.logger as logger @@ -644,18 +649,21 @@ def convert_request(request, guided_decoding_params = get_guided_decoding_params_from_request( request, batch_size, batch_index) - requests.append( - trtllm.Request( - **inputs, - sampling_config=sampling_config, - output_config=output_config, - external_draft_tokens_config=external_draft_tokens_config, - prompt_tuning_config=prompt_tuning_config, - mrope_config=mrope_config, - lora_config=lora_config, - guided_decoding_params=guided_decoding_params, - lookahead_config=request_lookahead_config, - kv_cache_retention_config=kv_cache_retention_config)) + request_kwargs = { + **inputs, + "sampling_config": sampling_config, + "output_config": output_config, + "external_draft_tokens_config": external_draft_tokens_config, + "prompt_tuning_config": prompt_tuning_config, + "mrope_config": mrope_config, + "lora_config": lora_config, + "guided_decoding_params": guided_decoding_params, + "lookahead_config": request_lookahead_config, + "kv_cache_retention_config": kv_cache_retention_config, + } + # Filter out None values - trtllm.Request doesn't accept NoneType for optional args + request_kwargs = {k: v for k, v in request_kwargs.items() if v is not None} + requests.append(trtllm.Request(**request_kwargs)) return requests @@ -1045,6 +1053,22 @@ def get_guided_decoding_config(self, model_config): "guided_decoding_backend", str) tokenizer_dir = get_parameter(model_config, "tokenizer_dir", str) + + # Handle native TRIE backend (requires binary trie file) + if guided_decoding_backend == 'trie': + trie_binary_path = get_parameter(model_config, "trie_binary_path", str) + if not trie_binary_path: + raise ValueError( + "Native TRIE guided decoding requires 'trie_binary_path' parameter." + ) + pb_utils.Logger.log_info( + f"Guided decoding set with native TRIE backend, path: {trie_binary_path}" + ) + return trtllm.GuidedDecodingConfig( + backend=trtllm.GuidedDecodingConfig.GuidedDecodingBackend.TRIE, + trie_path=trie_binary_path + ) + if guided_decoding_backend not in ['xgrammar']: if tokenizer_dir: pb_utils.Logger.log_warn( @@ -1068,6 +1092,84 @@ def get_guided_decoding_config(self, model_config): backend=guided_decoding_backend, **_xgrammar_tokenizer_info(tokenizer)) + def get_logits_post_processor_config(self, model_config): + """Returns LogitsPostProcessorConfig with trie mask processor.""" + trie_path = get_parameter(model_config, "trie_path") + enable_trie_mask = get_parameter(model_config, "enable_trie_mask", bool) + + if not enable_trie_mask and not trie_path: + return None + + # Load all processor variants for benchmarking + try: + from trie_mask import TrieMaskLogitsPostProcessor, XGRTrieMaskProcessor, XGRBatchedProcessor + from trie_loader import load_trie + + trie = load_trie(trie_path) + cached_processor = XGRTrieMaskProcessor(trie) + original_processor = TrieMaskLogitsPostProcessor(trie) + batched_processor = XGRBatchedProcessor(trie) + + config = trtllm.LogitsPostProcessorConfig() + + cpp_processor = self._try_cpp_trie_processor(trie_path) + if cpp_processor is not None: + config.processor_map = { + "trie_mask": cpp_processor, + "trie_mask_cached": cached_processor, + "trie_mask_python": original_processor, + } + pb_utils.Logger.log_info( + f"Using C++ trie_mask + Python variants (cached, python, batched)" + ) + else: + config.processor_map = { + "trie_mask": original_processor, + "trie_mask_cached": cached_processor, + "trie_mask_python": original_processor, + } + pb_utils.Logger.log_info( + f"Using Python trie mask processors (incl. batched)" + ) + + # Register batched processor - invoked with name "batched" + config.processor_batched = batched_processor + + return config + except Exception as e: + pb_utils.Logger.log_warn( + f"Failed to initialize trie mask processor: {e}" + ) + return None + + def _try_cpp_trie_processor(self, trie_path): + """Try to load C++ trie processor. Returns None if unavailable.""" + if not trie_path: + return None + + try: + import trie_mask_cpp + + # Check for binary format first, convert if needed + import os + binary_path = trie_path + if trie_path.endswith('.pkl'): + binary_path = trie_path.replace('.pkl', '.bin') + if not os.path.exists(binary_path): + pb_utils.Logger.log_info( + f"Binary trie not found, using pickle (convert with convert_trie_to_binary.py)" + ) + return None + + return trie_mask_cpp.TrieMaskProcessor(binary_path) + + except ImportError: + pb_utils.Logger.log_info("C++ trie_mask_cpp not available, using Python") + return None + except Exception as e: + pb_utils.Logger.log_warn(f"C++ trie processor failed: {e}") + return None + def get_executor_config(self, model_config): kwargs = { "max_beam_width": @@ -1100,7 +1202,9 @@ def get_executor_config(self, model_config): "extended_runtime_perf_knob_config": self.get_extended_runtime_perf_knob_config(model_config), "guided_decoding_config": - self.get_guided_decoding_config(model_config) + self.get_guided_decoding_config(model_config), + "logits_post_processor_config": + self.get_logits_post_processor_config(model_config), } kwargs = {k: v for k, v in kwargs.items() if v is not None} return trtllm.ExecutorConfig(**kwargs) @@ -1415,6 +1519,16 @@ def execute(self, requests): converted_reqs = convert_request( request, self.exclude_input_from_output, self.decoupled, self.executor_lookahead_config) + + # Set logits_post_processor_name per NVIDIA docs pattern + logits_post_processor_name = get_input_tensor_by_name(request, 'logits_post_processor_name') + if logits_post_processor_name is not None: + lpp_name = logits_post_processor_name.item() + if isinstance(lpp_name, bytes): + lpp_name = lpp_name.decode('utf-8') + for converted_req in converted_reqs: + converted_req.logits_post_processor_name = lpp_name + except Exception as e: response_sender.send( pb_utils.InferenceResponse(error=pb_utils.TritonError( diff --git a/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/config.pbtxt b/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/config.pbtxt index 42a1812b8bf4..80f0eff54afa 100644 --- a/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/config.pbtxt +++ b/triton_backend/all_models/inflight_batcher_llm/tensorrt_llm/config.pbtxt @@ -499,6 +499,12 @@ input [ dims: [ 1 ] optional: true allow_ragged_batch: true + }, + { + name: "logits_post_processor_name" + data_type: TYPE_STRING + dims: [ 1 ] + optional: true } ] output [ @@ -885,3 +891,15 @@ parameters: { string_value: "${xgrammar_tokenizer_info_path}" } } +parameters: { + key: "enable_trie_mask" + value: { + string_value: "${enable_trie_mask}" + } +} +parameters: { + key: "trie_path" + value: { + string_value: "${trie_path}" + } +} diff --git a/triton_backend/inflight_batcher_llm/src/model_instance_state.cc b/triton_backend/inflight_batcher_llm/src/model_instance_state.cc index bb7e5f198668..d93e995dca38 100644 --- a/triton_backend/inflight_batcher_llm/src/model_instance_state.cc +++ b/triton_backend/inflight_batcher_llm/src/model_instance_state.cc @@ -1,4 +1,4 @@ -// Copyright 2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// Copyright 2024-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. // // Redistribution and use in source and binary forms, with or without // modification, are permitted provided that the following conditions @@ -410,6 +410,21 @@ std::optional ModelInstanceState::getGuidedDecod "be ignored."); } + // Native TRIE backend: constrained decoding from a binary trie file. Unlike xgrammar it needs + // only the trie path (no tokenizer info / encoded vocab), so handle it before the xgrammar path. + if (guidedDecodingBackendStr == "trie") + { + std::string triePath = model_state_->GetParameter("guided_decoding_trie_path"); + TLLM_CHECK_WITH_INFO(!triePath.empty() && triePath != "${guided_decoding_trie_path}", + "Native TRIE guided decoding requires a valid 'guided_decoding_trie_path' parameter."); + TLLM_CHECK_WITH_INFO( + std::filesystem::exists(triePath), "Trie binary path at %s does not exist.", triePath.c_str()); + executor::GuidedDecodingConfig trieConfig(executor::GuidedDecodingConfig::GuidedDecodingBackend::kTRIE); + trieConfig.setTriePath(triePath); + TLLM_LOG_INFO("Guided decoding set with native TRIE backend, trie_path: %s", triePath.c_str()); + return trieConfig; + } + if (guidedDecodingBackendStr.empty() || guidedDecodingBackendStr == "${guided_decoding_backend}" || tokenizerInfoPath.empty() || tokenizerInfoPath == "${xgrammar_tokenizer_info_path}") {