chenchenghao/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

use published reasoning parser crate (#17709)

Browse Source
This commit is contained in:
Simo Lin
2026-01-25 09:15:03 -05:00
committed by GitHub
parent fb61164f27
commit 6af22f8dbf
14 changed files with 4 additions and 2464 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
sgl-model-gateway/Cargo.toml
View File

@@ -73,6 +73,8 @@ ulid = "1.2.1"
parking_lot = "0.12.4"
rayon = "1.10"
thiserror = "2.0.12"
regex = "1.10"
memchr = "2.7" # SIMD-optimized byte pattern searching
url = "2.5.4"
@@ -81,6 +83,7 @@ tokio-stream = { version = "0.1", features = ["sync"] }
anyhow = "1.0"
tokenizers = { version = "0.22.0" }
tiktoken-rs = { version = "0.7.0" }
reasoning-parser = "1.0.0"
minijinja = { version = "2.0", features = ["unstable_machinery", "json", "builtins"] }
minijinja-contrib = { version = "2.0", features = ["pycompat"] }
rustls = { version = "0.23", default-features = false, features = ["ring", "std"] }

2
sgl-model-gateway/src/lib.rs
View File

@@ -11,7 +11,7 @@ pub mod multimodal;
pub mod observability;
pub mod policies;
pub mod protocols;
pub mod reasoning_parser;
pub use reasoning_parser;
pub mod routers;
pub mod server;
pub mod service_discovery;

475
sgl-model-gateway/src/reasoning_parser/README.md
View File

@@ -1,475 +0,0 @@
# Reasoning Parser Architecture
## 1. Executive Summary
### High-Level Overview
The reasoning parser layer provides a unified interface for detecting and extracting reasoning content from Large Language Model (LLM) outputs, particularly from models that support Chain-of-Thought (CoT) reasoning with explicit thinking blocks. The architecture follows a trait-based design pattern enabling pluggable parser implementations while maintaining consistent APIs across different model families that use various reasoning token formats.
**Key Components:**
- **Factory Pattern**: Registry-based creation and pooling of model-specific parsers
- **Trait System**: `ReasoningParser` trait for implementation flexibility
- **Parser Pooling**: Efficient reuse of parser instances across concurrent requests
- **Streaming Support**: Incremental parsing with partial token buffering
- **Model Detection**: Pattern-based matching for automatic parser selection
- **State Management**: Stateful parsing for streaming scenarios with buffer management
- **Thread Safety**: Arc<Mutex> based sharing for high-concurrency environments
- **Extensibility**: Easy addition of new model-specific parsers
**Data Flow:**
1. Request → Factory (model detection) → Pooled Parser Retrieval
2. One-Shot: Text → Parser → ParserResult (normal + reasoning text)
3. Streaming: Chunks → Parser (stateful) → Incremental ParserResult
4. Buffer Management: Partial Tokens → Buffer → Complete Token Detection
5. Reset: Parser State → Clear Buffers → Ready for Reuse
### Architecture Highlights
- **Model-Specific Parsers**: DeepSeek-R1, Qwen3, Kimi, GLM45, GLM47, Step3 variants
- **Parser Pooling**: Singleton instances per model type for memory efficiency
- **High Concurrency**: Mutex-protected parsers handle 1000+ req/sec
- **Buffer Overflow Protection**: Configurable max buffer size (default 64KB)
- **Partial Token Detection**: Intelligent buffering for incomplete delimiters
- **Passthrough Mode**: Graceful fallback for unknown models
- **Zero-Copy Where Possible**: Efficient string handling in hot paths
## 2. Mermaid Diagrams
### Component Flow Diagram
```mermaid
graph TB
subgraph Input
R[Request] --> MID[Model ID]
end
subgraph Factory Layer
MID --> PF[ReasoningParserFactory]
PF --> REG[ParserRegistry]
REG --> PM[Pattern Matching]
PM --> PP[Parser Pool]
end
subgraph Parser Pool
PP --> DS[DeepSeek-R1]
PP --> QW[Qwen3]
PP --> QWT[Qwen3-Thinking]
PP --> KM[Kimi]
PP --> GL[GLM45/GLM47]
PP --> S3[Step3]
PP --> PT[Passthrough]
end
subgraph Parser Instance
DS --> BP[BaseReasoningParser]
QW --> BP
KM --> BP
GL --> BP
S3 --> BP
end
subgraph Processing
BP --> DAP[detect_and_parse]
BP --> PSI[parse_streaming]
BP --> RST[reset]
end
subgraph State Management
BP --> BUF[Buffer]
BP --> IR[in_reasoning flag]
BP --> STS[stripped_think_start]
end
subgraph Output
DAP --> PR[ParserResult]
PSI --> PR
PR --> NT[normal_text]
PR --> RT[reasoning_text]
end
```
### Sequence Flow Diagram
```mermaid
sequenceDiagram
participant C as Client
participant F as ReasoningParserFactory
participant R as Registry
participant P as Parser Pool
participant BP as BaseParser
participant PR as ParserResult
C->>F: get_pooled("deepseek-r1-model")
F->>R: find_pooled_parser_for_model()
R->>R: pattern_match("deepseek-r1")
R->>P: get_pooled_parser("deepseek_r1")
alt Parser exists in pool
P-->>F: Arc<Mutex<Parser>>
else Create new parser
P->>BP: new DeepSeekR1Parser()
P->>P: insert into pool
P-->>F: Arc<Mutex<Parser>>
end
F-->>C: PooledParser
C->>BP: lock().parse_reasoning_streaming_incremental()
loop streaming chunks
C->>BP: parse_reasoning_streaming_incremental(chunk)
BP->>BP: buffer.push_str(chunk)
BP->>BP: check partial tokens
alt Complete token found
BP->>PR: create result
BP->>BP: clear buffer
BP-->>C: ParserResult
else Partial token
BP->>BP: keep buffering
BP-->>C: ParserResult::default()
end
end
C->>BP: reset()
BP->>BP: clear buffers & flags
C->>BP: unlock()
```
### Class/Type Diagram
```mermaid
classDiagram
class ReasoningParser {
<<trait>>
+detect_and_parse_reasoning(&mut self, text: &str) Result~ParserResult~
+parse_reasoning_streaming_incremental(&mut self, text: &str) Result~ParserResult~
+reset(&mut self)
+model_type(&self) &str
}
class ParserResult {
+normal_text: String
+reasoning_text: String
+new(normal: String, reasoning: String) Self
+normal(text: String) Self
+reasoning(text: String) Self
+is_empty() bool
}
class ParserConfig {
+think_start_token: String
+think_end_token: String
+stream_reasoning: bool
+max_buffer_size: usize
+initial_in_reasoning: bool
+default() Self
}
class BaseReasoningParser {
-config: ParserConfig
-in_reasoning: bool
-buffer: String
-stripped_think_start: bool
-model_type: String
+new(config: ParserConfig) Self
+with_model_type(model: String) Self
-is_partial_token(&self, text: &str) bool
}
class DeepSeekR1Parser {
-base: BaseReasoningParser
+new() Self
}
class Qwen3Parser {
-base: BaseReasoningParser
+new() Self
}
class QwenThinkingParser {
-base: BaseReasoningParser
+new() Self
}
class KimiParser {
-base: BaseReasoningParser
+new() Self
}
class Glm45Parser {
-base: BaseReasoningParser
+new() Self
}
class Step3Parser {
-base: BaseReasoningParser
+new() Self
}
class ReasoningParserFactory {
-registry: ParserRegistry
+new() Self
+get_pooled(model_id: &str) PooledParser
+create(model_id: &str) Result~Box~dyn ReasoningParser~~
+clear_pool()
}
class ParserRegistry {
-creators: Arc~RwLock~HashMap~~
-pool: Arc~RwLock~HashMap~~
-patterns: Arc~RwLock~Vec~~
+register_parser(name: &str, creator: F)
+register_pattern(pattern: &str, parser_name: &str)
+get_pooled_parser(name: &str) Option~PooledParser~
+find_pooled_parser_for_model(model: &str) Option~PooledParser~
}
ReasoningParser <|.. BaseReasoningParser
ReasoningParser <|.. DeepSeekR1Parser
ReasoningParser <|.. Qwen3Parser
ReasoningParser <|.. QwenThinkingParser
ReasoningParser <|.. KimiParser
ReasoningParser <|.. Glm45Parser
ReasoningParser <|.. Step3Parser
DeepSeekR1Parser o-- BaseReasoningParser
Qwen3Parser o-- BaseReasoningParser
QwenThinkingParser o-- BaseReasoningParser
KimiParser o-- BaseReasoningParser
Glm45Parser o-- BaseReasoningParser
Step3Parser o-- BaseReasoningParser
BaseReasoningParser o-- ParserConfig
ReasoningParserFactory o-- ParserRegistry
ParserRegistry o-- ReasoningParser
```
## 3. Module-by-Module Deep Dive
### 3.1 mod.rs (Main Module)
**Key Responsibilities:**
- Module organization and public API surface
- Re-exports for convenient access to core types
- Separation of concerns across submodules
**Module Structure:**
- `factory`: Parser creation and pooling logic
- `parsers`: Concrete parser implementations
- `traits`: Core trait definitions and types
### 3.2 traits.rs (Trait Definitions)
**ParserResult Methods**:
- `new()`: Create with both normal and reasoning text
- `normal()`: Create with only normal text (convenience)
- `reasoning()`: Create with only reasoning text (convenience)
- `is_empty()`: Check if result contains any text
**ReasoningParser Trait**:
- **`detect_and_parse_reasoning`**: One-shot parsing for complete text
- **`parse_reasoning_streaming_incremental`**: Stateful streaming parser
- **`reset`**: Clear state for parser reuse
- **`model_type`**: Identify parser variant for debugging
**ParserConfig Defaults**:
- Default tokens: `<think>` and `</think>`
- Stream reasoning: true (immediate output)
- Max buffer: 65536 bytes (64KB)
- Initial state: false (explicit reasoning blocks)
### 3.3 factory.rs (Parser Creation & Pooling)
**ParserRegistry Methods**:
1. **`register_parser`**:
- Register creator function for parser type
- Lazy instantiation when requested
- Thread-safe registration
2. **`register_pattern`**:
- Map model ID patterns to parser names
- First-match-wins ordering
- Case-insensitive matching
3. **`get_pooled_parser`**:
- Check pool for existing instance
- Create and pool if not present
- Return Arc<Mutex> for sharing
4. **`find_pooled_parser_for_model`**:
- Pattern match against model ID
- Delegate to get_pooled_parser
- Case-insensitive comparison
**ReasoningParserFactory Methods**:
1. **`new()`**:
- Register all built-in parsers
- Setup model pattern mappings
- Initialize empty pool
2. **`get_pooled`**:
- Primary API for getting parsers
- Automatic passthrough fallback
- Guaranteed non-null return
3. **`create`**:
- Create fresh parser instance
- No pooling (for testing/isolation)
- Returns Result for error handling
**Registered Parsers**:
- `base`: Generic configurable parser
- `deepseek_r1`: DeepSeek-R1 (initial_in_reasoning=true)
- `qwen3`: Qwen3 base model (initial_in_reasoning=false)
- `qwen3_thinking`: Qwen3 thinking variant (initial_in_reasoning=true)
- `kimi`: Kimi with Unicode tokens
- `glm45`: GLM-4.5 / GLM-4.6 / GLM-4.7 parser
- `step3`: Step3 parser
- `passthrough`: No-op fallback parser
**Model Pattern Mappings**:
```
"deepseek-r1" → "deepseek_r1"
"qwen3-thinking" → "qwen3_thinking"
"qwen-thinking" → "qwen3_thinking"
"qwen3" → "qwen3"
"qwen" → "qwen3"
"glm45" → "glm45"
"glm47" → "glm45"
"kimi" → "kimi"
"step3" → "step3"
```
### 3.4 parsers/base.rs (Base Implementation)
**Key Methods:**
**`detect_and_parse_reasoning`**:
```
Algorithm:
1. Check buffer overflow protection
2. Detect reasoning presence (in_reasoning OR contains start_token)
3. If no reasoning → return as normal text
4. Remove start token and trim
5. If no end token → assume truncated reasoning
6. Split on end token
7. Extract reasoning and normal portions
```
**`parse_reasoning_streaming_incremental`**:
```
Algorithm:
1. Check buffer capacity
2. Append text to buffer
3. Check if buffer is partial token prefix
4. If partial → buffer and return empty
5. Strip start token if present
6. Find end token position
7. Handle based on state:
- In reasoning + end found → split and return both
- In reasoning + streaming → return accumulated reasoning
- Not in reasoning → return as normal text
- In reasoning + no end → continue buffering
```
**Critical Features:**
1. **Partial Token Detection**:
- Prevents premature token matching
- Buffers incomplete delimiters
- Essential for streaming correctness
2. **Buffer Management**:
- Overflow protection
- Accumulation for partial content
- Clear on complete token detection
3. **State Tracking**:
- `in_reasoning`: Current parsing state
- `stripped_think_start`: Prevent double processing
- `buffer`: Accumulated partial content
## 4. Extensibility Guide
### Adding a New Parser
**Step 1: Create Parser Implementation**
```rust
// src/reasoning_parser/parsers/mymodel.rs
use crate::reasoning_parser::parsers::BaseReasoningParser;
use crate::reasoning_parser::traits::{ParserConfig, ReasoningParser};
pub struct MyModelParser {
base: BaseReasoningParser,
}
impl MyModelParser {
pub fn new() -> Self {
let config = ParserConfig {
think_start_token: "<reasoning>".to_string(),
think_end_token: "</reasoning>".to_string(),
stream_reasoning: true,
max_buffer_size: 65536,
initial_in_reasoning: false, // or true for implicit
};
Self {
base: BaseReasoningParser::new(config)
.with_model_type("mymodel".to_string()),
}
}
}
impl ReasoningParser for MyModelParser {
// Delegate to base or implement custom logic
fn detect_and_parse_reasoning(&mut self, text: &str)
-> Result<ParserResult, ParseError> {
self.base.detect_and_parse_reasoning(text)
}
// ... other trait methods
}
```
**Step 2: Register in Factory**
```rust
// In factory.rs ReasoningParserFactory::new()
registry.register_parser("mymodel", || {
Box::new(MyModelParser::new())
});
// Register patterns
registry.register_pattern("my-model", "mymodel");
registry.register_pattern("mymodel", "mymodel");
```
**Step 3: Export from Module**
```rust
// In parsers/mod.rs
pub use self::mymodel::MyModelParser;
// In reasoning_parser/mod.rs
pub use parsers::MyModelParser;
```
### Custom Parsing Logic
For parsers requiring custom logic beyond configuration:
```rust
impl ReasoningParser for CustomParser {
fn parse_reasoning_streaming_incremental(&mut self, text: &str)
-> Result<ParserResult, ParseError> {
// Custom state machine
// Custom token detection
// Custom buffering strategy
// Return appropriate ParserResult
}
}
```

599
sgl-model-gateway/src/reasoning_parser/factory.rs
View File

@@ -1,599 +0,0 @@
// Factory and registry for creating model-specific reasoning parsers.
// Now with parser pooling support for efficient reuse across requests.
use std::{
collections::HashMap,
sync::{Arc, RwLock},
};
use tokio::sync::Mutex;
use crate::reasoning_parser::{
parsers::{
BaseReasoningParser, DeepSeekR1Parser, Glm45Parser, KimiParser, MiniMaxParser, Qwen3Parser,
QwenThinkingParser, Step3Parser,
},
traits::{ParseError, ParserConfig, ReasoningParser},
};
/// Type alias for pooled parser instances.
/// Uses tokio::Mutex to avoid blocking the async executor.
pub type PooledParser = Arc<Mutex<Box<dyn ReasoningParser>>>;
/// Type alias for parser creator functions.
type ParserCreator = Arc<dyn Fn() -> Box<dyn ReasoningParser> + Send + Sync>;
/// Registry for model-specific parsers with pooling support.
#[derive(Clone)]
pub struct ParserRegistry {
/// Creator functions for parsers (used when pool is empty)
creators: Arc<RwLock<HashMap<String, ParserCreator>>>,
/// Pooled parser instances for reuse
pool: Arc<RwLock<HashMap<String, PooledParser>>>,
/// Model pattern to parser name mappings
patterns: Arc<RwLock<Vec<(String, String)>>>, // (pattern, parser_name)
}
impl ParserRegistry {
/// Create a new empty registry.
pub fn new() -> Self {
Self {
creators: Arc::new(RwLock::new(HashMap::new())),
pool: Arc::new(RwLock::new(HashMap::new())),
patterns: Arc::new(RwLock::new(Vec::new())),
}
}
/// Register a parser creator for a given parser type.
pub fn register_parser<F>(&self, name: &str, creator: F)
where
F: Fn() -> Box<dyn ReasoningParser> + Send + Sync + 'static,
{
let mut creators = self.creators.write().unwrap();
creators.insert(name.to_string(), Arc::new(creator));
}
/// Register a model pattern to parser mapping.
/// Patterns are checked in order, first match wins.
pub fn register_pattern(&self, pattern: &str, parser_name: &str) {
let mut patterns = self.patterns.write().unwrap();
patterns.push((pattern.to_string(), parser_name.to_string()));
}
/// Get a pooled parser by exact name.
/// Returns a shared parser instance from the pool, creating one if needed.
pub fn get_pooled_parser(&self, name: &str) -> Option<PooledParser> {
// First check if we have a pooled instance
{
let pool = self.pool.read().unwrap();
if let Some(parser) = pool.get(name) {
return Some(Arc::clone(parser));
}
}
// If not in pool, create one and add to pool
let creators = self.creators.read().unwrap();
if let Some(creator) = creators.get(name) {
let parser = Arc::new(Mutex::new(creator()));
// Add to pool for future use
let mut pool = self.pool.write().unwrap();
pool.insert(name.to_string(), Arc::clone(&parser));
Some(parser)
} else {
None
}
}
/// Check if a parser with the given name is registered.
pub fn has_parser(&self, name: &str) -> bool {
let creators = self.creators.read().unwrap();
creators.contains_key(name)
}
/// Create a fresh parser instance by exact name (not pooled).
/// Returns a new parser instance for each call - useful for streaming where state isolation is needed.
pub fn create_parser(&self, name: &str) -> Option<Box<dyn ReasoningParser>> {
let creators = self.creators.read().unwrap();
creators.get(name).map(|creator| creator())
}
/// Find a pooled parser for a given model ID by pattern matching.
pub fn find_pooled_parser_for_model(&self, model_id: &str) -> Option<PooledParser> {
let patterns = self.patterns.read().unwrap();
let model_lower = model_id.to_lowercase();
for (pattern, parser_name) in patterns.iter() {
if model_lower.contains(&pattern.to_lowercase()) {
return self.get_pooled_parser(parser_name);
}
}
None
}
/// Check if a parser can be created for a specific model without actually creating it.
/// Returns true if a parser is available (registered) for this model.
pub fn has_parser_for_model(&self, model_id: &str) -> bool {
let patterns = self.patterns.read().unwrap();
let model_lower = model_id.to_lowercase();
for (pattern, parser_name) in patterns.iter() {
if model_lower.contains(&pattern.to_lowercase()) {
let creators = self.creators.read().unwrap();
return creators.contains_key(parser_name);
}
}
false
}
/// Create a fresh parser instance for a given model ID by pattern matching (not pooled).
/// Returns a new parser instance for each call - useful for streaming where state isolation is needed.
pub fn create_for_model(&self, model_id: &str) -> Option<Box<dyn ReasoningParser>> {
let patterns = self.patterns.read().unwrap();
let model_lower = model_id.to_lowercase();
for (pattern, parser_name) in patterns.iter() {
if model_lower.contains(&pattern.to_lowercase()) {
return self.create_parser(parser_name);
}
}
None
}
/// Clear the parser pool, forcing new instances to be created.
/// Useful for testing or when parsers need to be reset globally.
pub fn clear_pool(&self) {
let mut pool = self.pool.write().unwrap();
pool.clear();
}
}
impl Default for ParserRegistry {
fn default() -> Self {
Self::new()
}
}
/// Factory for creating reasoning parsers based on model type.
#[derive(Clone)]
pub struct ParserFactory {
registry: ParserRegistry,
}
impl ParserFactory {
/// Create a new factory with default parsers registered.
pub fn new() -> Self {
let registry = ParserRegistry::new();
// Register base parser
registry.register_parser("base", || {
Box::new(BaseReasoningParser::new(ParserConfig::default()))
});
// Register DeepSeek-R1 parser (starts with in_reasoning=true)
registry.register_parser("deepseek_r1", || Box::new(DeepSeekR1Parser::new()));
// Register Qwen3 parser (starts with in_reasoning=false)
registry.register_parser("qwen3", || Box::new(Qwen3Parser::new()));
// Register Qwen3-thinking parser (starts with in_reasoning=true)
registry.register_parser("qwen3_thinking", || Box::new(QwenThinkingParser::new()));
// Register Kimi parser with Unicode tokens (starts with in_reasoning=false)
registry.register_parser("kimi", || Box::new(KimiParser::new()));
// Register GLM45 parser (same format as Qwen3 but separate for debugging)
registry.register_parser("glm45", || Box::new(Glm45Parser::new()));
// Register Step3 parser (same format as DeepSeek-R1 but separate for debugging)
registry.register_parser("step3", || Box::new(Step3Parser::new()));
// Register MiniMax parser (appends <think> token at the beginning)
registry.register_parser("minimax", || Box::new(MiniMaxParser::new()));
// Register model patterns
registry.register_pattern("deepseek-r1", "deepseek_r1");
registry.register_pattern("qwen3-thinking", "qwen3_thinking");
registry.register_pattern("qwen-thinking", "qwen3_thinking");
registry.register_pattern("qwen3", "qwen3");
registry.register_pattern("qwen", "qwen3");
registry.register_pattern("glm45", "glm45");
registry.register_pattern("glm47", "glm45"); // glm47 uses same reasoning format as glm45
registry.register_pattern("kimi", "kimi");
registry.register_pattern("step3", "step3");
registry.register_pattern("minimax", "minimax");
registry.register_pattern("minimax-m2", "minimax");
registry.register_pattern("mm-m2", "minimax");
// Nano V3 uses same format as Qwen3 (requires explicit <think> token)
registry.register_pattern("nemotron-nano", "qwen3");
registry.register_pattern("nano-v3", "qwen3");
Self { registry }
}
/// Get a pooled parser for the given model ID.
/// Returns a shared instance that can be used concurrently.
/// Falls back to a passthrough parser if model is not recognized.
pub fn get_pooled(&self, model_id: &str) -> PooledParser {
// First try to find by pattern
if let Some(parser) = self.registry.find_pooled_parser_for_model(model_id) {
return parser;
}
// Fall back to no-op parser (get or create passthrough in pool)
self.registry
.get_pooled_parser("passthrough")
.unwrap_or_else(|| {
// Register passthrough if not already registered
self.registry.register_parser("passthrough", || {
let config = ParserConfig {
think_start_token: "".to_string(),
think_end_token: "".to_string(),
stream_reasoning: true,
max_buffer_size: 65536,
initial_in_reasoning: false,
};
Box::new(
BaseReasoningParser::new(config).with_model_type("passthrough".to_string()),
)
});
self.registry.get_pooled_parser("passthrough").unwrap()
})
}
/// Create a new parser instance for the given model ID.
/// Returns a fresh instance (not pooled).
/// Use this when you need an isolated parser instance.
pub fn create(&self, model_id: &str) -> Result<Box<dyn ReasoningParser>, ParseError> {
// First try to find by pattern
if let Some(parser) = self.registry.create_for_model(model_id) {
return Ok(parser);
}
// Fall back to no-op parser (base parser without reasoning detection)
let config = ParserConfig {
think_start_token: "".to_string(),
think_end_token: "".to_string(),
stream_reasoning: true,
max_buffer_size: 65536,
initial_in_reasoning: false,
};
Ok(Box::new(
BaseReasoningParser::new(config).with_model_type("passthrough".to_string()),
))
}
/// Get the internal registry for custom registration.
pub fn registry(&self) -> &ParserRegistry {
&self.registry
}
/// Clear the parser pool.
/// Useful for testing or when parsers need to be reset globally.
pub fn clear_pool(&self) {
self.registry.clear_pool();
}
}
impl Default for ParserFactory {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_factory_creates_deepseek_r1() {
let factory = ParserFactory::new();
let parser = factory.create("deepseek-r1-distill").unwrap();
assert_eq!(parser.model_type(), "deepseek_r1");
}
#[test]
fn test_factory_creates_qwen3() {
let factory = ParserFactory::new();
let parser = factory.create("qwen3-7b").unwrap();
assert_eq!(parser.model_type(), "qwen3");
}
#[test]
fn test_factory_creates_kimi() {
let factory = ParserFactory::new();
let parser = factory.create("kimi-chat").unwrap();
assert_eq!(parser.model_type(), "kimi");
}
#[test]
fn test_factory_fallback_to_passthrough() {
let factory = ParserFactory::new();
let parser = factory.create("unknown-model").unwrap();
assert_eq!(parser.model_type(), "passthrough");
}
#[test]
fn test_case_insensitive_matching() {
let factory = ParserFactory::new();
let parser1 = factory.create("DeepSeek-R1").unwrap();
let parser2 = factory.create("QWEN3").unwrap();
let parser3 = factory.create("Kimi").unwrap();
assert_eq!(parser1.model_type(), "deepseek_r1");
assert_eq!(parser2.model_type(), "qwen3");
assert_eq!(parser3.model_type(), "kimi");
}
#[test]
fn test_step3_model() {
let factory = ParserFactory::new();
let step3 = factory.create("step3-model").unwrap();
assert_eq!(step3.model_type(), "step3");
}
#[test]
fn test_glm45_model() {
let factory = ParserFactory::new();
let glm45 = factory.create("glm45-v2").unwrap();
assert_eq!(glm45.model_type(), "glm45");
}
#[test]
fn test_minimax_model() {
let factory = ParserFactory::new();
let minimax = factory.create("minimax-m2").unwrap();
assert_eq!(minimax.model_type(), "minimax");
// Also test alternate patterns
let mm = factory.create("mm-m2-chat").unwrap();
assert_eq!(mm.model_type(), "minimax");
}
#[tokio::test]
async fn test_pooled_parser_reuse() {
let factory = ParserFactory::new();
// Get the same parser twice - should be the same instance
let parser1 = factory.get_pooled("deepseek-r1");
let parser2 = factory.get_pooled("deepseek-r1");
// Both should point to the same Arc
assert!(Arc::ptr_eq(&parser1, &parser2));
// Different models should get different parsers
let parser3 = factory.get_pooled("qwen3");
assert!(!Arc::ptr_eq(&parser1, &parser3));
}
#[tokio::test]
async fn test_pooled_parser_concurrent_access() {
let factory = ParserFactory::new();
let parser = factory.get_pooled("deepseek-r1");
// Spawn multiple async tasks that use the same parser
let mut handles = vec![];
for i in 0..3 {
let parser_clone = Arc::clone(&parser);
let handle = tokio::spawn(async move {
let mut parser = parser_clone.lock().await;
let input = format!("thread {} reasoning</think>answer", i);
let result = parser.detect_and_parse_reasoning(&input).unwrap();
assert_eq!(result.normal_text, "answer");
assert!(result.reasoning_text.contains("reasoning"));
});
handles.push(handle);
}
// Wait for all tasks to complete
for handle in handles {
handle.await.unwrap();
}
}
#[tokio::test]
async fn test_pool_clearing() {
let factory = ParserFactory::new();
// Get a pooled parser
let parser1 = factory.get_pooled("deepseek-r1");
// Clear the pool
factory.clear_pool();
// Get another parser - should be a new instance
let parser2 = factory.get_pooled("deepseek-r1");
// They should be different instances (different Arc pointers)
assert!(!Arc::ptr_eq(&parser1, &parser2));
}
#[tokio::test]
async fn test_passthrough_parser_pooling() {
let factory = ParserFactory::new();
// Unknown models should get passthrough parser
let parser1 = factory.get_pooled("unknown-model-1");
let parser2 = factory.get_pooled("unknown-model-2");
// Both should use the same passthrough parser instance
assert!(Arc::ptr_eq(&parser1, &parser2));
let parser = parser1.lock().await;
assert_eq!(parser.model_type(), "passthrough");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn test_high_concurrency_parser_access() {
use std::{
sync::atomic::{AtomicUsize, Ordering},
time::Instant,
};
let factory = ParserFactory::new();
let num_tasks = 100;
let requests_per_task = 50;
let models = vec!["deepseek-r1", "qwen3", "kimi", "qwen3-thinking"];
// Track successful operations
let success_count = Arc::new(AtomicUsize::new(0));
let error_count = Arc::new(AtomicUsize::new(0));
let start = Instant::now();
let mut handles = vec![];
for task_id in 0..num_tasks {
let factory = factory.clone();
let models = models.clone();
let success_count = Arc::clone(&success_count);
let error_count = Arc::clone(&error_count);
let handle = tokio::spawn(async move {
for request_id in 0..requests_per_task {
// Rotate through different models
let model = &models[(task_id + request_id) % models.len()];
let parser = factory.get_pooled(model);
// Use async lock - tokio::Mutex doesn't poison
let mut p = parser.lock().await;
// Simulate realistic parsing work with substantial text
// Typical reasoning can be 500-5000 tokens
let reasoning_text = format!(
"Task {} is processing request {}. Let me think through this step by step. \
First, I need to understand the problem. The problem involves analyzing data \
and making calculations. Let me break this down: \n\
1. Initial analysis shows that we have multiple variables to consider. \
2. The data suggests a pattern that needs further investigation. \
3. Computing the values: {} * {} = {}. \
4. Cross-referencing with previous results indicates consistency. \
5. The mathematical proof follows from the axioms... \
6. Considering edge cases and boundary conditions... \
7. Validating against known constraints... \
8. The conclusion follows logically from premises A, B, and C. \
This reasoning chain demonstrates the validity of our approach.",
task_id, request_id, task_id, request_id, task_id * request_id
);
let answer_text = format!(
"Based on my analysis, the answer for task {} request {} is: \
The solution involves multiple steps as outlined in the reasoning. \
The final result is {} with confidence level high. \
This conclusion is supported by rigorous mathematical analysis \
and has been validated against multiple test cases. \
The implementation should handle edge cases appropriately.",
task_id,
request_id,
task_id * request_id
);
let input = format!("<think>{}</think>{}", reasoning_text, answer_text);
match p.detect_and_parse_reasoning(&input) {
Ok(result) => {
// Note: Some parsers with stream_reasoning=true won't accumulate reasoning text
assert!(result.normal_text.contains(&format!("task {}", task_id)));
// For parsers that accumulate reasoning (stream_reasoning=false)
// the reasoning_text should be populated
if !result.reasoning_text.is_empty() {
assert!(result
.reasoning_text
.contains(&format!("Task {}", task_id)));
assert!(result.reasoning_text.len() > 500); // Ensure substantial reasoning
}
// Normal text should always be present
assert!(result.normal_text.len() > 100); // Ensure substantial answer
success_count.fetch_add(1, Ordering::Relaxed);
}
Err(e) => {
eprintln!("Parse error: {:?}", e);
error_count.fetch_add(1, Ordering::Relaxed);
}
}
// Explicitly drop the lock to release it quickly
drop(p);
}
});
handles.push(handle);
}
// Wait for all tasks
for handle in handles {
handle.await.unwrap();
}
let duration = start.elapsed();
let total_requests = num_tasks * requests_per_task;
let successes = success_count.load(Ordering::Relaxed);
let errors = error_count.load(Ordering::Relaxed);
// Print stats for debugging
println!(
"High concurrency test: {} tasks, {} requests each",
num_tasks, requests_per_task
);
println!(
"Completed in {:?}, {} successes, {} errors",
duration, successes, errors
);
println!(
"Throughput: {:.0} requests/sec",
(total_requests as f64) / duration.as_secs_f64()
);
// All requests should succeed
assert_eq!(successes, total_requests);
assert_eq!(errors, 0);
// Performance check: should handle at least 1000 req/sec
let throughput = (total_requests as f64) / duration.as_secs_f64();
assert!(
throughput > 1000.0,
"Throughput too low: {:.0} req/sec",
throughput
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn test_concurrent_pool_modifications() {
let factory = ParserFactory::new();
let mut handles = vec![];
// Task 1: Continuously get parsers
let factory1 = factory.clone();
handles.push(tokio::spawn(async move {
for _ in 0..100 {
let _parser = factory1.get_pooled("deepseek-r1");
}
}));
// Task 2: Continuously clear pool
let factory2 = factory.clone();
handles.push(tokio::spawn(async move {
for _ in 0..10 {
factory2.clear_pool();
tokio::time::sleep(tokio::time::Duration::from_micros(100)).await;
}
}));
// Task 3: Get different parsers
let factory3 = factory.clone();
handles.push(tokio::spawn(async move {
for i in 0..100 {
let models = ["qwen3", "kimi", "unknown"];
let _parser = factory3.get_pooled(models[i % 3]);
}
}));
// Wait for all tasks - should not deadlock or panic
for handle in handles {
handle.await.unwrap();
}
}
}

10
sgl-model-gateway/src/reasoning_parser/mod.rs
View File

@@ -1,10 +0,0 @@
pub mod factory;
pub mod parsers;
pub mod traits;
pub use factory::{ParserFactory, ParserRegistry, PooledParser};
pub use parsers::{
BaseReasoningParser, DeepSeekR1Parser, Glm45Parser, KimiParser, MiniMaxParser, Qwen3Parser,
QwenThinkingParser, Step3Parser,
};
pub use traits::{ParseError, ParserConfig, ParserResult, ReasoningParser};

362
sgl-model-gateway/src/reasoning_parser/parsers/base.rs
View File

@@ -1,362 +0,0 @@
// Base implementation of reasoning parser that handles common logic
// for detecting and extracting reasoning blocks from text.
use crate::reasoning_parser::traits::{ParseError, ParserConfig, ParserResult, ReasoningParser};
/// Base reasoning parser implementation.
///
/// This parser handles the common logic for detecting reasoning blocks
/// delimited by start and end tokens (e.g., <think> and </think>).
#[derive(Debug, Clone)]
pub struct BaseReasoningParser {
config: ParserConfig,
in_reasoning: bool,
buffer: String,
stripped_think_start: bool,
model_type: String,
}
impl BaseReasoningParser {
/// Create a new BaseReasoningParser with the given configuration.
pub fn new(config: ParserConfig) -> Self {
let in_reasoning = config.initial_in_reasoning;
Self {
config,
in_reasoning,
buffer: String::new(),
stripped_think_start: false,
model_type: "base".to_string(),
}
}
/// Create with custom model type identifier.
pub fn with_model_type(mut self, model_type: String) -> Self {
self.model_type = model_type;
self
}
/// Check if the current buffer is a prefix of one of the tokens.
fn is_partial_token(&self, text: &str) -> bool {
(self.config.think_start_token.starts_with(text) && self.config.think_start_token != text)
|| (self.config.think_end_token.starts_with(text)
&& self.config.think_end_token != text)
}
}
impl ReasoningParser for BaseReasoningParser {
fn detect_and_parse_reasoning(&mut self, text: &str) -> Result<ParserResult, ParseError> {
// Check input size against buffer limit
if text.len() > self.config.max_buffer_size {
return Err(ParseError::BufferOverflow(text.len()));
}
let in_reasoning = self.in_reasoning || text.contains(&self.config.think_start_token);
if !in_reasoning {
return Ok(ParserResult::normal(text.to_string()));
}
// The text is considered to be in a reasoning block.
let processed_text = text
.replace(&self.config.think_start_token, "")
.trim()
.to_string();
if !processed_text.contains(&self.config.think_end_token) {
// Assume reasoning was truncated before end token
return Ok(ParserResult::reasoning(processed_text));
}
// Extract reasoning content
let splits: Vec<&str> = processed_text
.splitn(2, &self.config.think_end_token)
.collect();
let reasoning_text = splits.first().unwrap_or(&"").to_string();
let normal_text = splits
.get(1)
.map(|s| s.trim().to_string())
.unwrap_or_default();
Ok(ParserResult::new(normal_text, reasoning_text))
}
fn parse_reasoning_streaming_incremental(
&mut self,
text: &str,
) -> Result<ParserResult, ParseError> {
// Check if adding this text would exceed buffer limit
if self.buffer.len() + text.len() > self.config.max_buffer_size {
return Err(ParseError::BufferOverflow(self.buffer.len() + text.len()));
}
// Incrementally parse the streaming text
self.buffer.push_str(text);
let mut current_text = self.buffer.clone();
// If the current text is a prefix of a token, keep buffering
if self.is_partial_token(&current_text) {
return Ok(ParserResult::default());
}
// Strip start token if present
if !self.stripped_think_start && current_text.contains(&self.config.think_start_token) {
current_text = current_text.replace(&self.config.think_start_token, "");
self.buffer = current_text.clone();
self.stripped_think_start = true;
self.in_reasoning = true;
}
// Handle end of reasoning block
let think_end_idx = if self.in_reasoning {
current_text
.find(&self.config.think_end_token)
.unwrap_or(current_text.len())
} else {
current_text.len()
};
if self.in_reasoning && think_end_idx < current_text.len() {
let reasoning_text = &current_text[..think_end_idx];
self.buffer.clear();
self.in_reasoning = false;
let start_idx = think_end_idx + self.config.think_end_token.len();
let normal_text = if start_idx < current_text.len() {
&current_text[start_idx..]
} else {
""
};
return Ok(ParserResult::new(
normal_text.to_string(),
reasoning_text.trim().to_string(),
));
}
// Continue with reasoning content
if self.in_reasoning && self.config.stream_reasoning {
// Stream the content immediately
let reasoning_text = current_text;
self.buffer.clear();
Ok(ParserResult::reasoning(reasoning_text))
} else if !self.in_reasoning {
// If we're not in a reasoning block, return as normal text
// CRITICAL FIX: Return current_text (with buffer) not just text
// This prevents buffer loss when partial tokens are followed by normal text
let normal_text = current_text;
self.buffer.clear();
Ok(ParserResult::normal(normal_text))
} else {
// If we are in a reasoning block but no end token is found, buffer it
Ok(ParserResult::default())
}
}
fn reset(&mut self) {
self.in_reasoning = self.config.initial_in_reasoning;
self.buffer.clear();
self.stripped_think_start = false;
}
fn model_type(&self) -> &str {
&self.model_type
}
fn is_in_reasoning(&self) -> bool {
self.in_reasoning
}
}
#[cfg(test)]
mod tests {
use super::*;
fn create_test_parser(
initial_in_reasoning: bool,
stream_reasoning: bool,
) -> BaseReasoningParser {
let config = ParserConfig {
think_start_token: "<think>".to_string(),
think_end_token: "</think>".to_string(),
stream_reasoning,
max_buffer_size: 65536,
initial_in_reasoning,
};
BaseReasoningParser::new(config)
}
#[test]
fn test_detect_and_parse_reasoning() {
let mut parser = create_test_parser(false, true);
let result = parser
.detect_and_parse_reasoning("<think>with reasoning</think> and more text.")
.unwrap();
assert_eq!(result.normal_text, "and more text.");
assert_eq!(result.reasoning_text, "with reasoning");
}
#[test]
fn test_detect_and_parse_no_reasoning() {
let mut parser = create_test_parser(false, true);
let result = parser
.detect_and_parse_reasoning("This is a test without reasoning.")
.unwrap();
assert_eq!(result.normal_text, "This is a test without reasoning.");
assert_eq!(result.reasoning_text, "");
}
#[test]
fn test_detect_and_parse_truncated_reasoning() {
let mut parser = create_test_parser(false, true);
let result = parser
.detect_and_parse_reasoning("<think>with truncated reasoning")
.unwrap();
assert_eq!(result.normal_text, "");
assert_eq!(result.reasoning_text, "with truncated reasoning");
}
#[test]
fn test_parse_streaming_partial_token() {
let mut parser = create_test_parser(false, true);
let result = parser
.parse_reasoning_streaming_incremental("<thi")
.unwrap();
assert_eq!(result.normal_text, "");
assert_eq!(result.reasoning_text, "");
}
#[test]
fn test_parse_streaming_complete() {
let mut parser = create_test_parser(false, true);
let result = parser
.parse_reasoning_streaming_incremental("<think>with reasoning</think> and more text.")
.unwrap();
assert_eq!(result.normal_text, " and more text.");
assert_eq!(result.reasoning_text, "with reasoning");
}
#[test]
fn test_parse_streaming_no_end_token() {
let mut parser = create_test_parser(true, true);
let result = parser
.parse_reasoning_streaming_incremental("<think>with reasoning")
.unwrap();
assert_eq!(result.normal_text, "");
assert_eq!(result.reasoning_text, "with reasoning");
}
#[test]
fn test_initial_in_reasoning_true() {
// Parser starts with in_reasoning=true (like DeepSeek-R1)
let mut parser = create_test_parser(true, true);
let result = parser
.detect_and_parse_reasoning("no think tags here")
.unwrap();
assert_eq!(result.normal_text, "");
assert_eq!(result.reasoning_text, "no think tags here");
}
#[test]
fn test_buffer_loss_bug_fix() {
// Critical test for buffer preservation
let mut parser = create_test_parser(false, true);
// Step 1: Send partial end tag when not in reasoning mode
let result1 = parser.parse_reasoning_streaming_incremental("</").unwrap();
assert_eq!(result1.normal_text, "");
assert_eq!(result1.reasoning_text, "");
// Step 2: Send normal text that doesn't complete the end tag
// Must return "</answer" not just "answer"
let result2 = parser
.parse_reasoning_streaming_incremental("answer")
.unwrap();
assert_eq!(result2.normal_text, "</answer");
assert_eq!(result2.reasoning_text, "");
}
#[test]
fn test_streaming_with_stream_reasoning_enabled() {
let mut parser = create_test_parser(false, true);
// Start reasoning block
let result1 = parser
.parse_reasoning_streaming_incremental("<think>reasoning ")
.unwrap();
assert_eq!(result1.normal_text, "");
assert_eq!(result1.reasoning_text, "reasoning ");
// Continue streaming reasoning
let result2 = parser
.parse_reasoning_streaming_incremental("content ")
.unwrap();
assert_eq!(result2.normal_text, "");
assert_eq!(result2.reasoning_text, "content ");
// End reasoning block
let result3 = parser
.parse_reasoning_streaming_incremental("more</think> normal")
.unwrap();
assert_eq!(result3.normal_text, " normal");
assert_eq!(result3.reasoning_text, "more");
}
#[test]
fn test_reset_state() {
let mut parser = create_test_parser(false, true);
// Process some text
parser
.parse_reasoning_streaming_incremental("<think>reasoning</think> normal")
.unwrap();
// Reset and verify state
parser.reset();
assert!(!parser.in_reasoning);
assert!(parser.buffer.is_empty());
assert!(!parser.stripped_think_start);
}
#[test]
fn test_buffer_overflow_detect_and_parse() {
let config = ParserConfig {
max_buffer_size: 10, // Set a very small buffer
..Default::default()
};
let mut parser = BaseReasoningParser::new(config);
let large_text = "a".repeat(20);
let result = parser.detect_and_parse_reasoning(&large_text);
assert!(result.is_err());
match result {
Err(ParseError::BufferOverflow(size)) => {
assert_eq!(size, 20);
}
_ => panic!("Expected BufferOverflow error"),
}
}
#[test]
fn test_buffer_overflow_streaming() {
let config = ParserConfig {
max_buffer_size: 10, // Set a very small buffer
..Default::default()
};
let mut parser = BaseReasoningParser::new(config);
// Send a partial token that will be buffered
let result1 = parser.parse_reasoning_streaming_incremental("<thi");
assert!(result1.is_ok());
assert_eq!(result1.unwrap().normal_text, "");
// Second chunk would exceed buffer
// Buffer has "<thi" (4 chars) + "this_is_too_large" (17 chars) = 21 total
let result2 = parser.parse_reasoning_streaming_incremental("this_is_too_large");
assert!(result2.is_err());
match result2 {
Err(ParseError::BufferOverflow(size)) => {
assert_eq!(size, 21); // 4 + 17
}
_ => panic!("Expected BufferOverflow error"),
}
}
}

118
sgl-model-gateway/src/reasoning_parser/parsers/deepseek_r1.rs
View File

@@ -1,118 +0,0 @@
// DeepSeek-R1 specific reasoning parser.
// This parser starts with in_reasoning=true, assuming all text is reasoning
// until an end token is encountered.
use crate::reasoning_parser::{
parsers::BaseReasoningParser,
traits::{ParseError, ParserConfig, ParserResult, ReasoningParser},
};
/// DeepSeek-R1 reasoning parser.
///
/// This parser assumes reasoning from the start of text (in_reasoning=true)
/// and uses <think> and </think> tokens.
pub struct DeepSeekR1Parser {
base: BaseReasoningParser,
}
impl DeepSeekR1Parser {
/// Create a new DeepSeek-R1 parser.
pub fn new() -> Self {
let config = ParserConfig {
think_start_token: "<think>".to_string(),
think_end_token: "</think>".to_string(),
stream_reasoning: true,
max_buffer_size: 65536,
initial_in_reasoning: true, // Always starts with reasoning
};
Self {
base: BaseReasoningParser::new(config).with_model_type("deepseek_r1".to_string()),
}
}
}
impl Default for DeepSeekR1Parser {
fn default() -> Self {
Self::new()
}
}
impl ReasoningParser for DeepSeekR1Parser {
fn detect_and_parse_reasoning(&mut self, text: &str) -> Result<ParserResult, ParseError> {
self.base.detect_and_parse_reasoning(text)
}
fn parse_reasoning_streaming_incremental(
&mut self,
text: &str,
) -> Result<ParserResult, ParseError> {
self.base.parse_reasoning_streaming_incremental(text)
}
fn reset(&mut self) {
self.base.reset()
}
fn model_type(&self) -> &str {
self.base.model_type()
}
fn is_in_reasoning(&self) -> bool {
self.base.is_in_reasoning()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_deepseek_r1_initial_state() {
let mut parser = DeepSeekR1Parser::new();
// Should treat text as reasoning even without start token
let result = parser
.detect_and_parse_reasoning("This is reasoning content")
.unwrap();
assert_eq!(result.normal_text, "");
assert_eq!(result.reasoning_text, "This is reasoning content");
}
#[test]
fn test_deepseek_r1_with_end_token() {
let mut parser = DeepSeekR1Parser::new();
// Should extract reasoning until end token
let result = parser
.detect_and_parse_reasoning("reasoning content</think>normal content")
.unwrap();
assert_eq!(result.normal_text, "normal content");
assert_eq!(result.reasoning_text, "reasoning content");
}
#[test]
fn test_deepseek_r1_streaming() {
let mut parser = DeepSeekR1Parser::new();
// First chunk - all reasoning
let result1 = parser
.parse_reasoning_streaming_incremental("thinking about")
.unwrap();
assert_eq!(result1.reasoning_text, "thinking about");
assert_eq!(result1.normal_text, "");
// Second chunk - ends reasoning
let result2 = parser
.parse_reasoning_streaming_incremental(" the problem</think>answer")
.unwrap();
assert_eq!(result2.reasoning_text, "the problem"); // Text is trimmed
assert_eq!(result2.normal_text, "answer");
}
#[test]
fn test_model_type() {
let parser = DeepSeekR1Parser::new();
assert_eq!(parser.model_type(), "deepseek_r1");
}
}

124
sgl-model-gateway/src/reasoning_parser/parsers/glm45.rs
View File

@@ -1,124 +0,0 @@
// GLM45 specific reasoning parser.
// Uses the same format as Qwen3 but has its own implementation for debugging.
use crate::reasoning_parser::{
parsers::BaseReasoningParser,
traits::{ParseError, ParserConfig, ParserResult, ReasoningParser},
};
/// GLM45 reasoning parser.
///
/// This parser uses the same format as Qwen3 (<think>...</think>) but has
/// its own implementation for better debugging and potential future customization.
pub struct Glm45Parser {
base: BaseReasoningParser,
}
impl Glm45Parser {
/// Create a new GLM45 parser.
pub fn new() -> Self {
let config = ParserConfig {
think_start_token: "<think>".to_string(),
think_end_token: "</think>".to_string(),
stream_reasoning: true,
max_buffer_size: 65536,
initial_in_reasoning: false, // Requires explicit start token like Qwen3
};
Self {
base: BaseReasoningParser::new(config).with_model_type("glm45".to_string()),
}
}
}
impl Default for Glm45Parser {
fn default() -> Self {
Self::new()
}
}
impl ReasoningParser for Glm45Parser {
fn detect_and_parse_reasoning(&mut self, text: &str) -> Result<ParserResult, ParseError> {
self.base.detect_and_parse_reasoning(text)
}
fn parse_reasoning_streaming_incremental(
&mut self,
text: &str,
) -> Result<ParserResult, ParseError> {
self.base.parse_reasoning_streaming_incremental(text)
}
fn reset(&mut self) {
self.base.reset()
}
fn model_type(&self) -> &str {
self.base.model_type()
}
fn is_in_reasoning(&self) -> bool {
self.base.is_in_reasoning()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_glm45_initial_state() {
let mut parser = Glm45Parser::new();
// Should NOT treat text as reasoning without start token
let result = parser
.detect_and_parse_reasoning("This is normal content")
.unwrap();
assert_eq!(result.normal_text, "This is normal content");
assert_eq!(result.reasoning_text, "");
}
#[test]
fn test_glm45_with_tokens() {
let mut parser = Glm45Parser::new();
// Should extract reasoning with proper tokens
let result = parser
.detect_and_parse_reasoning("<think>reasoning content</think>answer")
.unwrap();
assert_eq!(result.normal_text, "answer");
assert_eq!(result.reasoning_text, "reasoning content");
}
#[test]
fn test_glm45_streaming() {
let mut parser = Glm45Parser::new();
// First chunk - normal text
let result1 = parser
.parse_reasoning_streaming_incremental("normal text ")
.unwrap();
assert_eq!(result1.normal_text, "normal text ");
assert_eq!(result1.reasoning_text, "");
// Second chunk - enters reasoning
let result2 = parser
.parse_reasoning_streaming_incremental("<think>reasoning")
.unwrap();
assert_eq!(result2.normal_text, "");
assert_eq!(result2.reasoning_text, "reasoning");
// Third chunk - exits reasoning
let result3 = parser
.parse_reasoning_streaming_incremental("</think>answer")
.unwrap();
assert_eq!(result3.normal_text, "answer");
assert_eq!(result3.reasoning_text, "");
}
#[test]
fn test_model_type() {
let parser = Glm45Parser::new();
assert_eq!(parser.model_type(), "glm45");
}
}

142
sgl-model-gateway/src/reasoning_parser/parsers/kimi.rs
View File

@@ -1,142 +0,0 @@
// Kimi specific reasoning parser.
// This parser uses Unicode tokens and starts with in_reasoning=false.
use crate::reasoning_parser::{
parsers::BaseReasoningParser,
traits::{ParseError, ParserConfig, ParserResult, ReasoningParser},
};
/// Kimi reasoning parser.
///
/// This parser uses Unicode tokens (◁think▷ and ◁/think▷) and requires
/// explicit start tokens to enter reasoning mode.
pub struct KimiParser {
base: BaseReasoningParser,
}
impl KimiParser {
/// Create a new Kimi parser.
pub fn new() -> Self {
let config = ParserConfig {
think_start_token: "◁think▷".to_string(),
think_end_token: "◁/think▷".to_string(),
stream_reasoning: true,
max_buffer_size: 65536,
initial_in_reasoning: false, // Requires explicit start token
};
Self {
base: BaseReasoningParser::new(config).with_model_type("kimi".to_string()),
}
}
}
impl Default for KimiParser {
fn default() -> Self {
Self::new()
}
}
impl ReasoningParser for KimiParser {
fn detect_and_parse_reasoning(&mut self, text: &str) -> Result<ParserResult, ParseError> {
self.base.detect_and_parse_reasoning(text)
}
fn parse_reasoning_streaming_incremental(
&mut self,
text: &str,
) -> Result<ParserResult, ParseError> {
self.base.parse_reasoning_streaming_incremental(text)
}
fn reset(&mut self) {
self.base.reset()
}
fn model_type(&self) -> &str {
self.base.model_type()
}
fn is_in_reasoning(&self) -> bool {
self.base.is_in_reasoning()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_kimi_initial_state() {
let mut parser = KimiParser::new();
// Should NOT treat text as reasoning without start token
let result = parser
.detect_and_parse_reasoning("This is normal content")
.unwrap();
assert_eq!(result.normal_text, "This is normal content");
assert_eq!(result.reasoning_text, "");
}
#[test]
fn test_kimi_with_unicode_tokens() {
let mut parser = KimiParser::new();
// Should extract reasoning with Unicode tokens
let result = parser
.detect_and_parse_reasoning("◁think▷reasoning content◁/think▷answer")
.unwrap();
assert_eq!(result.normal_text, "answer");
assert_eq!(result.reasoning_text, "reasoning content");
}
#[test]
fn test_kimi_partial_unicode() {
let mut parser = KimiParser::new();
let result1 = parser
.parse_reasoning_streaming_incremental("◁thi")
.unwrap();
assert_eq!(result1.normal_text, "");
assert_eq!(result1.reasoning_text, "");
// Complete the token
let result2 = parser
.parse_reasoning_streaming_incremental("nk▷reasoning")
.unwrap();
assert_eq!(result2.normal_text, "");
assert_eq!(result2.reasoning_text, "reasoning");
}
#[test]
fn test_kimi_streaming() {
let mut parser = KimiParser::new();
// Normal text first
let result1 = parser
.parse_reasoning_streaming_incremental("normal ")
.unwrap();
assert_eq!(result1.normal_text, "normal ");
assert_eq!(result1.reasoning_text, "");
// Enter reasoning with Unicode token
let result2 = parser
.parse_reasoning_streaming_incremental("◁think▷thinking")
.unwrap();
assert_eq!(result2.normal_text, "");
assert_eq!(result2.reasoning_text, "thinking");
// Exit reasoning
let result3 = parser
.parse_reasoning_streaming_incremental("◁/think▷answer")
.unwrap();
assert_eq!(result3.normal_text, "answer");
assert_eq!(result3.reasoning_text, ""); // Already returned in stream mode
}
#[test]
fn test_model_type() {
let parser = KimiParser::new();
assert_eq!(parser.model_type(), "kimi");
}
}

166
sgl-model-gateway/src/reasoning_parser/parsers/minimax.rs
View File

@@ -1,166 +0,0 @@
// MiniMax M2 specific reasoning parser.
// This parser automatically appends <think> token at the beginning of text,
// similar to the Python MiniMaxAppendThinkDetector.
use crate::reasoning_parser::{
parsers::BaseReasoningParser,
traits::{ParseError, ParserConfig, ParserResult, ReasoningParser},
};
/// MiniMax M2 reasoning parser.
///
/// This parser automatically appends <think> token at the beginning of the first chunk
/// and uses <think> and </think> tokens for reasoning blocks.
pub struct MiniMaxParser {
base: BaseReasoningParser,
is_first_chunk: bool,
}
impl MiniMaxParser {
/// Create a new MiniMax M2 parser.
pub fn new() -> Self {
let config = ParserConfig {
think_start_token: "<think>".to_string(),
think_end_token: "</think>".to_string(),
stream_reasoning: true,
max_buffer_size: 65536,
initial_in_reasoning: false, // Start with false, we'll add <think> manually
};
Self {
base: BaseReasoningParser::new(config).with_model_type("minimax".to_string()),
is_first_chunk: true,
}
}
}
impl Default for MiniMaxParser {
fn default() -> Self {
Self::new()
}
}
impl ReasoningParser for MiniMaxParser {
fn detect_and_parse_reasoning(&mut self, text: &str) -> Result<ParserResult, ParseError> {
// For one-shot parsing, prepend <think> token to the text
let modified_text = format!("<think>{}", text);
self.base.detect_and_parse_reasoning(&modified_text)
}
fn parse_reasoning_streaming_incremental(
&mut self,
text: &str,
) -> Result<ParserResult, ParseError> {
// For the first chunk, prepend <think> token
let modified_text = if self.is_first_chunk {
self.is_first_chunk = false;
format!("<think>{}", text)
} else {
text.to_string()
};
self.base
.parse_reasoning_streaming_incremental(&modified_text)
}
fn reset(&mut self) {
self.base.reset();
self.is_first_chunk = true; // Reset the first chunk flag
}
fn model_type(&self) -> &str {
self.base.model_type()
}
fn is_in_reasoning(&self) -> bool {
self.base.is_in_reasoning()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_minimax_append_think_oneshot() {
let mut parser = MiniMaxParser::new();
// Should automatically prepend <think> and parse as reasoning
let result = parser
.detect_and_parse_reasoning("reasoning content</think>normal content")
.unwrap();
assert_eq!(result.normal_text, "normal content");
assert_eq!(result.reasoning_text, "reasoning content");
}
#[test]
fn test_minimax_without_end_token() {
let mut parser = MiniMaxParser::new();
// Should treat all content as reasoning when no end token
let result = parser
.detect_and_parse_reasoning("all reasoning content")
.unwrap();
assert_eq!(result.normal_text, "");
assert_eq!(result.reasoning_text, "all reasoning content");
}
#[test]
fn test_minimax_streaming_first_chunk() {
let mut parser = MiniMaxParser::new();
// First chunk should have <think> prepended
let result1 = parser
.parse_reasoning_streaming_incremental("thinking about")
.unwrap();
assert_eq!(result1.reasoning_text, "thinking about");
assert_eq!(result1.normal_text, "");
// Second chunk should not have <think> prepended
let result2 = parser
.parse_reasoning_streaming_incremental(" the problem</think>answer")
.unwrap();
assert_eq!(result2.reasoning_text, "the problem"); // Text is trimmed
assert_eq!(result2.normal_text, "answer");
}
#[test]
fn test_minimax_reset() {
let mut parser = MiniMaxParser::new();
// First use
let result1 = parser
.parse_reasoning_streaming_incremental("first")
.unwrap();
assert_eq!(result1.reasoning_text, "first");
// Reset the parser
parser.reset();
// After reset, should be first chunk again
let result2 = parser
.parse_reasoning_streaming_incremental("second")
.unwrap();
assert_eq!(result2.reasoning_text, "second");
}
#[test]
fn test_minimax_already_has_think() {
let mut parser = MiniMaxParser::new();
// Even if text already has <think>, it will add another one
// This mimics the Python behavior
let result = parser
.detect_and_parse_reasoning("<think>content</think>answer")
.unwrap();
// The double <think> gets handled by the base parser which removes duplicates
assert_eq!(result.normal_text, "answer");
assert_eq!(result.reasoning_text, "content");
}
#[test]
fn test_model_type() {
let parser = MiniMaxParser::new();
assert_eq!(parser.model_type(), "minimax");
}
}

15
sgl-model-gateway/src/reasoning_parser/parsers/mod.rs
View File

@@ -1,15 +0,0 @@
pub mod base;
pub mod deepseek_r1;
pub mod glm45;
pub mod kimi;
pub mod minimax;
pub mod qwen3;
pub mod step3;
pub use base::BaseReasoningParser;
pub use deepseek_r1::DeepSeekR1Parser;
pub use glm45::Glm45Parser;
pub use kimi::KimiParser;
pub use minimax::MiniMaxParser;
pub use qwen3::{Qwen3Parser, QwenThinkingParser};
pub use step3::Step3Parser;

188
sgl-model-gateway/src/reasoning_parser/parsers/qwen3.rs
View File

@@ -1,188 +0,0 @@
// Qwen3 specific reasoning parser.
// This parser starts with in_reasoning=false, requiring an explicit
// start token to enter reasoning mode.
use crate::reasoning_parser::{
parsers::BaseReasoningParser,
traits::{ParseError, ParserConfig, ParserResult, ReasoningParser},
};
/// Qwen3 reasoning parser.
///
/// This parser requires explicit <think> tokens to enter reasoning mode
/// (in_reasoning=false initially).
pub struct Qwen3Parser {
base: BaseReasoningParser,
}
impl Qwen3Parser {
/// Create a new Qwen3 parser.
pub fn new() -> Self {
let config = ParserConfig {
think_start_token: "<think>".to_string(),
think_end_token: "</think>".to_string(),
stream_reasoning: true,
max_buffer_size: 65536,
initial_in_reasoning: false, // Requires explicit start token
};
Self {
base: BaseReasoningParser::new(config).with_model_type("qwen3".to_string()),
}
}
}
impl Default for Qwen3Parser {
fn default() -> Self {
Self::new()
}
}
impl ReasoningParser for Qwen3Parser {
fn detect_and_parse_reasoning(&mut self, text: &str) -> Result<ParserResult, ParseError> {
self.base.detect_and_parse_reasoning(text)
}
fn parse_reasoning_streaming_incremental(
&mut self,
text: &str,
) -> Result<ParserResult, ParseError> {
self.base.parse_reasoning_streaming_incremental(text)
}
fn reset(&mut self) {
self.base.reset()
}
fn model_type(&self) -> &str {
self.base.model_type()
}
fn is_in_reasoning(&self) -> bool {
self.base.is_in_reasoning()
}
}
/// QwenThinking parser - variant that assumes reasoning from start.
///
/// This is for qwen*thinking models that behave like DeepSeek-R1.
pub struct QwenThinkingParser {
base: BaseReasoningParser,
}
impl QwenThinkingParser {
/// Create a new QwenThinking parser.
pub fn new() -> Self {
let config = ParserConfig {
think_start_token: "<think>".to_string(),
think_end_token: "</think>".to_string(),
stream_reasoning: true,
max_buffer_size: 65536,
initial_in_reasoning: true, // Assumes reasoning from start
};
Self {
base: BaseReasoningParser::new(config).with_model_type("qwen_thinking".to_string()),
}
}
}
impl Default for QwenThinkingParser {
fn default() -> Self {
Self::new()
}
}
impl ReasoningParser for QwenThinkingParser {
fn detect_and_parse_reasoning(&mut self, text: &str) -> Result<ParserResult, ParseError> {
self.base.detect_and_parse_reasoning(text)
}
fn parse_reasoning_streaming_incremental(
&mut self,
text: &str,
) -> Result<ParserResult, ParseError> {
self.base.parse_reasoning_streaming_incremental(text)
}
fn reset(&mut self) {
self.base.reset()
}
fn model_type(&self) -> &str {
self.base.model_type()
}
fn is_in_reasoning(&self) -> bool {
self.base.is_in_reasoning()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_qwen3_initial_state() {
let mut parser = Qwen3Parser::new();
// Should NOT treat text as reasoning without start token
let result = parser
.detect_and_parse_reasoning("This is normal content")
.unwrap();
assert_eq!(result.normal_text, "This is normal content");
assert_eq!(result.reasoning_text, "");
}
#[test]
fn test_qwen3_with_tokens() {
let mut parser = Qwen3Parser::new();
// Should extract reasoning with proper tokens
let result = parser
.detect_and_parse_reasoning("<think>reasoning</think>answer")
.unwrap();
assert_eq!(result.normal_text, "answer");
assert_eq!(result.reasoning_text, "reasoning");
}
#[test]
fn test_qwen_thinking_initial_state() {
let mut parser = QwenThinkingParser::new();
// Should treat text as reasoning even without start token
let result = parser
.detect_and_parse_reasoning("This is reasoning content")
.unwrap();
assert_eq!(result.normal_text, "");
assert_eq!(result.reasoning_text, "This is reasoning content");
}
#[test]
fn test_qwen3_streaming() {
let mut parser = Qwen3Parser::new();
// First chunk - normal text (no start token yet)
let result1 = parser
.parse_reasoning_streaming_incremental("normal text ")
.unwrap();
assert_eq!(result1.normal_text, "normal text ");
assert_eq!(result1.reasoning_text, "");
// Second chunk - enters reasoning
let result2 = parser
.parse_reasoning_streaming_incremental("<think>reasoning")
.unwrap();
assert_eq!(result2.normal_text, "");
assert_eq!(result2.reasoning_text, "reasoning");
}
#[test]
fn test_model_types() {
let qwen3 = Qwen3Parser::new();
assert_eq!(qwen3.model_type(), "qwen3");
let qwen_thinking = QwenThinkingParser::new();
assert_eq!(qwen_thinking.model_type(), "qwen_thinking");
}
}

129
sgl-model-gateway/src/reasoning_parser/parsers/step3.rs
View File

@@ -1,129 +0,0 @@
// Step3 specific reasoning parser.
// Uses the same format as DeepSeek-R1 but has its own implementation for debugging.
use crate::reasoning_parser::{
parsers::BaseReasoningParser,
traits::{ParseError, ParserConfig, ParserResult, ReasoningParser},
};
/// Step3 reasoning parser.
///
/// This parser uses the same format as DeepSeek-R1 (<think>...</think>) but has
/// its own implementation for better debugging and potential future customization.
pub struct Step3Parser {
base: BaseReasoningParser,
}
impl Step3Parser {
/// Create a new Step3 parser.
pub fn new() -> Self {
let config = ParserConfig {
think_start_token: "<think>".to_string(),
think_end_token: "</think>".to_string(),
stream_reasoning: true,
max_buffer_size: 65536,
initial_in_reasoning: true, // Assumes reasoning from start like DeepSeek-R1
};
Self {
base: BaseReasoningParser::new(config).with_model_type("step3".to_string()),
}
}
}
impl Default for Step3Parser {
fn default() -> Self {
Self::new()
}
}
impl ReasoningParser for Step3Parser {
fn detect_and_parse_reasoning(&mut self, text: &str) -> Result<ParserResult, ParseError> {
self.base.detect_and_parse_reasoning(text)
}
fn parse_reasoning_streaming_incremental(
&mut self,
text: &str,
) -> Result<ParserResult, ParseError> {
self.base.parse_reasoning_streaming_incremental(text)
}
fn reset(&mut self) {
self.base.reset()
}
fn model_type(&self) -> &str {
self.base.model_type()
}
fn is_in_reasoning(&self) -> bool {
self.base.is_in_reasoning()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_step3_initial_state() {
let mut parser = Step3Parser::new();
// Should treat text as reasoning even without start token
let result = parser
.detect_and_parse_reasoning("This is reasoning content")
.unwrap();
assert_eq!(result.normal_text, "");
assert_eq!(result.reasoning_text, "This is reasoning content");
}
#[test]
fn test_step3_with_end_token() {
let mut parser = Step3Parser::new();
// Should handle text with end token
let result = parser
.detect_and_parse_reasoning("reasoning content</think>answer")
.unwrap();
assert_eq!(result.normal_text, "answer");
assert_eq!(result.reasoning_text, "reasoning content");
}
#[test]
fn test_step3_with_both_tokens() {
let mut parser = Step3Parser::new();
// Should handle both start and end tokens
let result = parser
.detect_and_parse_reasoning("<think>reasoning content</think>answer")
.unwrap();
assert_eq!(result.normal_text, "answer");
assert_eq!(result.reasoning_text, "reasoning content");
}
#[test]
fn test_step3_streaming() {
let mut parser = Step3Parser::new();
// First chunk - treated as reasoning (initial_in_reasoning=true)
let result1 = parser
.parse_reasoning_streaming_incremental("reasoning text ")
.unwrap();
assert_eq!(result1.normal_text, "");
assert_eq!(result1.reasoning_text, "reasoning text ");
// Second chunk - continues reasoning until end token
let result2 = parser
.parse_reasoning_streaming_incremental("more reasoning</think>answer")
.unwrap();
assert_eq!(result2.normal_text, "answer");
assert_eq!(result2.reasoning_text, "more reasoning");
}
#[test]
fn test_model_type() {
let parser = Step3Parser::new();
assert_eq!(parser.model_type(), "step3");
}
}

135
sgl-model-gateway/src/reasoning_parser/traits.rs
View File

@@ -1,135 +0,0 @@
use std::fmt;
/// Result of parsing text for reasoning content.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct ParserResult {
/// The normal text outside reasoning blocks.
pub normal_text: String,
/// The extracted reasoning text from within reasoning blocks.
pub reasoning_text: String,
}
impl ParserResult {
/// Create a new ParserResult with the given normal and reasoning text.
pub fn new(normal_text: String, reasoning_text: String) -> Self {
Self {
normal_text,
reasoning_text,
}
}
/// Create a result with only normal text.
pub fn normal(text: String) -> Self {
Self {
normal_text: text,
reasoning_text: String::new(),
}
}
/// Create a result with only reasoning text.
pub fn reasoning(text: String) -> Self {
Self {
normal_text: String::new(),
reasoning_text: text,
}
}
/// Check if this result contains any text.
pub fn is_empty(&self) -> bool {
self.normal_text.is_empty() && self.reasoning_text.is_empty()
}
}
/// Trait for parsing reasoning content from LLM outputs.
pub trait ReasoningParser: Send + Sync {
/// Detects and parses reasoning from the input text (one-time parsing).
///
/// This method is used for non-streaming scenarios where the complete
/// text is available at once.
///
/// Returns an error if the text exceeds buffer limits or contains invalid UTF-8.
fn detect_and_parse_reasoning(&mut self, text: &str) -> Result<ParserResult, ParseError>;
/// Parses reasoning incrementally from streaming input.
///
/// This method maintains internal state across calls to handle partial
/// tokens and chunk boundaries correctly.
///
/// Returns an error if the buffer exceeds max_buffer_size.
fn parse_reasoning_streaming_incremental(
&mut self,
text: &str,
) -> Result<ParserResult, ParseError>;
/// Reset the parser state for reuse.
///
/// This should clear any buffers and reset flags to initial state.
fn reset(&mut self);
/// Get the model type this parser is designed for.
fn model_type(&self) -> &str;
/// Check if the parser is currently in reasoning mode.
///
/// Returns true if the parser is currently parsing reasoning content.
fn is_in_reasoning(&self) -> bool;
}
/// Error types for reasoning parsing operations.
#[derive(Debug, thiserror::Error)]
pub enum ParseError {
#[error("Invalid UTF-8 in stream: {0}")]
Utf8Error(#[from] std::str::Utf8Error),
#[error("Buffer overflow: {0} bytes exceeds maximum")]
BufferOverflow(usize),
#[error("Unknown model type: {0}")]
UnknownModel(String),
#[error("Parser configuration error: {0}")]
ConfigError(String),
}
/// Configuration for parser behavior.
#[derive(Debug, Clone)]
pub struct ParserConfig {
/// The token that marks the start of reasoning content.
pub think_start_token: String,
/// The token that marks the end of reasoning content.
pub think_end_token: String,
/// Whether to stream reasoning content as it arrives.
pub stream_reasoning: bool,
/// Maximum buffer size in bytes.
pub max_buffer_size: usize,
/// Initial state for in_reasoning flag (fixed per parser type).
pub initial_in_reasoning: bool,
}
impl Default for ParserConfig {
fn default() -> Self {
Self {
think_start_token: "<think>".to_string(),
think_end_token: "</think>".to_string(),
stream_reasoning: true,
max_buffer_size: 65536, // 64KB default
initial_in_reasoning: false, // Default to false (explicit reasoning)
}
}
}
impl fmt::Display for ParserResult {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"ParserResult {{ normal: {} chars, reasoning: {} chars }}",
self.normal_text.len(),
self.reasoning_text.len()
)
}
}