Compare commits

..

15 Commits

Author SHA1 Message Date
ViperEkura a30e3d5114 fix: 修复 shields.io GitHub badge 因 token 耗尽而无法显示
- 新增 Action 每天及 push 时同步 badges 至 gh-pages
- README 改用 endpoint 格式指向自建静态 JSON, 不依赖 shields.io GitHub token 池
- 同步更新中英两份 README
2026-06-16 22:21:58 +08:00
ViperEkura 1818d06576 feat: 新增 IFD 数据质量评分工具, 移动 ppl 至 eval
- 计算指令遵循难度分数用于数据筛选
- IFD = 条件交叉熵 / 无条件交叉熵
- perplexity 移至 scripts/eval/
2026-06-16 22:03:45 +08:00
ViperEkura 4e8d1ee24e feat: 新增 IFEval 指令遵循评测
- 实现 25 种正则约束 verifier
- 将评测脚本从 scripts/tools/ 移至 scripts/eval/
2026-06-16 21:57:34 +08:00
ViperEkura fec376b0dd fix : 修复策略相关文件的类型注解与抽象方法体
- 修复 strategy.py 单元素 Union 与缺失的参数/返回类型注解
- 修复 train_context.py 8 个 default=None 字段缺 Optional 标记
- 修复 sample.py/packing.py/position_id.py 方法缺参数及返回类型注解
- 修复 factory.py _resolve_type/list_registered 缺类型注解
- 修复 train_config.py 裸 dict/list 缺泛型参数
- abstractmethod body 从 ... 改为 raise NotImplementedError
- feat : checkpoint meta.json 保存 TrainConfig 超参供人工查阅
2026-06-14 16:20:10 +08:00
ViperEkura a2512f8a5a fix : resume_dir 无权重文件时不强制加载,支持仅配置训练
- Checkpoint.load_any 统一处理 meta.json / model.safetensors / 无文件三种情况
- train_context.py 调用简化为单一路径,移除 load_model_weights 直接依赖
2026-06-13 15:40:14 +08:00
ViperEkura 457e16ea3c fix : val_loss 默认改为 None,日志跳过空值;val_dataloader 补 Optional 注解 2026-06-13 14:24:13 +08:00
ViperEkura daf627a6de fix : _save_log 前确保日志目录存在,防止跨进程反序列化后目录丢失 2026-06-12 15:39:54 +08:00
ViperEkura 445378667f feat : NEFTune 噪声注入 + label_smoothing 默认值修正
- Embedding.forward 训练时注入 randn 噪声,缩放系数 neftune_noise_alpha / sqrt(seq_len)
- TrainConfig.neftune_alpha 通过 config 传递(默认 0=关闭)
- TrainContextBuilder 将 config.neftune_alpha 写入 embed_tokens
- --neftune_alpha CLI 参数(典型值 5.0)
- label_smoothing 默认值 0.05 -> 0.0
2026-06-11 15:32:43 +08:00
ViperEkura 6ae1828449 refactor : 清理工厂和配置系统中的死代码与冗余抽象
- 删除 Registry 中未使用的 category/priority 字段,_entries 简化为直接存储类引用
- 修正 __init_subclass__ 避免叶子类(AutoRegressiveLM 等)创建空注册表
- 删除 5 个工厂的薄 create() 覆写,统一使用 BaseFactory.create(name, *args, **kwargs)
- 删除 3 处零调用的 available_types/available_strategies 别名死代码
- 删除零调用的 BaseModelConfig.to_file 死代码
- 将 BaseConfig.from_json/to_json 重命名为 from_file/to_file,消除与子类重复
- 移除两个 inference builder 中总是被覆写的 prompt_tokens=0
2026-06-07 11:39:50 +08:00
ViperEkura e7b18b7c03 refactor : BaseFactory 基类类型自动推导 + 移除冗余代码
- _validate_component 从 BaseFactory[T] 泛型参数自动解析基类类型,9 个子类覆写移除
- Registry 类内联到 BaseFactory._entries,移除未用的 list_by_category/list_by_priority
- _component_base 在 __init_subclass__ 时立即解析
- 数据集 4 个子类冗余 __init__ 移除
2026-06-06 21:23:41 +08:00
ViperEkura 9e31d4ef2b feat : BaseToolParser.feed 增加可选 token_ids 参数
- format_chunk ABC 改为 (token, **kwargs),body/token_ids 通过 kw 传入
- ProtocolHandler._handle_stream 逐 token encode 并透传
- Anthropic builder 用 **kwargs 吸收不使用的参数,零变更
- 新增 3 个 token_ids 参数测试
2026-06-06 11:19:30 +08:00
ViperEkura 52aa4d01d5 feat : 推理层增加 vLLM 风格工具调用解析
- 新增 BaseToolParser 抽象基类,定义 feed/parse_complete 流式接口
- 新增 SimpleJsonToolParser,解析 {"name":"...","arguments":{...}} 格式
- 新增 ToolParserFactory,基于 BaseFactory 实现可插拔注册
- 集成 parser 到 OpenAIResponseBuilder,支持流式/非流式工具调用
- 扩展 ChatMessage 和 ChatCompletionRequest,增加 tools/tool_choice 字段
- 重构 format_chunk 接口,传入累积文本支持全量重新解析
- 新增 74 个单元测试,覆盖扫描/查找/流式解析/完整解析/工厂
2026-06-06 08:54:10 +08:00
ViperEkura 986be957ec refactor : on_batch_begin 移入 accumulate 上下文 2026-06-06 01:19:21 +08:00
ViperEkura cf9c60841b docs : 按代码反向修正所有文档错误
- 更新预处理模块目录结构和类名(SectionedMaskBuilder)
- 修正 ResponseBuilder.prepare 签名(tokenizer → engine)
- 补全缺失的 CLI 参数、配置字段和数据键名
- 修正 README 中 download.py 的描述
2026-06-06 01:06:30 +08:00
ViperEkura 31bc7f5c2a refactor : pipeline 策略化拆分,消除 _flush if/else
- PackingStrategy / PositionIdStrategy / StoreWriter 独立文件 + Factory
- Pipeline._flush 零 if/else,纯编排
- SectionRenderer 从 SectionedMaskBuilder 分离
- OutputConfig.position_ids_mode 默认改为 ""none""
2026-06-06 00:45:33 +08:00
52 changed files with 2741 additions and 705 deletions

44
.github/workflows/update-badges.yml vendored Normal file
View File

@ -0,0 +1,44 @@
name: Update Badges
on:
push:
branches: [main]
schedule:
- cron: "0 0 * * *"
workflow_dispatch:
permissions:
contents: write
jobs:
update:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Fetch repo stats
id: api
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
run: |
mkdir -p badges
REPO=$(gh repo view --json stargazerCount,forkCount,latestRelease --jq '.')
STARS=$(echo "$REPO" | jq -r '.stargazerCount')
FORKS=$(echo "$REPO" | jq -r '.forkCount')
RELEASE=$(echo "$REPO" | jq -r '.latestRelease.tagName // "N/A"')
echo '{"schemaVersion":1,"label":"release","message":"'"$RELEASE"'","color":"76bad9"}' > badges/release.json
echo '{"schemaVersion":1,"label":"stars","message":"'"$STARS"'","color":"76bad9"}' > badges/stars.json
echo '{"schemaVersion":1,"label":"forks","message":"'"$FORKS"'","color":"76bad9"}' > badges/forks.json
- name: Deploy to gh-pages
uses: peaceiris/actions-gh-pages@v4
with:
github_token: ${{ secrets.GITHUB_TOKEN }}
publish_dir: badges
destination_dir: badges
commit_message: "Sync badges"
user_name: "github-actions[bot]"
user_email: "github-actions[bot]@users.noreply.github.com"

View File

@ -9,9 +9,9 @@
<div align="center">
<img src="https://img.shields.io/badge/python-3.12+-blue.svg" alt="python">
<img src="https://img.shields.io/badge/license-GPL--3.0-blue.svg" alt="license">
<img src="https://img.shields.io/github/v/release/ViperEkura/AstrAI?color=76bad9" alt="release">
<img src="https://img.shields.io/badge/dynamic/json?url=https%3A%2F%2Fapi.github.com%2Frepos%2FViperEkura%2FAstrAI&query=%24.stargazers_count&label=stars&suffix=%20stars&color=76bad9" alt="stars">
<img src="https://img.shields.io/badge/dynamic/json?url=https%3A%2F%2Fapi.github.com%2Frepos%2FViperEkura%2FAstrAI&query=%24.forks_count&label=forks&suffix=%20forks&color=76bad9" alt="forks">
<img src="https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/ViperEkura/AstrAI/gh-pages/badges/release.json" alt="release">
<img src="https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/ViperEkura/AstrAI/gh-pages/badges/stars.json" alt="stars">
<img src="https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/ViperEkura/AstrAI/gh-pages/badges/forks.json" alt="forks">
</div>
<br>
@ -201,7 +201,7 @@ curl http://localhost:8000/health
Check out the demos in the `scripts/demo/` folder:
```bash
# Download preprocessed data (required before running demos)
# Download model weights (required before running demos)
python scripts/demo/download.py
# Interactive streaming chat

View File

@ -15,9 +15,9 @@
<div align="center">
<img src="https://img.shields.io/badge/python-3.12+-blue.svg" alt="python">
<img src="https://img.shields.io/badge/license-GPL--3.0-blue.svg" alt="license">
<img src="https://img.shields.io/github/v/release/ViperEkura/AstrAI?color=76bad9" alt="release">
<img src="https://img.shields.io/badge/dynamic/json?url=https%3A%2F%2Fapi.github.com%2Frepos%2FViperEkura%2FAstrAI&query=%24.stargazers_count&label=stars&suffix=%20stars&color=76bad9" alt="stars">
<img src="https://img.shields.io/badge/dynamic/json?url=https%3A%2F%2Fapi.github.com%2Frepos%2FViperEkura%2FAstrAI&query=%24.forks_count&label=forks&suffix=%20forks&color=76bad9" alt="forks">
<img src="https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/ViperEkura/AstrAI/gh-pages/badges/release.json" alt="release">
<img src="https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/ViperEkura/AstrAI/gh-pages/badges/stars.json" alt="stars">
<img src="https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/ViperEkura/AstrAI/gh-pages/badges/forks.json" alt="forks">
</div>
<br>
@ -207,7 +207,7 @@ curl http://localhost:8000/health
查看 `scripts/demo/` 文件夹中的演示:
```bash
# 下载预处理数据(运行演示前必需)
# 下载模型权重(运行演示前必需)
python scripts/demo/download.py
# 交互式流式聊天

View File

@ -352,16 +352,11 @@ classDiagram
+build(item, config, tokenizer) Optional[dict]
}
class ChatMaskBuilder {
+build(item, config, tokenizer) Optional[dict]
}
class InstructionMaskBuilder {
+build(item, config, tokenizer) Optional[dict]
}
class TextMaskBuilder {
class SectionedMaskBuilder {
+SectionRenderer renderer
+build(item, config, tokenizer) Optional[dict]
+_build_single(item, config, tokenizer) Optional[dict]
+_build_multi(item, sources_spec, config, tokenizer) Optional[dict]
}
class Pipeline {
@ -370,8 +365,12 @@ classDiagram
+str output_dir
+str tokenizer_path
+BaseMaskBuilder mask_builder
+PackingStrategy _packer
+PositionIdStrategy _position_id
+StoreWriter _writer
+transform(item) Optional[dict]
+run()
+_flush(domains, shard_idx)
}
}
@ -841,7 +840,7 @@ classDiagram
class ResponseBuilder {
<<abstract>>
+prepare(request, tokenizer) Tuple[str, GenContext, List[str]]
+prepare(request, engine) Tuple[str, GenContext, List[str]]
+format_stream_start(ctx) List[str]
+format_chunk(token) str
+format_stream_end(ctx, stop) List[str]
@ -849,7 +848,7 @@ classDiagram
}
class OpenAIResponseBuilder {
+prepare(request, tokenizer) Tuple
+prepare(request, engine) Tuple
+format_stream_start(ctx) List[str]
+format_chunk(token) str
+format_stream_end(ctx, stop) List[str]
@ -857,7 +856,7 @@ classDiagram
}
class AnthropicResponseBuilder {
+prepare(request, tokenizer) Tuple
+prepare(request, engine) Tuple
+format_stream_start(ctx) List[str]
+format_chunk(token) str
+format_stream_end(ctx, stop) List[str]
@ -1034,7 +1033,6 @@ classDiagram
BaseSamplingStrategy <|-- TemperatureStrategy
BaseSamplingStrategy <|-- TopKStrategy
BaseSamplingStrategy <|-- TopPStrategy
BaseSamplingStrategy <|-- SamplingPipeline
ParallelModel <|-- RowParallelLinear
ParallelModel <|-- ColumnParallelLinear
AutoModel <|-- AutoRegressiveLM
@ -1063,9 +1061,7 @@ classDiagram
BaseExecutor <|-- FSDPExecutor
ResponseBuilder <|-- OpenAIResponseBuilder
ResponseBuilder <|-- AnthropicResponseBuilder
BaseMaskBuilder <|-- ChatMaskBuilder
BaseMaskBuilder <|-- InstructionMaskBuilder
BaseMaskBuilder <|-- TextMaskBuilder
BaseMaskBuilder <|-- SectionedMaskBuilder
%% --- Composition (strong ownership, part destroyed with whole) ---
KVCache *-- PagePool
@ -1162,7 +1158,7 @@ classDiagram
| Module | Components | Description |
|--------|------------|-------------|
| **astrai.config** | BaseConfig, BaseModelConfig, AutoRegressiveLMConfig, EncoderConfig, ConfigFactory, TrainConfig, PipelineConfig, InputConfig, ProcessingConfig, OutputConfig | Configuration management (to_dict/from_dict, to_file/from_file, from_json/to_json) |
| **astrai.preprocessing** | BaseMaskBuilder, MaskBuilderFactory, ChatMaskBuilder, InstructionMaskBuilder, TextMaskBuilder, Pipeline, filter_by_length, dedup_signature | Declarative JSON-driven data preprocessing |
| **astrai.preprocessing** | BaseMaskBuilder, MaskBuilderFactory, SectionedMaskBuilder, Pipeline, filter_by_length, PackingStrategy, PackingStrategyFactory, PositionIdStrategy, PositionIdStrategyFactory, StoreWriter, StoreWriterFactory | Declarative JSON-driven data preprocessing |
| **astrai.dataset** | BaseDatasetGRPODataset, StoreMmapStore, StoreFactory, ResumableDistributedSampler, DatasetFactory | Dataset loading and management |
| **astrai.serialization** | Checkpoint | Model serialization |
| **astrai.model** | AutoModel, AutoRegressiveLM, EmbeddingEncoder, DecoderBlock, GQA, MLA, MLP, DeepSeekMoE, AttnFactory, FFNFactory, RMSNorm, Linear, RotaryEmbedding, Embedding | Neural network model |

View File

@ -26,7 +26,7 @@ H5 backend supports shared memory via `.share_memory_()`. Bin (mmap) uses OS pag
| Type | Storage Keys |
|------|-------------|
| `seq` | `sequence` (→ input_ids, target_ids via offset-by-1) |
| `sft` | `sequence`, `loss_mask` |
| `sft` | `sequence`, `loss_mask`, `position_ids` |
| `dpo` | `chosen`, `rejected`, `chosen_mask`, `rejected_mask` |
| `grpo` | `prompts`, `responses`, `masks`, `rewards` |

View File

@ -48,6 +48,27 @@
| `--start_epoch` | Resume from epoch (0 = from scratch) | 0 |
| `--start_batch` | Resume from batch iteration | 0 |
### Validation
| Parameter | Description | Default |
|-----------|-------------|---------|
| `--val_split` | Ratio to split from training dataset for validation (e.g. 0.05) | None |
| `--val_step` | Number of optimizer steps between validation runs | 1000 |
### Logging
| Parameter | Description | Default |
|-----------|-------------|---------|
| `--log_dir` | Directory for metric logs | checkpoint/logs |
| `--log_interval` | Number of batch iterations between metric logs | 100 |
| `--metrics` | Metrics to log (e.g. --metrics loss lr val_loss) | ["loss", "lr"] |
### Gradient Checkpointing
| Parameter | Description | Default |
|-----------|-------------|---------|
| `--gradient_checkpointing` | Enable activation checkpointing for DecoderBlock modules | False |
### Distributed Training
| Parameter | Description | Default |
@ -56,6 +77,9 @@
| `--parallel_mode` | Parallel strategy (`none`, `ddp`, or `fsdp`) | none |
| `--device_type` | Device type | cuda |
| `--start_method` | Multiprocessing start method (`spawn`, `fork`, `forkserver`) | spawn |
| `--backend` | Distributed training backend | nccl |
| `--master_addr` | Master node address | localhost |
| `--master_port` | Master node port | 29500 |
### Strategy-specific

View File

@ -243,6 +243,9 @@ When `sources` is set, `sections` is ignored.
| `min_chars` | int | `50` | Skip text-mode items shorter than this |
| `max_chars` | int | `2000000` | Skip text-mode items longer than this |
| `max_items` | int or null | `null` | Stop after N documents |
| `packing_strategy` | str | `"simple"` | Packing strategy: `"simple"`, `"bfd"`, `"bfd_split"` |
| `max_packed_len` | int | `8192` | Maximum length of a packed bin |
| `truncation_mode` | str | `"keep_start"` | How to truncate sequences: `"keep_start"` or `"keep_end"` |
### `output`
@ -252,6 +255,7 @@ When `sources` is set, `sections` is ignored.
| `storage_format` | str | `"bin"` | `"bin"` (mmap) or `"h5"` |
| `max_tokens_per_shard` | int | `100000000` | Flush threshold in cumulative tokens |
| `dtype` | dict[str, str] | `{}` | Per-key tensor dtype override (e.g. `{"loss_mask": "bool"}`) |
| `position_ids_mode` | str | `"none"` | How to compute position_ids: `"none"`, `"doc_reset"`, `"continuous"` |
---

View File

@ -89,10 +89,10 @@ class BaseConfig:
raise TypeError
@classmethod
def from_json(cls, path: Union[str, Path]) -> Self:
def from_file(cls, path: Union[str, Path]) -> Self:
with open(path, "r", encoding="utf-8") as f:
return cls.from_dict(json.load(f))
def to_json(self, path: Union[str, Path]):
def to_file(self, path: Union[str, Path]):
with open(path, "w", encoding="utf-8") as f:
json.dump(self.to_dict(), f, indent=2, ensure_ascii=False)

View File

@ -1,6 +1,5 @@
import json
from dataclasses import dataclass
from typing import Any, Dict, Optional, Self
from typing import Any, Dict, Optional
from astrai.config.base import BaseConfig
from astrai.factory import BaseFactory
@ -22,18 +21,6 @@ class BaseModelConfig(BaseConfig):
model_type: Optional[str] = None
@classmethod
def from_file(cls, config_path: str) -> Self:
with open(config_path, "r") as f:
raw: Dict[str, Any] = json.load(f)
return cls.from_dict(raw)
def to_file(self, config_path: str):
d = self.to_dict()
config_dict = {k: v for k, v in d.items() if v is not None}
with open(config_path, "w") as f:
json.dump(config_dict, f, indent=4)
@dataclass
@ConfigFactory.register("autoregressive_lm")

View File

@ -87,7 +87,7 @@ class OutputConfig(BaseConfig):
position_ids_mode : Optional[str]
How to compute position_ids in packed sequences.
- ``None`` / ``"none"``: do not generate (backward compatible).
- ``"none"``: do not generate (default).
- ``"doc_reset"``: reset to 0 at each document boundary.
- ``"continuous"``: sequential 0, 1, 2, ... (pretrain, single doc).
"""
@ -96,7 +96,7 @@ class OutputConfig(BaseConfig):
storage_format: str = "bin"
max_tokens_per_shard: int = 100_000_000
dtype: Dict[str, str] = field(default_factory=dict)
position_ids_mode: Optional[str] = None
position_ids_mode: str = "none"
@dataclass

View File

@ -1,5 +1,5 @@
from dataclasses import dataclass, field, fields
from typing import Callable, List, Optional
from typing import Any, Callable, Dict, List, Optional
import torch.nn as nn
from torch.optim import Optimizer
@ -40,7 +40,7 @@ class TrainConfig(BaseConfig):
max_grad_norm: float = field(
default=1.0, metadata={"help": "Maximum gradient norm."}
)
gradient_checkpointing_modules: list = field(
gradient_checkpointing_modules: List[str] = field(
default_factory=list,
metadata={"help": "Module types to enable activation checkpointing for."},
)
@ -128,12 +128,16 @@ class TrainConfig(BaseConfig):
default=1000,
metadata={"help": "Number of optimizer steps between validation runs."},
)
neftune_alpha: float = field(
default=0.0,
metadata={"help": "NEFTune noise alpha (0=disabled, typical: 5.0)."},
)
executor_kwargs: dict = field(
executor_kwargs: Dict[str, Any] = field(
default_factory=dict,
metadata={"help": "Extra kwargs passed to ExecutorFactory.create()."},
)
extra_kwargs: dict = field(
extra_kwargs: Dict[str, Any] = field(
default_factory=dict, metadata={"help": "Other arguments."}
)

View File

@ -136,26 +136,6 @@ class DatasetFactory(BaseFactory["BaseDataset"]):
dataset = DatasetFactory.create("custom", window_size, stride)
"""
@classmethod
def _validate_component(cls, dataset_cls: type):
"""Validate that the dataset class inherits from BaseDataset."""
if not issubclass(dataset_cls, BaseDataset):
raise TypeError(f"{dataset_cls.__name__} must inherit from BaseDataset")
@classmethod
def create(cls, train_type: str, window_size: int, stride: int) -> "BaseDataset":
"""Create a dataset instance.
Args:
train_type: Type of training ("seq", "sft", "dpo", "grpo")
window_size: Window size for data sampling
stride: Stride between consecutive samples
Returns:
Dataset instance
"""
return super().create(train_type, window_size, stride)
@classmethod
def load(
cls,
@ -185,19 +165,11 @@ class DatasetFactory(BaseFactory["BaseDataset"]):
return dataset
@classmethod
def available_types(cls) -> list:
"""Return list of registered dataset type names."""
return cls.list_registered()
@DatasetFactory.register("seq")
class SEQDataset(BaseDataset):
"""Dataset for sequential next-token prediction training."""
def __init__(self, window_size: int, stride: int):
super().__init__(window_size, stride)
@property
def required_keys(self) -> List[str]:
return ["sequence"]
@ -218,9 +190,6 @@ class SEQDataset(BaseDataset):
class SFTDataset(BaseDataset):
"""Dataset for supervised fine-tuning with loss masking."""
def __init__(self, window_size: int, stride: int):
super().__init__(window_size, stride)
@property
def required_keys(self) -> List[str]:
return ["sequence", "loss_mask", "position_ids"]
@ -248,9 +217,6 @@ class SFTDataset(BaseDataset):
class DPODataset(BaseDataset):
"""Dataset for Direct Preference Optimization training."""
def __init__(self, window_size: int, stride: int):
super().__init__(window_size, stride)
@property
def required_keys(self) -> List[str]:
return ["chosen", "rejected", "chosen_mask", "rejected_mask"]
@ -282,9 +248,6 @@ class DPODataset(BaseDataset):
class GRPODataset(BaseDataset):
"""Dataset for Group Relative Policy Optimization training."""
def __init__(self, window_size: int, stride: int):
super().__init__(window_size, stride)
@property
def required_keys(self) -> List[str]:
return ["prompts", "responses", "masks", "rewards"]

View File

@ -222,11 +222,6 @@ class StoreFactory(BaseFactory["Store"]):
...
"""
@classmethod
def _validate_component(cls, store_cls: type):
if not issubclass(store_cls, Store):
raise TypeError(f"{store_cls.__name__} must inherit from Store")
@StoreFactory.register("h5")
class H5Store(Store):

View File

@ -1,149 +1,104 @@
"""Base factory class for extensible component registration."""
"""Base factory with decorator-based registration and kwarg-filtered instantiation."""
import inspect
import sys
from abc import ABC
from typing import Callable, Dict, Generic, List, Optional, Tuple, Type, TypeVar
from typing import (
Any,
Callable,
Dict,
ForwardRef,
Generic,
List,
Optional,
Type,
TypeVar,
Union,
)
from typing import get_args as _get_args
from typing import get_origin as _get_origin
T = TypeVar("T")
class Registry:
"""Flexible registry for component classes with category and priority support.
def _resolve_type(
arg: Union[Type, str, ForwardRef], factory_cls: type
) -> Optional[Type]:
"""Resolve a generic type-arg (str forward-ref, ForwardRef, or class)."""
if not isinstance(arg, (str, ForwardRef)):
return arg
This registry stores component classes with optional metadata (category, priority).
It provides methods for registration, retrieval, and listing with filtering.
"""
name = arg if isinstance(arg, str) else arg.__forward_arg__
if name == factory_cls.__name__:
return factory_cls
def __init__(self):
self._entries = {} # name -> (component_cls, category, priority)
mod = sys.modules.get(factory_cls.__module__)
if mod is None:
return None
ns = vars(mod)
def register(
self,
name: str,
component_cls: Type,
category: Optional[str] = None,
priority: int = 0,
):
"""Register a component class with optional category and priority."""
if name in self._entries:
raise ValueError(f"Component '{name}' is already registered")
self._entries[name] = (component_cls, category, priority)
if isinstance(arg, ForwardRef):
return arg._evaluate(ns, None, frozenset(), recursive_guard=frozenset())
def get(self, name: str) -> Type:
"""Get component class by name."""
if name not in self._entries:
raise KeyError(f"Component '{name}' not found in registry")
return self._entries[name][0]
def get_with_metadata(self, name: str) -> Tuple[Type, Optional[str], int]:
"""Get component class with its metadata."""
entry = self._entries.get(name)
if entry is None:
raise KeyError(f"Component '{name}' not found in registry")
return entry
def contains(self, name: str) -> bool:
"""Check if a name is registered."""
return name in self._entries
def list_names(self) -> List[str]:
"""Return list of registered component names."""
return sorted(self._entries.keys())
def list_by_category(self, category: str) -> List[str]:
"""Return names of components belonging to a specific category."""
return sorted(
name for name, (_, cat, _) in self._entries.items() if cat == category
)
def list_by_priority(self, reverse: bool = False) -> List[str]:
"""Return names sorted by priority (default ascending)."""
return sorted(
self._entries.keys(),
key=lambda name: self._entries[name][2],
reverse=reverse,
)
def entries(self) -> Dict[str, Tuple[Type, Optional[str], int]]:
"""Return raw entries dictionary."""
return self._entries.copy()
return ns.get(name)
class BaseFactory(ABC, Generic[T]):
"""Generic factory class for component registration and creation.
"""Generic factory with decorator-based component registration.
This base class provides a decorator-based registration pattern
for creating extensible component factories.
Example usage:
class MyFactory(BaseFactory[MyBaseClass]):
class MyFactory(BaseFactory[MyBase]):
pass
@MyFactory.register("custom")
class CustomComponent(MyBaseClass):
class CustomComponent(MyBase):
...
component = MyFactory.create("custom", *args, **kwargs)
obj = MyFactory.create("custom", *args, **kwargs)
``create()`` filters kwargs to match the component's ``__init__``
signature so components don't need ``**kwargs`` just to absorb
unrelated parameters.
"""
_registry: Registry
_entries: Dict[str, Type[T]]
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
cls._registry = Registry()
for orig_base in getattr(cls, "__orig_bases__", ()):
if _get_origin(orig_base) is BaseFactory:
(arg,) = _get_args(orig_base)
cls._entries = {}
cls._component_base = _resolve_type(arg, cls)
return
@classmethod
def register(
cls, name: str, category: Optional[str] = None, priority: int = 0
) -> Callable[[Type[T]], Type[T]]:
"""Decorator to register a component class with optional category and priority.
def register(cls, name: str) -> Callable[[Type[T]], Type[T]]:
"""Decorator to register a component class.
Args:
name: Registration name for the component
category: Optional category for grouping components
priority: Priority for ordering (default 0)
Returns:
Decorator function that registers the component class
Raises:
TypeError: If the decorated class doesn't inherit from the base type
Validates that the decorated class inherits from the generic
type parameter ``T`` declared on the factory.
"""
def decorator(component_cls: Type[T]) -> Type[T]:
cls._validate_component(component_cls)
cls._registry.register(
name, component_cls, category=category, priority=priority
)
if name in cls._entries:
raise ValueError(f"Component '{name}' is already registered")
cls._entries[name] = component_cls
return component_cls
return decorator
@classmethod
def create(cls, name: str, *args, **kwargs) -> T:
"""Create a component instance by name.
Filters kwargs to match the component's __init__ signature,
so components don't need to declare **kwargs just to absorb
parameters meant for other components.
Args:
name: Registered name of the component
*args: Positional arguments passed to component constructor
**kwargs: Keyword arguments passed to component constructor
Returns:
Component instance
Raises:
ValueError: If the component name is not registered
"""Create a component instance by name, filtering kwargs to match
the component's ``__init__`` signature.
"""
if not cls._registry.contains(name):
entry = cls._entries.get(name)
if entry is None:
raise ValueError(
f"Unknown component: '{name}'. "
f"Supported types: {sorted(cls._registry.list_names())}"
f"Unknown component: '{name}'. Supported types: {sorted(cls._entries)}"
)
component_cls = cls._registry.get(name)
component_cls = entry
sig = inspect.signature(component_cls.__init__)
has_var_kwargs = any(
p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values()
@ -159,68 +114,32 @@ class BaseFactory(ABC, Generic[T]):
@classmethod
def _validate_component(cls, component_cls: Type[T]):
"""Validate that the component class is valid for this factory.
"""Validate the decorated class inherits from the factory's base type.
Override this method in subclasses to add custom validation.
Args:
component_cls: Component class to validate
Raises:
TypeError: If the component class is invalid
Override for custom validation beyond ``issubclass``.
"""
pass
base = cls._component_base
if base is not None and not issubclass(component_cls, base):
raise TypeError(
f"{component_cls.__name__} must inherit from {base.__name__}"
)
@classmethod
def get_component_class(cls, name: str) -> Type[T]:
"""Get the registered component class by name without instantiating it.
Args:
name: Registered name of the component
Returns:
The component class itself
Raises:
ValueError: If the component name is not registered
"""
if not cls._registry.contains(name):
"""Get the registered component class without instantiating it."""
entry = cls._entries.get(name)
if entry is None:
raise ValueError(
f"Unknown component: '{name}'. "
f"Supported types: {sorted(cls._registry.list_names())}"
f"Unknown component: '{name}'. Supported types: {sorted(cls._entries)}"
)
return cls._registry.get(name)
return entry
@classmethod
def list_registered(cls) -> list:
"""List all registered component names.
Returns:
List of registered component names
"""
return cls._registry.list_names()
def list_registered(cls) -> List[str]:
"""List all registered component names."""
return sorted(cls._entries)
@classmethod
def is_registered(cls, name: str) -> bool:
"""Check if a component name is registered.
Args:
name: Component name to check
Returns:
True if registered, False otherwise
"""
return cls._registry.contains(name)
@classmethod
def list_by_category(cls, category: str) -> List[str]:
"""List registered component names in a category."""
return cls._registry.list_by_category(category)
@classmethod
def list_by_priority(cls, reverse: bool = False) -> List[str]:
"""List registered component names sorted by priority."""
return cls._registry.list_by_priority(reverse)
__all__ = ["Registry", "BaseFactory"]
"""Check if a component name is registered."""
return name in cls._entries

View File

@ -11,12 +11,17 @@ Layers:
from astrai.inference.api import (
AnthropicMessage,
BaseToolParser,
ChatCompletionRequest,
ChatMessage,
FunctionDef,
GenContext,
MessagesRequest,
ProtocolHandler,
SimpleJsonToolParser,
StopChecker,
ToolDef,
ToolParserFactory,
get_app,
run_server,
)
@ -74,10 +79,15 @@ __all__ = [
"ProtocolHandler",
"StopChecker",
"GenContext",
"BaseToolParser",
"SimpleJsonToolParser",
"ToolParserFactory",
"OpenAIResponseBuilder",
"AnthropicResponseBuilder",
"ChatMessage",
"ChatCompletionRequest",
"FunctionDef",
"ToolDef",
"AnthropicMessage",
"MessagesRequest",
"get_app",

View File

@ -1,4 +1,4 @@
"""Inference API: protocol handler, stop checker, and FastAPI server.
"""Inference API: protocol handler, stop checker, tool parsers, and FastAPI server.
``app`` is no longer a module-level global. Use :func:`get_app` to access the
lazy singleton FastAPI instance.
@ -9,18 +9,30 @@ from astrai.inference.api.server import (
AnthropicMessage,
ChatCompletionRequest,
ChatMessage,
FunctionDef,
MessagesRequest,
ToolDef,
get_app,
run_server,
)
from astrai.inference.api.tool_parser import (
BaseToolParser,
SimpleJsonToolParser,
ToolParserFactory,
)
__all__ = [
"ProtocolHandler",
"StopChecker",
"GenContext",
"BaseToolParser",
"SimpleJsonToolParser",
"ToolParserFactory",
"AnthropicMessage",
"ChatCompletionRequest",
"ChatMessage",
"FunctionDef",
"ToolDef",
"MessagesRequest",
"get_app",
"run_server",

View File

@ -42,7 +42,6 @@ class AnthropicResponseBuilder(ResponseBuilder):
resp_id=f"msg_{uuid.uuid4().hex[:24]}",
created=int(time.time()),
model=request.model,
prompt_tokens=0,
)
stop_sequences = getattr(request, "stop_sequences", None) or []
return prompt, ctx, stop_sequences
@ -73,15 +72,17 @@ class AnthropicResponseBuilder(ResponseBuilder):
),
]
def format_chunk(self, token: str) -> str:
return sse_event(
{
"type": "content_block_delta",
"index": 0,
"delta": {"type": "text_delta", "text": token},
},
event="content_block_delta",
)
def format_chunk(self, token: str, **kwargs) -> List[str]:
return [
sse_event(
{
"type": "content_block_delta",
"index": 0,
"delta": {"type": "text_delta", "text": token},
},
event="content_block_delta",
)
]
def format_stream_end(self, ctx: GenContext, stop: StopInfo) -> List[str]:
events: List[str] = []

View File

@ -3,7 +3,7 @@
import logging
import time
import uuid
from typing import Any, Dict, List, Tuple
from typing import Any, Dict, List, Optional, Tuple, Union
from pydantic import BaseModel
@ -13,6 +13,7 @@ from astrai.inference.api.protocol import (
StopInfo,
sse_event,
)
from astrai.inference.api.tool_parser import BaseToolParser, ToolParserFactory
from astrai.inference.engine import InferenceEngine
logger = logging.getLogger(__name__)
@ -26,12 +27,37 @@ _UNSUPPORTED_PARAMS = (
)
def _resolve_tool_choice(
request: BaseModel,
) -> Union[str, Dict[str, Any]]:
tc = getattr(request, "tool_choice", None)
if tc is None:
return "auto"
if isinstance(tc, str):
return tc
if isinstance(tc, dict):
return tc
return "auto"
def _resolve_tools(request: BaseModel) -> Optional[List[Dict[str, Any]]]:
raw = getattr(request, "tools", None)
if not raw:
return None
if isinstance(raw, list):
return [t.model_dump() if hasattr(t, "model_dump") else t for t in raw]
return None
class OpenAIResponseBuilder(ResponseBuilder):
def prepare(
self, request: BaseModel, engine: InferenceEngine
) -> Tuple[str, GenContext, List[str]]:
messages = [{"role": m.role, "content": m.content} for m in request.messages]
prompt = engine.tokenizer.apply_chat_template(messages, tokenize=False)
tools = _resolve_tools(request)
prompt = engine.tokenizer.apply_chat_template(
messages, tokenize=False, tools=tools or []
)
self._resp_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
self._model = request.model
@ -42,22 +68,24 @@ class OpenAIResponseBuilder(ResponseBuilder):
default = fields[param].default if param in fields else None
if value is not None and value != default:
logger.warning(
"ChatCompletionRequest param '%s'=%r is not supported and will be ignored",
param,
value,
)
if value is not None and value != default:
logger.warning(
"ChatCompletionRequest param '%s'=%r is not supported and will be ignored",
"ChatCompletionRequest param '%s'=%r is not supported"
" and will be ignored",
param,
value,
)
self._parser: Optional[BaseToolParser] = None
if tools:
tool_choice = _resolve_tool_choice(request)
self._parser = ToolParserFactory.create(
"simple_json", tools=tools, tool_choice=tool_choice
)
self._content_started = False
ctx = GenContext(
resp_id=self._resp_id,
created=int(time.time()),
model=self._model,
prompt_tokens=0,
)
stop = request.stop
stop_sequences = (
@ -84,7 +112,82 @@ class OpenAIResponseBuilder(ResponseBuilder):
)
]
def format_chunk(self, token: str) -> str:
def format_chunk(self, token: str, **kwargs) -> List[str]:
body = kwargs.get("body", "")
if self._parser is not None:
return self._format_tool_chunk(body, **kwargs)
return [
sse_event(
{
"id": self._resp_id,
"object": "chat.completion.chunk",
"created": 0,
"model": self._model,
"choices": [
{
"index": 0,
"delta": {"content": token},
"finish_reason": None,
}
],
}
)
]
def _format_tool_chunk(self, body: str, **kwargs) -> List[str]:
deltas = self._parser.feed(
body,
current_token_ids=kwargs.get("current_token_ids"),
delta_token_ids=kwargs.get("delta_token_ids"),
)
events: List[str] = []
for d in deltas:
if "content" in d:
if not self._content_started:
events.append(self._role_chunk())
self._content_started = True
events.append(
sse_event(
{
"id": self._resp_id,
"object": "chat.completion.chunk",
"created": 0,
"model": self._model,
"choices": [
{
"index": 0,
"delta": {"content": d["content"]},
"finish_reason": None,
}
],
}
)
)
elif "tool_calls" in d:
if not self._content_started:
events.append(self._role_chunk())
self._content_started = True
events.append(
sse_event(
{
"id": self._resp_id,
"object": "chat.completion.chunk",
"created": 0,
"model": self._model,
"choices": [
{
"index": 0,
"delta": {"tool_calls": d["tool_calls"]},
"finish_reason": None,
}
],
}
)
)
return events
def _role_chunk(self) -> str:
return sse_event(
{
"id": self._resp_id,
@ -92,12 +195,19 @@ class OpenAIResponseBuilder(ResponseBuilder):
"created": 0,
"model": self._model,
"choices": [
{"index": 0, "delta": {"content": token}, "finish_reason": None}
{
"index": 0,
"delta": {"role": "assistant"},
"finish_reason": None,
}
],
}
)
def format_stream_end(self, ctx: GenContext, stop: StopInfo) -> List[str]:
finish_reason = "stop"
if self._parser is not None and self._parser.has_tool_calls:
finish_reason = "tool_calls"
return [
sse_event(
{
@ -105,7 +215,9 @@ class OpenAIResponseBuilder(ResponseBuilder):
"object": "chat.completion.chunk",
"created": ctx.created,
"model": self._model,
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
"choices": [
{"index": 0, "delta": {}, "finish_reason": finish_reason}
],
}
),
sse_event(
@ -120,6 +232,32 @@ class OpenAIResponseBuilder(ResponseBuilder):
def format_response(
self, ctx: GenContext, content: str, stop: StopInfo
) -> Dict[str, Any]:
if self._parser is not None:
parsed = self._parser.parse_complete(content)
if parsed and parsed.get("tool_calls"):
return {
"id": self._resp_id,
"object": "chat.completion",
"created": ctx.created,
"model": self._model,
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": parsed.get("content"),
"tool_calls": parsed["tool_calls"],
},
"finish_reason": "tool_calls",
}
],
"usage": {
"prompt_tokens": ctx.prompt_tokens,
"completion_tokens": ctx.completion_tokens,
"total_tokens": ctx.prompt_tokens + ctx.completion_tokens,
},
}
return {
"id": self._resp_id,
"object": "chat.completion",

View File

@ -35,7 +35,7 @@ class GenContext:
resp_id: str
created: int
model: str
prompt_tokens: int
prompt_tokens: int = 0
completion_tokens: int = 0
@ -78,8 +78,15 @@ class ResponseBuilder(ABC):
"""SSE events that open the stream."""
@abstractmethod
def format_chunk(self, token: str) -> str:
"""SSE event for a single generated token."""
def format_chunk(self, token: str, **kwargs) -> List[str]:
"""SSE events for a single generated token.
``body`` (the full accumulated text so far) is always provided
as a keyword argument. Additional keyword arguments such as
``current_token_ids`` and ``delta_token_ids`` may be included
for tool parsers that need token-level information.
Returns a list of SSE event strings (may be empty).
"""
@abstractmethod
def format_stream_end(self, ctx: GenContext, stop: StopInfo) -> List[str]:
@ -137,15 +144,25 @@ class ProtocolHandler:
body = ""
yielded = ""
matched = None
token_ids: List[int] = []
async for token in agen:
body += token
new_ids = self.engine.tokenizer.encode(token)
token_ids.extend(new_ids)
matched = checker.check(body)
if matched:
break
ctx.completion_tokens += 1
yield self.builder.format_chunk(token)
for event in self.builder.format_chunk(
token,
body=body,
current_token_ids=token_ids,
delta_token_ids=new_ids,
):
yield event
yielded += token
stop = StopInfo(matched=matched, body=body, yielded=yielded)

View File

@ -32,7 +32,20 @@ _app_instance: Optional[FastAPI] = None
class ChatMessage(BaseModel):
role: str
content: str
content: Optional[str] = None
tool_calls: Optional[List[Dict[str, Any]]] = None
tool_call_id: Optional[str] = None
class FunctionDef(BaseModel):
name: str
description: Optional[str] = None
parameters: Optional[Dict[str, Any]] = None
class ToolDef(BaseModel):
type: str = "function"
function: FunctionDef
class ChatCompletionRequest(BaseModel):
@ -51,6 +64,8 @@ class ChatCompletionRequest(BaseModel):
frequency_penalty: Optional[float] = Field(default=0.0, ge=-2.0, le=2.0)
logit_bias: Optional[Dict[int, float]] = None
user: Optional[str] = None
tools: Optional[List[ToolDef]] = None
tool_choice: Optional[Union[str, Dict[str, Any]]] = "auto"
class AnthropicMessage(BaseModel):

View File

@ -0,0 +1,325 @@
"""Tool call parsers for extracting structured tool calls from model output.
Patterned after vLLM's ToolParser abstraction. Each parser knows how to
detect and incrementally extract tool calls from raw generated text.
Subclasses may optionally consume ``token_ids`` for token-level parsing
(e.g. Harmony / VLM-style parsers).
"""
import re
import uuid
from abc import ABC, abstractmethod
from typing import Dict, List, Optional
from astrai.factory import BaseFactory
class BaseToolParser(ABC):
"""Abstract tool call parser — one instance per request.
Maintains streaming state internally so that each call to :meth:`feed`
can diff against previously emitted content.
Parameters
----------
tools : list of dict, optional
Tool definitions from the request.
tool_choice : str
``"auto"`` / ``"required"`` / ``"none"`` or a named tool choice
dict.
"""
def __init__(self, tools: Optional[List[Dict]] = None, tool_choice: str = "auto"):
self.tools = tools or []
self.tool_choice = tool_choice
@abstractmethod
def feed(
self,
body: str,
current_token_ids: Optional[List[int]] = None,
delta_token_ids: Optional[List[int]] = None,
) -> List[Dict]:
"""Feed the *full* accumulated text each step.
Returns a list of delta dicts to emit. Each delta is one of:
- ``{"content": "text"}`` plain text delta
- ``{"tool_calls": [...]}`` tool-call delta (OpenAI format)
Returns an empty list when nothing new should be emitted.
Parameters
----------
body : str
The complete accumulated generated text so far.
current_token_ids : list of int, optional
All token IDs decoded into *body* (cumulative).
delta_token_ids : list of int, optional
Only the token IDs for this chunk.
"""
@abstractmethod
def parse_complete(self, body: str) -> Optional[Dict]:
"""Parse the *complete* generated text after generation ends.
Returns ``None`` when no tool calls were found, otherwise a dict
with ``content`` (str or None) and ``tool_calls`` (list of dicts).
"""
@property
@abstractmethod
def has_tool_calls(self) -> bool:
"""True if the parser detected at least one tool call in the stream."""
class ToolParserFactory(BaseFactory["BaseToolParser"]):
pass
_TOOL_CALL_HEAD_RE = re.compile(r'\{\s*"name"\s*:')
def _scan_json(text: str, start: int = 0):
"""Scan for a complete JSON object starting at *start*.
Returns ``(end, complete)`` where *end* is one-past the closing
brace (or ``len(text)`` if unclosed), and *complete* is a bool.
"""
depth = 0
in_string = False
escape = False
for i in range(start, len(text)):
c = text[i]
if escape:
escape = False
continue
if c == "\\":
escape = True
continue
if c == '"':
in_string = not in_string
continue
if in_string:
continue
if c == "{":
depth += 1
elif c == "}":
depth -= 1
if depth == 0:
return i + 1, True
return len(text), False
def _parse_tool_call_json(json_str: str, complete: bool):
"""Extract *name* and *arguments* from a tool-call JSON string.
Returns ``(name, args, valid)``.
"""
name_match = re.search(r'"name"\s*:\s*"([^"]*)"', json_str)
if not name_match:
return None, "", False
name = name_match.group(1)
args_match = re.search(r'"arguments"\s*:\s*(.*)', json_str, re.DOTALL)
if not args_match:
return name, "", True
raw = args_match.group(1).rstrip()
if complete and raw.endswith("}"):
raw = raw[:-1].rstrip()
if raw.startswith("{"):
inner = raw[1:].rstrip()
if inner.endswith("}"):
inner = inner[:-1].rstrip()
raw = inner
return name, raw, True
def _find_tool_calls(text: str, start_pos: int = 0):
"""Find all complete ``{...}`` tool-call objects in *text*.
Returns a list of dicts with keys *start*, *end*, *name*, *args*,
*complete*.
"""
results = []
pos = start_pos
while True:
brace = text.find("{", pos)
if brace == -1:
break
end, complete = _scan_json(text, brace)
if not complete:
break
json_str = text[brace:end]
if not _TOOL_CALL_HEAD_RE.search(json_str):
pos = end
continue
name, args, valid = _parse_tool_call_json(json_str, complete=True)
if not valid or name is None:
pos = end
continue
results.append(
{
"start": brace,
"end": end,
"name": name,
"args": args,
"complete": True,
}
)
pos = end
return results
def _find_partial_tool_call(text: str, start_pos: int = 0):
"""Find one incomplete (still-generating) tool-call JSON object."""
brace = text.find("{", start_pos)
if brace == -1:
return None
json_str = text[brace:]
if not _TOOL_CALL_HEAD_RE.search(json_str):
return None
name, args, valid = _parse_tool_call_json(json_str, complete=False)
if not valid or name is None:
return None
return {
"start": brace,
"name": name,
"args": args,
"complete": False,
}
@ToolParserFactory.register("simple_json")
class SimpleJsonToolParser(BaseToolParser):
"""Parser for models that output tool calls as plain JSON objects.
Detects ``{"name": "<func>", "arguments": {...}}`` anywhere in the
generated text. Handles single and (non-overlapping) multiple tool
calls. Text preceding the first tool call is emitted as plain
``content`` deltas.
"""
def __init__(self, tools=None, tool_choice="auto"):
super().__init__(tools, tool_choice)
self._emitted_content_len = 0
self._tc_state: List[Dict] = []
self._has_tool_calls = False
# -------------------------------------------------------------- feed
def feed(
self,
body: str,
current_token_ids: Optional[List[int]] = None,
delta_token_ids: Optional[List[int]] = None,
) -> List[Dict]:
deltas: List[Dict] = []
completed = _find_tool_calls(body)
if not completed:
partial = _find_partial_tool_call(body)
if not partial:
return self._emit_plain_content(body, deltas)
all_tcs = [partial]
else:
all_tcs = completed
partial = _find_partial_tool_call(body, completed[-1]["end"])
if partial:
all_tcs = completed + [partial]
first_start = all_tcs[0]["start"]
if first_start > self._emitted_content_len:
content = body[self._emitted_content_len : first_start]
self._emitted_content_len = first_start
if content:
deltas.append({"content": content})
for i, tc in enumerate(all_tcs):
if i >= len(self._tc_state):
self._tc_state.append(
{
"id": f"call_{uuid.uuid4().hex[:12]}",
"name_emitted": False,
"args_emitted_len": 0,
}
)
self._has_tool_calls = True
st = self._tc_state[i]
if not st["name_emitted"]:
st["name_emitted"] = True
deltas.append(
{
"tool_calls": [
{
"index": i,
"id": st["id"],
"type": "function",
"function": {"name": tc["name"], "arguments": ""},
}
]
}
)
new_args = tc["args"]
if len(new_args) > st["args_emitted_len"]:
diff = new_args[st["args_emitted_len"] :]
st["args_emitted_len"] = len(new_args)
deltas.append(
{
"tool_calls": [
{
"index": i,
"function": {"arguments": diff},
}
]
}
)
return deltas
def _emit_plain_content(self, body: str, deltas: List[Dict]) -> List[Dict]:
new_content = body[self._emitted_content_len :]
if new_content:
self._emitted_content_len = len(body)
deltas.append({"content": new_content})
return deltas
# -------------------------------------------------------- complete
def parse_complete(self, body: str) -> Optional[Dict]:
completed = _find_tool_calls(body)
if not completed:
return None
content = body[: completed[0]["start"]].strip() or None
tool_calls = []
for i, tc in enumerate(completed):
tool_calls.append(
{
"id": f"call_{uuid.uuid4().hex[:12]}",
"type": "function",
"function": {
"name": tc["name"],
"arguments": tc["args"],
},
}
)
return {"content": content, "tool_calls": tool_calls}
@property
def has_tool_calls(self) -> bool:
return self._has_tool_calls

View File

@ -29,6 +29,7 @@ class BaseSamplingStrategy(ABC):
Returns:
Transformed logits tensor.
"""
raise NotImplementedError
class TemperatureStrategy(BaseSamplingStrategy):
@ -41,7 +42,7 @@ class TemperatureStrategy(BaseSamplingStrategy):
def __init__(self, temperature: Union[float, Tensor] = 1.0):
self.temperature = temperature
def apply(self, logits, filter_value=-float("inf")):
def apply(self, logits: Tensor, filter_value: float = -float("inf")) -> Tensor:
t = self.temperature
if isinstance(t, Tensor):
t = t.to(logits.device, non_blocking=True).view(-1, 1)
@ -63,7 +64,7 @@ class TopKStrategy(BaseSamplingStrategy):
def __init__(self, top_k: Union[int, Tensor] = 0):
self.top_k = top_k
def apply(self, logits, filter_value=-float("inf")):
def apply(self, logits: Tensor, filter_value: float = -float("inf")) -> Tensor:
tk = self.top_k
if isinstance(tk, Tensor):
tk = tk.to(logits.device, non_blocking=True).long().clamp(min=0)
@ -100,7 +101,9 @@ class TopPStrategy(BaseSamplingStrategy):
def __init__(self, top_p: Union[float, Tensor] = 1.0):
self.top_p = top_p
def _apply(self, logits, top_p, filter_value):
def _apply(
self, logits: Tensor, top_p: Union[float, Tensor], filter_value: float
) -> Tensor:
sorted_logits, sorted_indices = torch.sort(logits, descending=True, dim=-1)
cum_probs = torch.cumsum(torch.softmax(sorted_logits, dim=-1), dim=-1)
remove = cum_probs > top_p
@ -111,7 +114,7 @@ class TopPStrategy(BaseSamplingStrategy):
logits[mask] = filter_value
return logits
def apply(self, logits, filter_value=-float("inf")):
def apply(self, logits: Tensor, filter_value: float = -float("inf")) -> Tensor:
tp = self.top_p
if isinstance(tp, Tensor):
tp = tp.to(logits.device, non_blocking=True)
@ -142,7 +145,7 @@ class SamplingPipeline(BaseSamplingStrategy):
def __init__(self, strategies: List[BaseSamplingStrategy]):
self.strategies = strategies
def apply(self, logits, filter_value=-float("inf")):
def apply(self, logits: Tensor, filter_value: float = -float("inf")) -> Tensor:
for strategy in self.strategies:
logits = strategy.apply(logits, filter_value)
return logits

View File

@ -24,9 +24,7 @@ def repeat_kv(x: Tensor, n_rep: int) -> Tensor:
class AttnFactory(BaseFactory[nn.Module]):
@classmethod
def create(cls, attn_type: str, **kwargs) -> nn.Module:
return super().create(attn_type, **kwargs)
pass
@AttnFactory.register("gqa")

View File

@ -1,3 +1,5 @@
import math
import torch
import torch.nn as nn
import torch.nn.functional as F
@ -8,9 +10,14 @@ class Embedding(nn.Module):
def __init__(self, vocab_size: int, embedding_dim: int):
super().__init__()
self.weight = nn.Parameter(torch.empty((vocab_size, embedding_dim)))
self.neftune_noise_alpha = 0.0
def reset_parameters(self):
nn.init.normal_(self.weight, mean=0.0, std=0.02)
def forward(self, x: Tensor) -> Tensor:
return F.embedding(x, self.weight)
out = F.embedding(x, self.weight)
if self.training and self.neftune_noise_alpha > 0.0:
eps = self.neftune_noise_alpha / math.sqrt(out.size(1))
out = out + eps * torch.randn_like(out)
return out

View File

@ -8,9 +8,7 @@ from astrai.model.components.linear import Linear
class FFNFactory(BaseFactory[nn.Module]):
@classmethod
def create(cls, ffn_type: str, dim: int, dim_ffn: int, **kwargs) -> nn.Module:
return super().create(ffn_type, dim, dim_ffn, **kwargs)
pass
@FFNFactory.register("mlp")

View File

@ -3,12 +3,30 @@ from astrai.preprocessing.builder import (
MaskBuilderFactory,
SectionedMaskBuilder,
)
from astrai.preprocessing.packing import (
PackingStrategy,
PackingStrategyFactory,
)
from astrai.preprocessing.pipeline import Pipeline, filter_by_length
from astrai.preprocessing.position_id import (
PositionIdStrategy,
PositionIdStrategyFactory,
)
from astrai.preprocessing.writer import (
StoreWriter,
StoreWriterFactory,
)
__all__ = [
"BaseMaskBuilder",
"MaskBuilderFactory",
"SectionedMaskBuilder",
"PackingStrategy",
"PackingStrategyFactory",
"Pipeline",
"PositionIdStrategy",
"PositionIdStrategyFactory",
"SectionedMaskBuilder",
"StoreWriter",
"StoreWriterFactory",
"filter_by_length",
]

View File

@ -1,8 +1,8 @@
"""Mask building strategies for preprocessing pipeline.
"""Mask building for preprocessing pipeline.
The single :class:`SectionedMaskBuilder` handles all input formats
(single-sequence / DPO / GRPO) via declarative config: ``input.sections``
for single-output or ``input.sources`` for multi-output.
:class:`SectionRenderer` converts section specs into token ids and loss
masks (template / text / value extraction). :class:`SectionedMaskBuilder`
orchestrates single-output / multi-output (DPO / GRPO) assembly.
"""
from abc import ABC, abstractmethod
@ -11,27 +11,6 @@ from typing import Optional
from astrai.factory import BaseFactory
class BaseMaskBuilder(ABC):
"""Convert a JSONL item into token ids and optional loss_mask."""
@abstractmethod
def build(self, item: dict, config, tokenizer) -> Optional[dict]:
"""Build ``{ids, loss_mask?, domain}`` from a JSONL record.
Returns ``None`` to skip the item entirely.
"""
...
class MaskBuilderFactory(BaseFactory["BaseMaskBuilder"]):
@classmethod
def _validate_component(cls, component_cls: type):
if not issubclass(component_cls, BaseMaskBuilder):
raise TypeError(
f"{component_cls.__name__} must inherit from BaseMaskBuilder"
)
def _extract_domain(item: dict, domain_key: Optional[str]) -> str:
if not domain_key:
return "__default__"
@ -40,142 +19,15 @@ def _extract_domain(item: dict, domain_key: Optional[str]) -> str:
def _resolve_action(action: str, role: str, config) -> str:
"""Resolve action to "train" or "mask".
- ``"train"`` / ``"mask"`` literal
- ``"$role"`` look up ``role`` in ``config.mask``, fall back to ``config.mask_default``
"""
if action == "$role":
return config.mask.get(role, config.mask_default)
return action
@MaskBuilderFactory.register("sectioned")
class SectionedMaskBuilder(BaseMaskBuilder):
"""Config-driven builder supporting single and multi-output modes.
class SectionRenderer:
"""Render section specs into ``(ids, loss_mask)`` tuples."""
Single-output (backward-compatible)::
{"input": {"sections": [
{"field": "messages", "action": "$role", "template": true}
]}}
{"sequence": [...], "loss_mask": [...], "domain": "..."}
Multi-output (DPO / GRPO)::
{"input": {"sources": {
"chosen": {"sections": [
{"field": "chosen", "action": "$role", "template": true}
]},
"rejected": {"sections": [
{"field": "rejected", "action": "$role", "template": true}
]}
}}}
{"chosen": [...], "chosen_mask": [...],
"rejected": [...], "rejected_mask": [...], "domain": "..."}
Output spec fields::
sections list of section specs (same format as single-output)
list_field True when the JSONL field holds a list of values to
tokenise individually and concatenate (GRPO responses)
mask_key explicit output key for the loss mask
(default: ``"{output_key}_mask"``)
dtype explicit tensor dtype for this output key
(default: "int32")
"""
def build(self, item: dict, config, tokenizer) -> Optional[dict]:
sources_spec = getattr(config.input, "sources", None)
if sources_spec:
return self._build_multi(item, sources_spec, config, tokenizer)
return self._build_single(item, config, tokenizer)
def _build_single(self, item: dict, config, tokenizer) -> Optional[dict]:
sections = config.input.sections
if not sections:
return None
ids, mask = self._process_sections(
item, sections, config, tokenizer, is_top_level=True
)
if ids is None:
return None
result: dict = {
"sequence": ids,
"domain": _extract_domain(item, config.output.domain_key),
}
if not all(m == 1 for m in mask):
result["loss_mask"] = mask
return result
def _build_multi(
self, item: dict, sources_spec: dict, config, tokenizer
) -> Optional[dict]:
result: dict = {}
any_output = False
for output_key, spec in sources_spec.items():
sections = spec.get("sections", [])
if not sections:
continue
if self._is_value_section(sections):
ids = self._extract_raw_value(item, sections)
if ids is None:
continue
result[output_key] = ids
any_output = True
continue
list_field = spec.get("list_field", False)
mask_key = spec.get("mask_key", f"{output_key}_mask")
if list_field:
ids, mask = self._process_list_field(item, sections, config, tokenizer)
else:
ids, mask = self._process_sections(
item, sections, config, tokenizer, is_top_level=True
)
if ids is None:
continue
result[output_key] = ids
if not all(m == 1 for m in mask):
result[mask_key] = mask
elif "mask_key" in spec:
result[mask_key] = mask
any_output = True
if not any_output:
return None
result["domain"] = _extract_domain(item, config.output.domain_key)
return result
@staticmethod
def _is_value_section(sections: list) -> bool:
return len(sections) == 1 and sections[0].get("action") == "value"
@staticmethod
def _extract_raw_value(item: dict, sections: list):
"""Extract a raw value from a JSONL field without tokenisation.
Used for GRPO rewards where the field contains float values.
"""
sec = sections[0]
field = sec["field"]
raw = item.get(field)
if raw is None:
return None
if isinstance(raw, list):
return [float(v) for v in raw]
return [float(raw)]
def _process_sections(
def process_sections(
self,
item: dict,
sections: list,
@ -184,10 +36,6 @@ class SectionedMaskBuilder(BaseMaskBuilder):
*,
is_top_level: bool = False,
):
"""Process a list of sections into ``(ids, loss_mask)``.
Returns ``(None, None)`` if the item should be skipped.
"""
all_ids: list[int] = []
loss_mask: list[int] = []
@ -210,13 +58,13 @@ class SectionedMaskBuilder(BaseMaskBuilder):
)
if use_template:
success = self._append_template_section(
success = self._append_template(
item, field, action, tokenizer, config, all_ids, loss_mask
)
if not success:
continue
else:
success = self._append_text_section(
success = self._append_text(
item,
field,
action,
@ -244,7 +92,70 @@ class SectionedMaskBuilder(BaseMaskBuilder):
return all_ids, loss_mask
def _append_template_section(
def process_list_field(self, item: dict, sections: list, config, tokenizer):
all_ids: list[int] = []
loss_mask: list[int] = []
for sec in sections:
field = sec["field"]
action = sec["action"]
use_template = sec.get("template", False)
values = item.get(field)
if not isinstance(values, list):
continue
for val in values:
if use_template:
if isinstance(val, list):
wrapper = {field: val}
self._append_template(
wrapper,
field,
action,
tokenizer,
config,
all_ids,
loss_mask,
)
else:
wrapper = {field: str(val)}
self._append_text(
wrapper,
field,
action,
tokenizer,
False,
False,
config,
all_ids,
loss_mask,
)
max_len = config.preprocessing.max_seq_len
all_ids = all_ids[:max_len]
loss_mask = loss_mask[: len(all_ids)]
if not all_ids:
return None, None
return all_ids, loss_mask
@staticmethod
def is_value_section(sections: list) -> bool:
return len(sections) == 1 and sections[0].get("action") == "value"
@staticmethod
def extract_raw_value(item: dict, sections: list):
sec = sections[0]
field = sec["field"]
raw = item.get(field)
if raw is None:
return None
if isinstance(raw, list):
return [float(v) for v in raw]
return [float(raw)]
def _append_template(
self, item, field, action, tokenizer, config, all_ids, loss_mask
):
messages = item.get(field)
@ -262,7 +173,7 @@ class SectionedMaskBuilder(BaseMaskBuilder):
loss_mask.extend([val] * len(ids))
return True
def _append_text_section(
def _append_text(
self,
item,
field,
@ -289,50 +200,116 @@ class SectionedMaskBuilder(BaseMaskBuilder):
loss_mask.extend([val] * len(ids))
return True
def _process_list_field(self, item: dict, sections: list, config, tokenizer):
all_ids: list[int] = []
loss_mask: list[int] = []
for sec in sections:
field = sec["field"]
action = sec["action"]
use_template = sec.get("template", False)
class BaseMaskBuilder(ABC):
"""Convert a JSONL item into token ids and optional loss_mask."""
values = item.get(field)
if not isinstance(values, list):
@abstractmethod
def build(self, item: dict, config, tokenizer) -> Optional[dict]: ...
class MaskBuilderFactory(BaseFactory["BaseMaskBuilder"]):
pass
@MaskBuilderFactory.register("sectioned")
class SectionedMaskBuilder(BaseMaskBuilder):
"""Config-driven builder supporting single and multi-output modes.
Single-output::
{"input": {"sections": [
{"field": "messages", "action": "$role", "template": true}
]}}
{"sequence": [...], "loss_mask": [...], "domain": "..."}
Multi-output (DPO / GRPO)::
{"input": {"sources": {
"chosen": {"sections": [{"field": "chosen", "action": "$role", "template": true}]},
"rejected": {"sections": [{"field": "rejected", "action": "$role", "template": true}]},
}}}
{"chosen": [...], "chosen_mask": [...], "rejected": [...], "rejected_mask": [...], "domain": "..."}
Output spec fields::
sections list of section specs (same format as single-output)
list_field True when JSONL field holds a list (GRPO responses)
mask_key explicit loss-mask output key (default: ``"{output_key}_mask"``)
"""
def __init__(self):
self.renderer = SectionRenderer()
def build(self, item: dict, config, tokenizer) -> Optional[dict]:
sources_spec = getattr(config.input, "sources", None)
if sources_spec:
return self._build_multi(item, sources_spec, config, tokenizer)
return self._build_single(item, config, tokenizer)
def _build_single(self, item: dict, config, tokenizer) -> Optional[dict]:
sections = config.input.sections
if not sections:
return None
ids, mask = self.renderer.process_sections(
item, sections, config, tokenizer, is_top_level=True
)
if ids is None:
return None
result: dict = {
"sequence": ids,
"domain": _extract_domain(item, config.output.domain_key),
}
if not all(m == 1 for m in mask):
result["loss_mask"] = mask
return result
def _build_multi(
self, item: dict, sources_spec: dict, config, tokenizer
) -> Optional[dict]:
result: dict = {}
any_output = False
for output_key, spec in sources_spec.items():
sections = spec.get("sections", [])
if not sections:
continue
for val in values:
if use_template:
if isinstance(val, list):
wrapper = {field: val}
self._append_template_section(
wrapper,
field,
action,
tokenizer,
config,
all_ids,
loss_mask,
)
else:
wrapper = {field: str(val)}
self._append_text_section(
wrapper,
field,
action,
tokenizer,
False,
False,
config,
all_ids,
loss_mask,
)
if self.renderer.is_value_section(sections):
ids = self.renderer.extract_raw_value(item, sections)
if ids is None:
continue
result[output_key] = ids
any_output = True
continue
max_len = config.preprocessing.max_seq_len
all_ids = all_ids[:max_len]
loss_mask = loss_mask[: len(all_ids)]
list_field = spec.get("list_field", False)
mask_key = spec.get("mask_key", f"{output_key}_mask")
if not all_ids:
return None, None
return all_ids, loss_mask
if list_field:
ids, mask = self.renderer.process_list_field(
item, sections, config, tokenizer
)
else:
ids, mask = self.renderer.process_sections(
item, sections, config, tokenizer, is_top_level=True
)
if ids is None:
continue
result[output_key] = ids
if not all(m == 1 for m in mask):
result[mask_key] = mask
elif "mask_key" in spec:
result[mask_key] = mask
any_output = True
if not any_output:
return None
result["domain"] = _extract_domain(item, config.output.domain_key)
return result

View File

@ -0,0 +1,101 @@
"""Sequence packing strategies for shard-level reordering and truncation.
Each strategy receives the accumulated ``{key: [list of token lists]}``
dict for a shard and returns a reordered / truncated version. The
pipeline later flattens the result into contiguous tensors.
"""
from abc import ABC, abstractmethod
from collections import defaultdict
from typing import Dict, List, Tuple
from astrai.factory import BaseFactory
def _truncate(seq: List[int], max_len: int, mode: str) -> List[int]:
if len(seq) <= max_len:
return seq
if mode == "keep_end":
return seq[-max_len:]
return seq[:max_len]
class PackingStrategy(ABC):
"""Reorder and truncate sequences within a shard."""
@abstractmethod
def apply(
self,
keys: Dict[str, List[List[int]]],
max_packed_len: int,
truncation_mode: str,
) -> Dict[str, List[List[int]]]:
raise NotImplementedError
class PackingStrategyFactory(BaseFactory["PackingStrategy"]):
pass
@PackingStrategyFactory.register("simple")
class SimplePacking(PackingStrategy):
def apply(
self,
keys: Dict[str, List[List[int]]],
max_packed_len: int,
truncation_mode: str,
) -> Dict[str, List[List[int]]]:
return {
k: [_truncate(v, max_packed_len, truncation_mode) for v in vals]
for k, vals in keys.items()
}
@PackingStrategyFactory.register("bfd")
class BFDPacking(PackingStrategy):
def apply(
self,
keys: Dict[str, List[List[int]]],
max_packed_len: int,
truncation_mode: str,
) -> Dict[str, List[List[int]]]:
sequences = keys.get("sequence", [])
if not sequences:
return keys
plan = self._plan(sequences, max_packed_len)
reordered: dict = defaultdict(list)
for orig_idx, _ in plan:
for k, vals in keys.items():
reordered[k].append(
_truncate(vals[orig_idx], max_packed_len, truncation_mode)
)
return dict(reordered)
@staticmethod
def _plan(sequences: List[List[int]], max_packed_len: int) -> List[Tuple[int, int]]:
n = len(sequences)
order = sorted(range(n), key=lambda i: len(sequences[i]), reverse=True)
bins: List[List[int]] = []
bin_lengths: List[int] = []
for orig_idx in order:
seq_len = min(len(sequences[orig_idx]), max_packed_len)
best_bin = None
best_remain = max_packed_len + 1
for i, bl in enumerate(bin_lengths):
remain = max_packed_len - bl
if seq_len <= remain < best_remain:
best_remain = remain
best_bin = i
if best_bin is not None:
bins[best_bin].append(orig_idx)
bin_lengths[best_bin] += seq_len
else:
bins.append([orig_idx])
bin_lengths.append(seq_len)
plan: List[Tuple[int, int]] = []
for bin_indices in bins:
for orig_idx in bin_indices:
plan.append((orig_idx, min(len(sequences[orig_idx]), max_packed_len)))
return plan

View File

@ -1,21 +1,25 @@
"""Config-driven JSONL preprocessing pipeline.
Composes a :class:`BaseMaskBuilder` (selected by ``input.type``) with
sharding and flush to ``.h5`` / ``.bin`` storage.
sharding and flush to ``.h5`` / ``.bin`` storage. Packing, position-id
generation and storage writing are each delegated to pluggable strategies,
dispatched by configuration keys.
"""
import json
import os
from collections import defaultdict
from itertools import chain
from typing import List, Optional, Tuple
from typing import Dict, List, Optional
import torch
import tqdm
from astrai.config.preprocess_config import PipelineConfig
from astrai.dataset.storage import save_bin, save_h5
from astrai.preprocessing.builder import SectionedMaskBuilder
from astrai.preprocessing.builder import MaskBuilderFactory
from astrai.preprocessing.packing import PackingStrategyFactory
from astrai.preprocessing.position_id import PositionIdStrategyFactory
from astrai.preprocessing.writer import StoreWriterFactory
from astrai.tokenize import AutoTokenizer
_STR_TO_DTYPE: dict[str, torch.dtype] = {
@ -35,71 +39,12 @@ def filter_by_length(text: str, min_len: int = 50, max_len: int = 2_000_000) ->
return min_len <= len(text) <= max_len
def _truncate(seq: list, max_len: int, mode: str) -> list:
if len(seq) <= max_len:
return seq
if mode == "keep_end":
return seq[-max_len:]
return seq[:max_len]
def pack_sequences(
sequences: List[list],
max_packed_len: int,
strategy: str,
truncation_mode: str,
) -> List[Tuple[int, int]]:
"""Pack *sequences* into bins and return a reorder plan.
Returns a list of ``(orig_idx, truncated_length)`` in flush order.
All keys (sequence, loss_mask, ) must be reordered and truncated
identically according to this plan.
Supported *strategy* values:
- ``"simple"``: sequential, no reordering.
- ``"bfd"``: best-fit decreasing bin packing.
"""
n = len(sequences)
if strategy == "simple":
return [(i, min(len(sequences[i]), max_packed_len)) for i in range(n)]
order = sorted(range(n), key=lambda i: len(sequences[i]), reverse=True)
bins: List[List[int]] = []
bin_lengths: List[int] = []
for orig_idx in order:
seq_len = min(len(sequences[orig_idx]), max_packed_len)
best_bin = None
best_remain = max_packed_len + 1
for i, bl in enumerate(bin_lengths):
remain = max_packed_len - bl
if seq_len <= remain < best_remain:
best_remain = remain
best_bin = i
if best_bin is not None:
bins[best_bin].append(orig_idx)
bin_lengths[best_bin] += seq_len
else:
bins.append([orig_idx])
bin_lengths.append(seq_len)
plan: List[Tuple[int, int]] = []
for bin_indices in bins:
for orig_idx in bin_indices:
plan.append((orig_idx, min(len(sequences[orig_idx]), max_packed_len)))
return plan
class Pipeline:
"""Tokenization pipeline driven by a declarative :class:`PipelineConfig`.
Usage::
config = PipelineConfig.from_json("sft_pipeline.json")
config = PipelineConfig.from_file("sft_pipeline.json")
Pipeline(config, ["data.jsonl"], output_dir="out", tokenizer_path="params").run()
"""
@ -116,7 +61,14 @@ class Pipeline:
self.output_dir = output_dir
self.tokenizer_path = tokenizer_path
self.mask_builder = SectionedMaskBuilder()
self.mask_builder = MaskBuilderFactory.create("sectioned")
self._packer = PackingStrategyFactory.create(
config.preprocessing.packing_strategy
)
self._position_id = PositionIdStrategyFactory.create(
config.output.position_ids_mode
)
self._writer = StoreWriterFactory.create(config.output.storage_format)
def transform(self, item: dict) -> Optional[dict]:
return self.mask_builder.build(item, self.config, self._tokenizer)
@ -168,8 +120,6 @@ class Pipeline:
if total_tokens > 0:
self._flush(domains, shard_idx)
print(f"Done. {count} documents tokenized.")
@staticmethod
def _primary_ids(result: dict) -> list:
"""Return the first list-valued entry in *result* as the primary id
@ -203,27 +153,11 @@ class Pipeline:
def _flush(self, domains, shard_idx):
for domain, keys in domains.items():
idx = shard_idx[domain]
chunk_dir = os.path.join(self.output_dir, domain)
pp = self.config.preprocessing
if pp.packing_strategy != "simple" and "sequence" in keys:
plan = pack_sequences(
keys["sequence"],
pp.max_packed_len,
pp.packing_strategy,
pp.truncation_mode,
)
reordered = defaultdict(list)
for orig_idx, truncated_len in plan:
for k, vals in keys.items():
reordered[k].append(
_truncate(
vals[orig_idx], pp.max_packed_len, pp.truncation_mode
)
)
keys = reordered
keys = self._packer.apply(dict(keys), pp.max_packed_len, pp.truncation_mode)
tensors = {}
tensors: Dict[str, List[torch.Tensor]] = {}
for key, ids_list in keys.items():
dt = _STR_TO_DTYPE.get(
self.config.output.dtype.get(key, "int32"), torch.int32
@ -232,24 +166,13 @@ class Pipeline:
torch.tensor(list(chain.from_iterable(ids_list)), dtype=dt)
]
pid_mode = self.config.output.position_ids_mode
if pid_mode and pid_mode != "none" and "sequence" in tensors:
pos_ids = []
if pid_mode == "doc_reset":
for item in keys["sequence"]:
pos_ids.extend(range(len(item)))
else:
total = sum(len(item) for item in keys["sequence"])
pos_ids = list(range(total))
pos_ids = self._position_id.generate(keys.get("sequence", []))
if pos_ids:
tensors["position_ids"] = [torch.tensor(pos_ids, dtype=torch.int32)]
shard_path = os.path.join(chunk_dir, f"shard_{idx:04d}")
fmt = self.config.output.storage_format
if fmt == "bin":
save_bin(shard_path, tensors)
else:
save_h5(chunk_dir, f"data_{idx:04d}", tensors)
self._writer.save(self.output_dir, domain, idx, tensors)
shard_idx[domain] = idx + 1
first_key = "sequence" if "sequence" in tensors else next(iter(tensors))
tqdm.tqdm.write(
f" saved {domain}/shard_{idx:04d} "

View File

@ -0,0 +1,46 @@
"""Position-id generation strategies for packed sequences.
Each strategy takes the list of per-document token sequences after packing
and returns a flat list of position ids (same total length as all
sequences combined). The pipeline wraps the result into a tensor and
attaches it as ``position_ids``.
"""
from abc import ABC, abstractmethod
from typing import List
from astrai.factory import BaseFactory
class PositionIdStrategy(ABC):
"""Generate ``position_ids`` for packed sequences."""
@abstractmethod
def generate(self, sequences: List[List[int]]) -> List[int]:
raise NotImplementedError
class PositionIdStrategyFactory(BaseFactory["PositionIdStrategy"]):
pass
@PositionIdStrategyFactory.register("none")
class NoPositionId(PositionIdStrategy):
def generate(self, sequences: List[List[int]]) -> List[int]:
return []
@PositionIdStrategyFactory.register("doc_reset")
class DocResetPositionId(PositionIdStrategy):
def generate(self, sequences: List[List[int]]) -> List[int]:
pos_ids = []
for seq in sequences:
pos_ids.extend(range(len(seq)))
return pos_ids
@PositionIdStrategyFactory.register("continuous")
class ContinuousPositionId(PositionIdStrategy):
def generate(self, sequences: List[List[int]]) -> List[int]:
total = sum(len(seq) for seq in sequences)
return list(range(total))

View File

@ -0,0 +1,47 @@
"""Storage writer strategies for pipeline output.
The :class:`StoreWriter` abstraction decouples the pipeline from the
concrete storage format (bin / h5). The pipeline builds a ``{key:
List[Tensor]}`` dict and delegates the write to the writer selected
by ``output.storage_format``.
"""
import os
from abc import ABC, abstractmethod
from typing import Dict, List
import torch
from astrai.dataset.storage import save_bin, save_h5
from astrai.factory import BaseFactory
class StoreWriter(ABC):
"""Write pre-tokenized tensors to disk in a format-specific way."""
@abstractmethod
def save(
self,
output_dir: str,
domain: str,
shard_idx: int,
tensors: Dict[str, List[torch.Tensor]],
) -> None: ...
class StoreWriterFactory(BaseFactory["StoreWriter"]):
pass
@StoreWriterFactory.register("bin")
class BinWriter(StoreWriter):
def save(self, output_dir, domain, shard_idx, tensors):
shard_path = os.path.join(output_dir, domain, f"shard_{shard_idx:04d}")
save_bin(shard_path, tensors)
@StoreWriterFactory.register("h5")
class H5Writer(StoreWriter):
def save(self, output_dir, domain, shard_idx, tensors):
chunk_dir = os.path.join(output_dir, domain)
save_h5(chunk_dir, f"data_{shard_idx:04d}", tensors)

View File

@ -3,7 +3,7 @@ import json
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, Union
from typing import Any, Dict, Optional, Union
import safetensors.torch as st
import torch
@ -180,3 +180,22 @@ class Checkpoint:
extra=extra,
config=config,
)
@classmethod
def load_any(cls, save_dir: str, broadcast: bool = False) -> Optional["Checkpoint"]:
save_path = Path(save_dir)
meta_path = save_path / _META_FILE
weights_path = save_path / _WEIGHTS_FILE
if meta_path.exists():
return cls.load(save_dir, broadcast=broadcast)
if weights_path.exists():
state_dict = load_state_dict(weights_path, broadcast=broadcast)
config = {}
config_path = save_path / _CONFIG_FILE
if config_path.exists():
config = load_json(config_path, broadcast)
return cls(state_dict=state_dict, config=config)
return None

View File

@ -2,7 +2,7 @@
import math
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Type
from typing import Any, Dict, List
from torch.optim.lr_scheduler import LRScheduler
@ -31,7 +31,6 @@ class SchedulerFactory(BaseFactory["BaseScheduler"]):
"""Factory class for creating learning rate schedulers.
Supports decorator-based registration for extensible scheduler types.
Also supports creation from ScheduleConfig objects.
Example usage:
@SchedulerFactory.register("custom")
@ -41,33 +40,6 @@ class SchedulerFactory(BaseFactory["BaseScheduler"]):
scheduler = SchedulerFactory.create("custom", optimizer, **kwargs)
"""
@classmethod
def _validate_component(cls, scheduler_cls: Type[BaseScheduler]):
"""Validate that the scheduler class inherits from BaseScheduler."""
if not issubclass(scheduler_cls, BaseScheduler):
raise TypeError(f"{scheduler_cls.__name__} must inherit from BaseScheduler")
@classmethod
def create(
cls, optimizer, schedule_type: str = "none", **kwargs
) -> "BaseScheduler":
"""Create a scheduler instance by type name.
Args:
optimizer: PyTorch optimizer
schedule_type: Type of scheduler ("cosine", "sgdr")
**kwargs: Arguments passed to the scheduler constructor
Returns:
Scheduler instance
"""
return super().create(schedule_type, optimizer, **kwargs)
@classmethod
def available_types(cls) -> list:
"""Return list of registered scheduler type names."""
return cls.list_registered()
# ----------- Scheduler implementations -----------

View File

@ -1,7 +1,7 @@
"""Training strategy implementations with factory pattern."""
from abc import ABC, abstractmethod
from typing import Any, Callable, Dict, Union
from typing import Callable, Dict, Union
import torch
import torch.nn as nn
@ -11,7 +11,9 @@ from torch import Tensor
from astrai.factory import BaseFactory
def create_ref_model(model_fn, state_dict: dict) -> nn.Module:
def create_ref_model(
model_fn: Callable[[], nn.Module], state_dict: Dict[str, Tensor]
) -> nn.Module:
"""Create a frozen reference model from model_fn + full state dict."""
ref_model = model_fn()
ref_model.load_state_dict(state_dict)
@ -20,7 +22,7 @@ def create_ref_model(model_fn, state_dict: dict) -> nn.Module:
return ref_model
def move_to_device(batch: Dict[str, Tensor], device: str) -> Any:
def move_to_device(batch: Dict[str, Tensor], device: str) -> Dict[str, Tensor]:
"""Move batch tensors to specified device with non-blocking transfer."""
return {key: value.to(device, non_blocking=True) for key, value in batch.items()}
@ -30,7 +32,7 @@ def get_logprobs(
input_ids: Tensor,
mask: Tensor,
reduction: str,
):
) -> Tensor:
"""Compute token-wise log probabilities from model outputs.
Args:
@ -88,7 +90,10 @@ class BaseStrategy(ABC):
"""Abstract base class for training strategies."""
def __init__(
self, model: Union[Callable[..., Dict[str, Tensor]]], device: str, **kwargs
self,
model: Union[nn.Module, Callable[..., Dict[str, Tensor]]],
device: str,
**kwargs,
):
self.model = model
self.device = device
@ -127,32 +132,6 @@ class StrategyFactory(BaseFactory["BaseStrategy"]):
strategy = StrategyFactory.create("custom", model, device)
"""
@classmethod
def _validate_component(cls, strategy_cls: type):
"""Validate that the strategy class inherits from BaseStrategy."""
if not issubclass(strategy_cls, BaseStrategy):
raise TypeError(f"{strategy_cls.__name__} must inherit from BaseStrategy")
@classmethod
def create(cls, train_type: str, model, device: str, **kwargs) -> "BaseStrategy":
"""Create a strategy instance based on training type.
Args:
train_type: Type of training ("seq", "sft", "dpo", "grpo")
model: Model instance for the strategy
device: Device to run the strategy on
**kwargs: Additional arguments passed to strategy constructor
Returns:
Strategy instance
"""
return super().create(train_type, model, device, **kwargs)
@classmethod
def available_strategies(cls) -> list:
"""Return list of registered strategy names."""
return cls.list_registered()
# ============== Strategy Classes ==============
# All strategies are registered at class definition time using the decorator
@ -165,7 +144,13 @@ class SEQStrategy(BaseStrategy):
Computes cross-entropy loss for next token prediction.
"""
def __init__(self, model, device, label_smoothing: float = 0.0, **kwargs):
def __init__(
self,
model: Union[nn.Module, Callable[..., Dict[str, Tensor]]],
device: str,
label_smoothing: float = 0.0,
**kwargs,
):
super().__init__(model, device, **kwargs)
self.label_smoothing = label_smoothing
@ -190,7 +175,13 @@ class SFTStrategy(BaseStrategy):
Applies cross-entropy loss only to tokens where loss_mask is True.
"""
def __init__(self, model, device, label_smoothing: float = 0.0, **kwargs):
def __init__(
self,
model: Union[nn.Module, Callable[..., Dict[str, Tensor]]],
device: str,
label_smoothing: float = 0.0,
**kwargs,
):
super().__init__(model, device, **kwargs)
self.label_smoothing = label_smoothing

View File

@ -154,11 +154,13 @@ class CheckpointCallback(TrainCallback):
self.save_dir, f"epoch_{context.epoch}_iter_{context.iteration}"
)
extra = self.save_extra_fn(context)
meta = context.config.to_dict()
context.checkpoint = Checkpoint(
state_dict=state_dict,
epoch=context.epoch,
iteration=context.iteration,
extra=extra,
meta=meta,
config=context.model_config,
)
context.checkpoint.save(save_path)
@ -213,7 +215,7 @@ class ProgressBarCallback(TrainCallback):
"loss": f"{context.loss:.4f}",
"lr": f"{context.optimizer.param_groups[-1]['lr']:.2e}",
}
if context.val_loss > 0:
if context.val_loss is not None:
postfix["val_loss"] = f"{context.val_loss:.4f}"
self.progress_bar.set_postfix(postfix)
self.progress_bar.update(1)
@ -257,12 +259,16 @@ class MetricLoggerCallback(TrainCallback):
}
def _get_log_data(self, context: TrainContext):
return {
data = {
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"),
"epoch": context.epoch,
"iter": context.iteration,
**{m: self._metric_funcs[m](context) for m in self.metrics},
}
for m in self.metrics:
val = self._metric_funcs[m](context)
if val is not None:
data[m] = val
return data
@only_on_rank(0)
def _add_log(self, log_data):
@ -271,6 +277,7 @@ class MetricLoggerCallback(TrainCallback):
@only_on_rank(0)
def _save_log(self, epoch, iter):
log_file = self.log_dir / f"epoch_{epoch}_iter_{iter}_metric.jsonl"
log_file.parent.mkdir(parents=True, exist_ok=True)
with open(log_file, "w") as f:
for log in self.log_cache:

View File

@ -1,6 +1,6 @@
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional, Self
from typing import Any, Dict, Optional, Self
import torch
import torch.nn as nn
@ -12,7 +12,7 @@ from astrai.model.components.lora import inject_lora
from astrai.parallel.executor import BaseExecutor, ExecutorFactory
from astrai.parallel.setup import get_current_device, get_rank, get_world_size
from astrai.protocols import OptimizerProtocol, SchedulerProtocol
from astrai.serialization import Checkpoint, load_json, load_model_weights
from astrai.serialization import Checkpoint, load_json
from astrai.trainer.strategy import BaseStrategy, StrategyFactory
@ -31,12 +31,12 @@ class TrainContext:
epoch: int = field(default=0)
iteration: int = field(default=0)
loss: float = field(default=0.0)
val_dataloader: DataLoader = field(default=None)
val_loss: float = field(default=0.0)
val_dataloader: Optional[DataLoader] = field(default=None)
val_loss: Optional[float] = field(default=None)
world_size: int = field(default=1)
rank: int = field(default=0)
kwargs: dict = field(default_factory=dict)
kwargs: Dict[str, Any] = field(default_factory=dict)
class TrainContextBuilder:
@ -63,6 +63,7 @@ class TrainContextBuilder:
model = cfg.model_fn()
model = model.to(device=device)
model.embed_tokens.neftune_noise_alpha = cfg.neftune_alpha
model_config = {}
if self._resume_dir:
@ -82,21 +83,15 @@ class TrainContextBuilder:
executor=executor,
)
if self._resume_dir is not None:
resume_path = Path(self._resume_dir)
if (resume_path / "meta.json").exists():
checkpoint = Checkpoint.load(self._resume_dir)
state_dict = checkpoint.state_dict
if self._resume_dir:
checkpoint = Checkpoint.load_any(self._resume_dir)
if checkpoint is not None:
model.load_state_dict(checkpoint.state_dict, strict=False)
if checkpoint.config:
context.model_config = checkpoint.config
else:
checkpoint = None
state_dict = load_model_weights(self._resume_dir)
model.load_state_dict(state_dict, strict=False)
if checkpoint is not None:
context.epoch = cfg.start_epoch
context.iteration = cfg.start_batch
context.checkpoint = checkpoint
context.epoch = checkpoint.epoch or cfg.start_epoch
context.iteration = checkpoint.iteration or cfg.start_batch
context.checkpoint = checkpoint
if cfg.lora is not None:
inject_lora(
@ -172,8 +167,8 @@ class TrainContextBuilder:
obj.load_state_dict(extra[name])
context.strategy = StrategyFactory.create(
cfg.strategy,
model=context.model,
train_type=cfg.strategy,
device=device,
executor=executor,
model_fn=cfg.model_fn,

View File

@ -68,9 +68,8 @@ class Trainer:
self._call_callbacks("on_epoch_begin", context)
for batch in context.dataloader:
self._call_callbacks("on_batch_begin", context)
with executor.accumulate(context.model):
self._call_callbacks("on_batch_begin", context)
loss = context.strategy(batch)
context.loss = loss.item()
stand_loss = loss / executor.grad_accum_steps

View File

@ -14,8 +14,6 @@ import argparse
import json
import os
import re
import signal
import sys
from math import prod
from multiprocessing import Process, Queue
from typing import Dict, List, Optional, Tuple

View File

@ -0,0 +1,183 @@
"""IFD (Instruction Following Difficulty) data quality scoring.
Computes IFD scores for instruction-response pairs to guide data selection.
IFD = conditional_NLL / unconditional_NLL, where:
- conditional_NLL: average CE loss on response tokens given instruction context
- unconditional_NLL: average CE loss on response tokens alone
Higher IFD (close to 1) = instruction provides less help = harder sample.
Lower IFD (close to 0) = instruction provides strong guidance = easy sample.
IFD > 1 = instruction misleads the model = likely low-quality data.
Usage::
python scripts/eval/ifd.py --param_path ./params \
--input data.jsonl --output data_with_ifd.jsonl \
--instr_key instruction --resp_key response
"""
import argparse
import json
import torch
import torch.nn.functional as F
import tqdm
from astrai.model import AutoModel
from astrai.tokenize import AutoTokenizer
def compute_ifd(
model,
tokenizer,
instruction: str,
response: str,
device: str,
max_len: int = 2048,
) -> dict:
instr_ids = tokenizer.encode(instruction)
resp_ids = tokenizer.encode(response)
if not resp_ids:
return {
"L_cond": None,
"L_uncond": None,
"ifd": None,
"error": "empty response",
}
# Truncate instruction if total length exceeds max_len
qa_len = len(instr_ids) + len(resp_ids)
if qa_len > max_len:
overflow = qa_len - max_len
instr_ids = instr_ids[overflow:]
instr_len = len(instr_ids)
resp_len = len(resp_ids)
# Conditional: instruction + response
qa_ids = instr_ids + resp_ids
qa_tensor = torch.tensor([qa_ids], device=device, dtype=torch.long)
with torch.inference_mode():
logits_qa = model(qa_tensor)["logits"][0] # [qa_len, vocab]
resp_logits = logits_qa[instr_len - 1 : -1] # predict response tokens
resp_targets = torch.tensor(resp_ids, device=device, dtype=torch.long)
L_cond = F.cross_entropy(resp_logits, resp_targets, reduction="mean").item()
# Unconditional: response alone
resp_tensor = torch.tensor([resp_ids], device=device, dtype=torch.long)
with torch.inference_mode():
logits_resp = model(resp_tensor)["logits"][0] # [resp_len, vocab]
unp_logits = logits_resp[:-1] # causal shift
unp_targets = resp_tensor[0, 1:]
L_uncond = F.cross_entropy(unp_logits, unp_targets, reduction="mean").item()
ifd = L_cond / L_uncond if L_uncond > 0 else None
return {
"L_cond": round(L_cond, 6),
"L_uncond": round(L_uncond, 6),
"ifd": round(ifd, 6) if ifd is not None else None,
"instr_len": instr_len,
"resp_len": resp_len,
"error": None,
}
def process_file(
param_path: str,
input_file: str,
output_file: str,
instr_key: str,
resp_key: str,
max_len: int,
):
device = "cuda" if torch.cuda.is_available() else "cpu"
dtype = torch.bfloat16 if device == "cuda" else torch.float32
model = AutoModel.from_pretrained(param_path)
tokenizer = AutoTokenizer.from_pretrained(param_path)
model.to(device=device, dtype=dtype)
model.eval()
with open(input_file, "r", encoding="utf-8") as f:
data = [json.loads(line) for line in f if line.strip()]
results = []
ifd_values = []
with torch.inference_mode():
for item in tqdm.tqdm(data, desc="Computing IFD", unit="sample"):
instruction = item[instr_key]
response = item[resp_key]
scores = compute_ifd(
model, tokenizer, instruction, response, device, max_len
)
ifd_values.append(scores["ifd"])
results.append({**item, "ifd": scores["ifd"], "ifd_detail": scores})
with open(output_file, "w", encoding="utf-8") as f:
for item in results:
f.write(json.dumps(item, ensure_ascii=False) + "\n")
valid_ifd = [v for v in ifd_values if v is not None]
if valid_ifd:
import statistics
print(f"\n{'=' * 50}")
print(f" Samples: {len(data)}")
print(f" Valid IFD: {len(valid_ifd)}")
print(f" Mean IFD: {statistics.mean(valid_ifd):.4f}")
print(f" Median IFD: {statistics.median(valid_ifd):.4f}")
print(f" Stdev IFD: {statistics.stdev(valid_ifd):.4f}")
print(f" Min IFD: {min(valid_ifd):.4f}")
print(f" Max IFD: {max(valid_ifd):.4f}")
print(f"{'=' * 50}")
print(f"Results saved to {output_file}")
def main():
parser = argparse.ArgumentParser(
description="Compute IFD scores for instruction-response data"
)
parser.add_argument("--param_path", type=str, required=True, help="Model directory")
parser.add_argument("--input", type=str, required=True, help="Input JSONL file")
parser.add_argument("--output", type=str, required=True, help="Output JSONL file")
parser.add_argument(
"--instr_key",
type=str,
default="instruction",
help="Key for instruction field",
)
parser.add_argument(
"--resp_key",
type=str,
default="response",
help="Key for response field",
)
parser.add_argument(
"--max_len",
type=int,
default=2048,
help="Max token length (instruction truncated to fit)",
)
args = parser.parse_args()
process_file(
args.param_path,
args.input,
args.output,
args.instr_key,
args.resp_key,
args.max_len,
)
if __name__ == "__main__":
main()

View File

@ -0,0 +1,600 @@
"""IFEval instruction-following evaluation benchmark.
Evaluates model responses against regex-based constraint verifiers.
Supports all IFEval constraint types except language detection.
Usage::
python scripts/tools/evaluate_ifeval.py --param_path ./params \
--data_path ifeval.jsonl --output results.json \
--temperature 0.1 --max_tokens 512
"""
import argparse
import json
import os
import re
import urllib.request
from typing import Callable, Dict, List, Optional
import torch
import tqdm
from astrai.inference import InferenceEngine
from astrai.model import AutoModel
from astrai.tokenize import AutoTokenizer
IFEVAL_URL = (
"https://raw.githubusercontent.com/google-research/"
"google-research/master/instruction_following_eval/data/input_data.jsonl"
)
CONSTRAINT_VERIFIERS: Dict[str, Callable[[str, dict], bool]] = {}
def register(instruction_id: str):
def decorator(fn):
CONSTRAINT_VERIFIERS[instruction_id] = fn
return fn
return decorator
@register("keywords:existence")
def check_keyword_existence(response: str, kwargs: dict) -> bool:
for kw in kwargs["keywords"]:
if not re.search(re.escape(kw), response, re.IGNORECASE):
return False
return True
@register("keywords:frequency")
def check_keyword_frequency(response: str, kwargs: dict) -> bool:
keyword = kwargs["keyword"]
frequency = kwargs.get("frequency", 1)
relation = kwargs.get("relation", "at least")
count = len(re.findall(re.escape(keyword), response, re.IGNORECASE))
if relation == "less than":
return count < frequency
return count >= frequency
@register("keywords:forbidden_words")
def check_forbidden_words(response: str, kwargs: dict) -> bool:
for word in kwargs["forbidden_words"]:
if re.search(r"\b" + re.escape(word) + r"\b", response, re.IGNORECASE):
return False
return True
@register("keywords:letter_frequency")
def check_letter_frequency(response: str, kwargs: dict) -> bool:
letter = kwargs["letter"].lower()
frequency = kwargs.get("let_frequency", 1)
relation = kwargs.get("let_relation", "at least")
count = response.lower().count(letter)
if relation == "less than":
return count < frequency
return count >= frequency
@register("detectable_content:number_placeholders")
def check_placeholders(response: str, kwargs: dict) -> bool:
num = kwargs.get("num_placeholders", 1)
placeholders = re.findall(r"\[.*?\]", response)
return len(placeholders) >= num
@register("detectable_content:postscript")
def check_postscript(response: str, kwargs: dict) -> bool:
marker = kwargs.get("postscript_marker", "P.S.")
response_lower = response.lower()
if marker == "P.P.S":
return bool(re.search(r"p\.\s?p\.\s?s", response_lower))
elif marker == "P.S.":
return bool(re.search(r"p\.\s?s\.", response_lower))
else:
return bool(re.search(re.escape(marker.lower()), response_lower))
@register("detectable_format:number_bullet_lists")
def check_bullet_lists(response: str, kwargs: dict) -> bool:
num = kwargs.get("num_bullets", 1)
bullets = re.findall(r"^\s*\*[^\*].*$", response, re.MULTILINE)
dashes = re.findall(r"^\s*-.*$", response, re.MULTILINE)
return len(bullets) + len(dashes) == num
@register("detectable_format:number_highlighted_sections")
def check_highlighted_sections(response: str, kwargs: dict) -> bool:
num = kwargs.get("num_highlights", 1)
highlights = re.findall(r"\*[^\n\*]+\*", response)
count = 0
for h in highlights:
if h.strip("*").strip():
count += 1
return count >= num
@register("detectable_format:multiple_sections")
def check_multiple_sections(response: str, kwargs: dict) -> bool:
splitter = kwargs.get("section_spliter", "Section")
num = kwargs.get("num_sections", 1)
pattern = r"\s?" + re.escape(splitter) + r"\s?\d+\s?"
sections = re.split(pattern, response)
return len(sections) - 1 >= num
@register("detectable_format:title")
def check_title(response: str, kwargs: dict) -> bool:
titles = re.findall(r"<<[^>\n]+>>", response)
for title in titles:
if title.strip("<>").strip():
return True
return False
@register("detectable_format:json_format")
def check_json_format(response: str, kwargs: dict) -> bool:
value = response.strip()
for prefix in ("```json", "```Json", "```JSON", "```"):
if value.lower().startswith(prefix.lower()):
value = value[len(prefix) :].strip()
if value.endswith("```"):
value = value[:-3].strip()
try:
json.loads(value)
return True
except (ValueError, json.JSONDecodeError):
return False
@register("detectable_format:general_punctuation")
def check_general_punctuation(response: str, kwargs: dict) -> bool:
punctuation_blacklist = kwargs.get("punctuation_blacklist", [])
for punct in punctuation_blacklist:
if punct in response:
return False
return True
@register("detectable_format:number_highlighted_words")
def check_highlighted_words(response: str, kwargs: dict) -> bool:
num = kwargs.get("num_highlights", 1)
highlights = re.findall(r"\*[^\s\*][^\*]*[^\s\*]\*", response)
return len(highlights) >= num
@register("startend:end_checker")
def check_end_checker(response: str, kwargs: dict) -> bool:
end_phrase = kwargs["end_phrase"]
return (
response.strip()
.rstrip('"')
.rstrip()
.lower()
.endswith(end_phrase.strip().lower())
)
@register("startend:quotation")
def check_quotation(response: str, kwargs: dict) -> bool:
value = response.strip()
return value.startswith('"') and value.endswith('"')
@register("startend:start_checker")
def check_start_checker(response: str, kwargs: dict) -> bool:
starter = kwargs["starter"]
return bool(re.search(r"^\s*" + re.escape(starter), response, re.MULTILINE))
@register("change_case:english_capital")
def check_english_capital(response: str, kwargs: dict) -> bool:
return response.isupper()
@register("change_case:english_lowercase")
def check_english_lowercase(response: str, kwargs: dict) -> bool:
return response.islower()
@register("change_case:capital_word_frequency")
def check_capital_word_frequency(response: str, kwargs: dict) -> bool:
frequency = kwargs.get("capital_frequency", 1)
relation = kwargs.get("capital_relation", "at least")
capital_words = re.findall(r"\b[A-Z]{2,}\b", response)
count = len(capital_words)
if relation == "less than":
return count < frequency
return count >= frequency
@register("punctuation:no_comma")
def check_no_comma(response: str, kwargs: dict) -> bool:
return "," not in response
def count_words(text: str) -> int:
return len(re.findall(r"\b\w+\b", text))
def count_sentences(text: str) -> int:
text = text.strip()
if not text:
return 0
sentences = re.split(r"(?<=[.!?])\s+", text)
return len([s for s in sentences if s.strip()])
@register("length_constraints:number_words")
def check_number_words(response: str, kwargs: dict) -> bool:
num = kwargs.get("num_words", 100)
relation = kwargs.get("relation", "at least")
cnt = count_words(response)
if relation == "less than":
return cnt < num
return cnt >= num
@register("length_constraints:number_sentences")
def check_number_sentences(response: str, kwargs: dict) -> bool:
num = kwargs.get("num_sentences", 5)
relation = kwargs.get("relation", "at least")
cnt = count_sentences(response)
if relation == "less than":
return cnt < num
return cnt >= num
@register("length_constraints:number_paragraphs")
def check_number_paragraphs(response: str, kwargs: dict) -> bool:
num = kwargs.get("num_paragraphs", 1)
if "***" in response:
paragraphs = re.split(r"\s?\*\*\*\s?", response)
else:
paragraphs = re.split(r"\n\n+", response)
actual = len([p for p in paragraphs if p.strip()])
return actual == num
@register("length_constraints:nth_paragraph_first_word")
def check_nth_paragraph_first_word(response: str, kwargs: dict) -> bool:
num_paragraphs = kwargs.get("num_paragraphs", 1)
nth = kwargs.get("nth_paragraph", 1)
first_word = kwargs.get("first_word", "").lower()
paragraphs = re.split(r"\n\n+", response)
paragraphs = [p.strip() for p in paragraphs if p.strip()]
if len(paragraphs) != num_paragraphs:
return False
if nth > len(paragraphs):
return False
target = paragraphs[nth - 1]
words = target.split()
if not words:
return False
word = words[0].strip().lstrip("'\"").rstrip(".,!?:;\"'")
return word.lower() == first_word
@register("length_constraints:nth_word_checker")
def check_nth_word(response: str, kwargs: dict) -> bool:
nth = kwargs.get("nth_word", 1)
target = kwargs.get("target_word", "").lower()
words = re.findall(r"\b\w+\b", response)
if nth > len(words):
return False
return words[nth - 1].lower() == target
@register("combination:repeat_prompt")
def check_repeat_prompt(response: str, kwargs: dict) -> bool:
prompt = kwargs["prompt_to_repeat"]
return response.strip().lower().startswith(prompt.strip().lower())
@register("combination:two_responses")
def check_two_responses(response: str, kwargs: dict) -> bool:
parts = response.split("******")
valid = [p for p in parts if p.strip()]
if len(valid) != 2:
return False
return valid[0].strip() != valid[1].strip()
def download_ifeval(data_path: str):
if os.path.exists(data_path):
return
os.makedirs(os.path.dirname(data_path) or ".", exist_ok=True)
print(f"Downloading IFEval from {IFEVAL_URL} ...")
tmp = data_path + ".tmp"
urllib.request.urlretrieve(IFEVAL_URL, tmp)
with open(tmp, "rb") as f_in:
content = f_in.read()
with open(data_path, "wb") as f_out:
f_out.write(content)
os.remove(tmp)
print(f" saved to {data_path}")
def load_problems(data_path: str) -> List[dict]:
problems = []
with open(data_path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if line:
problems.append(json.loads(line))
return problems
def verify_response(response: str, instruction_id: str, kwargs: dict) -> Optional[bool]:
verifier = CONSTRAINT_VERIFIERS.get(instruction_id)
if verifier is None:
return None
try:
return verifier(response, kwargs)
except Exception:
return False
def generate_one(
engine: InferenceEngine,
prompt: str,
max_tokens: int,
temperature: float,
top_p: float,
top_k: int,
) -> str:
output = engine.generate(
prompt=prompt,
stream=False,
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
top_k=top_k,
)
if isinstance(output, list):
return output[0]
return output
def evaluate(
engine: InferenceEngine,
problems: List[dict],
max_tokens: int,
temperature: float,
top_p: float,
top_k: int,
num_samples: int = 1,
) -> Dict:
results = {}
constraint_stats: Dict[str, Dict[str, int]] = {}
total_constraints = 0
total_passed = 0
for problem in tqdm.tqdm(problems, desc="IFEval", unit="problem"):
key = problem["key"]
prompt = problem["prompt"]
instruction_ids = problem["instruction_id_list"]
kwargs_list = problem["kwargs"]
samples = []
for _ in range(num_samples):
response = generate_one(
engine, prompt, max_tokens, temperature, top_p, top_k
)
samples.append(response)
constraint_results = []
passed = 0
verified = 0
for idx, instruction_id in enumerate(instruction_ids):
kwargs = kwargs_list[idx] if idx < len(kwargs_list) else {}
best_pass = False
for response in samples:
result = verify_response(response, instruction_id, kwargs)
if result is None:
continue
if result:
best_pass = True
break
verifier_exists = instruction_id in CONSTRAINT_VERIFIERS
if verifier_exists:
verified += 1
if best_pass:
passed += 1
constraint_results.append(
{
"instruction_id": instruction_id,
"passed": best_pass,
"supported": verifier_exists,
"kwargs": kwargs,
}
)
if verifier_exists:
if instruction_id not in constraint_stats:
constraint_stats[instruction_id] = {
"total": 0,
"passed": 0,
}
constraint_stats[instruction_id]["total"] += 1
if best_pass:
constraint_stats[instruction_id]["passed"] += 1
total_constraints += verified
total_passed += passed
accuracy = passed / verified if verified > 0 else None
results[str(key)] = {
"key": key,
"prompt": prompt,
"response": samples[0],
"num_samples": num_samples,
"num_constraints": len(instruction_ids),
"num_verified": verified,
"num_passed": passed,
"accuracy": round(accuracy, 4) if accuracy is not None else None,
"constraints": constraint_results,
}
overall_accuracy = (
round(total_passed / total_constraints, 4) if total_constraints > 0 else 0.0
)
type_summary = {}
for inst_id, stats in sorted(constraint_stats.items()):
type_summary[inst_id] = {
"total": stats["total"],
"passed": stats["passed"],
"accuracy": round(stats["passed"] / stats["total"], 4)
if stats["total"] > 0
else 0.0,
}
unsupported_count = sum(
1
for p in problems
for iid in p["instruction_id_list"]
if iid not in CONSTRAINT_VERIFIERS
)
results["_summary"] = {
"total_problems": len(problems),
"total_constraints": total_constraints,
"total_passed": total_passed,
"overall_accuracy": overall_accuracy,
"unsupported_constraints": unsupported_count,
"supported_types": sorted(CONSTRAINT_VERIFIERS.keys()),
"per_type_accuracy": type_summary,
}
return results
def main():
parser = argparse.ArgumentParser(description="IFEval benchmark")
parser.add_argument(
"--param_path", type=str, default="./params", help="Model directory"
)
parser.add_argument(
"--data_path",
type=str,
default="./ifeval/input_data.jsonl",
help="IFEval JSONL file (auto-download if missing)",
)
parser.add_argument("--output", type=str, default=None, help="Output JSON path")
parser.add_argument(
"--max_tokens", type=int, default=512, help="Max generation tokens"
)
parser.add_argument(
"--temperature",
type=float,
default=0.1,
help="Sampling temperature",
)
parser.add_argument("--top_p", type=float, default=0.95, help="Top-p sampling")
parser.add_argument("--top_k", type=int, default=50, help="Top-k sampling")
parser.add_argument(
"--num_samples",
type=int,
default=1,
help="Number of samples per problem (best-of-n scoring)",
)
parser.add_argument(
"--batch_size", type=int, default=1, help="Inference batch size"
)
parser.add_argument(
"--limit",
type=int,
default=None,
help="Limit to first N problems (for quick testing)",
)
parser.add_argument(
"--dump_responses",
type=str,
default=None,
help="Path to dump raw model responses (JSONL)",
)
args = parser.parse_args()
download_ifeval(args.data_path)
problems = load_problems(args.data_path)
if args.limit:
problems = problems[: args.limit]
print(f"Loaded {len(problems)} problems")
print(f"Supported constraint types: {len(CONSTRAINT_VERIFIERS)}")
model = AutoModel.from_pretrained(args.param_path)
tokenizer = AutoTokenizer.from_pretrained(args.param_path)
model.to(device="cuda", dtype=torch.bfloat16)
engine = InferenceEngine(
model=model,
tokenizer=tokenizer,
max_batch_size=args.batch_size,
)
results = evaluate(
engine=engine,
problems=problems,
max_tokens=args.max_tokens,
temperature=args.temperature,
top_p=args.top_p,
top_k=args.top_k,
num_samples=args.num_samples,
)
summary = results.pop("_summary")
print(f"\n{'=' * 60}")
print(f" Problems: {summary['total_problems']}")
print(f" Constraints: {summary['total_constraints']}")
print(f" Passed: {summary['total_passed']}")
print(f" Accuracy: {summary['overall_accuracy']:.2%}")
print(f" Unsupported: {summary['unsupported_constraints']}")
print(f"{'=' * 60}")
print(f"\nPer-type accuracy:")
for inst_id, stats in sorted(summary["per_type_accuracy"].items()):
print(
f" {inst_id:50s} {stats['accuracy']:.2%} "
f"({stats['passed']}/{stats['total']})"
)
if args.output:
results["_summary"] = summary
with open(args.output, "w", encoding="utf-8") as f:
json.dump(results, f, indent=2, ensure_ascii=False)
print(f"\nResults saved to {args.output}")
if args.dump_responses:
with open(args.dump_responses, "w", encoding="utf-8") as f:
for k, v in results.items():
if k.startswith("_"):
continue
f.write(
json.dumps(
{
"key": v["key"],
"prompt": v["prompt"],
"response": v["response"],
},
ensure_ascii=False,
)
+ "\n"
)
print(f"Responses dumped to {args.dump_responses}")
engine.shutdown()
if __name__ == "__main__":
main()

View File

@ -24,7 +24,7 @@ def main():
)
args = parser.parse_args()
config = PipelineConfig.from_json(args.config)
config = PipelineConfig.from_file(args.config)
Pipeline(
config=config,

View File

@ -113,7 +113,7 @@ def parse_args() -> argparse.Namespace:
parser.add_argument(
"--label_smoothing",
type=float,
default=0.05,
default=0.0,
help="cross_entropy function label smoothing parameter",
)
parser.add_argument(
@ -214,6 +214,12 @@ def parse_args() -> argparse.Namespace:
choices=["spawn", "fork", "forkserver"],
help="Multiprocessing start method.",
)
parser.add_argument(
"--neftune_alpha",
type=float,
default=0.0,
help="NEFTune noise alpha (0=disabled, typical: 5.0).",
)
args = parser.parse_args()
@ -231,7 +237,8 @@ def create_optimizer(model, **kwargs) -> optim.Optimizer:
def create_scheduler(
optimizer: optim.Optimizer, **kwargs
) -> optim.lr_scheduler.LRScheduler:
return SchedulerFactory.create(optimizer, **kwargs)
schedule_type = kwargs.pop("schedule_type")
return SchedulerFactory.create(schedule_type, optimizer, **kwargs)
def compute_total_steps(
@ -292,6 +299,7 @@ def train(
master_addr: str,
master_port: str,
start_method: str,
neftune_alpha: float,
):
assert train_type in ["seq", "sft", "dpo", "grpo"]
assert os.path.exists(param_path)
@ -384,6 +392,7 @@ def train(
gradient_checkpointing_modules=grad_ckpt_modules,
executor_kwargs=executor_kwargs,
extra_kwargs=strategy_kwargs,
neftune_alpha=neftune_alpha,
)
trainer = Trainer(train_config)

View File

@ -291,7 +291,7 @@ def test_sectioned_text_too_short(test_tokenizer):
def test_factory_registered():
names = MaskBuilderFactory._registry.list_names()
names = MaskBuilderFactory.list_registered()
assert "sectioned" in names

View File

@ -53,15 +53,15 @@ def test_to_dict_roundtrip():
assert config2.mask == {"prompt": "mask", "response": "train"}
def test_to_json_from_json(temp_dir):
def test_to_file_from_file(temp_dir):
config = PipelineConfig(
input=InputConfig(sections=_TEXT_SECTIONS),
mask={"text": "train"},
mask_default="mask",
)
path = os.path.join(temp_dir, "config.json")
config.to_json(path)
loaded = PipelineConfig.from_json(path)
config.to_file(path)
loaded = PipelineConfig.from_file(path)
assert loaded.input.sections == _TEXT_SECTIONS
assert loaded.mask == {"text": "train"}
@ -69,8 +69,8 @@ def test_to_json_from_json(temp_dir):
def test_dpo_config_roundtrip(temp_dir):
config = make_dpo_chat_config()
path = os.path.join(temp_dir, "config.json")
config.to_json(path)
loaded = PipelineConfig.from_json(path)
config.to_file(path)
loaded = PipelineConfig.from_file(path)
assert loaded.input.sources is not None
assert "chosen" in loaded.input.sources
assert "rejected" in loaded.input.sources

View File

@ -121,8 +121,8 @@ class TestOpenAIResponseBuilder:
assert p["choices"][0]["finish_reason"] is None
def test_format_chunk(self, builder):
event = builder.format_chunk("hello")
payload = json.loads(event.split("data: ", 1)[1])
events = builder.format_chunk("hello", body="hello")
payload = json.loads(events[0].split("data: ", 1)[1])
assert payload["choices"][0]["delta"]["content"] == "hello"
assert payload["choices"][0]["finish_reason"] is None
@ -192,8 +192,8 @@ class TestAnthropicResponseBuilder:
assert payloads[1]["type"] == "content_block_start"
def test_format_chunk(self, builder):
event = builder.format_chunk("tok")
payload = json.loads(event.split("data: ", 1)[1])
events = builder.format_chunk("tok", body="tok")
payload = json.loads(events[0].split("data: ", 1)[1])
assert payload["type"] == "content_block_delta"
assert payload["delta"]["text"] == "tok"

View File

@ -0,0 +1,691 @@
"""Unit tests for tool call parsers."""
import pytest
from astrai.inference.api.tool_parser import (
_TOOL_CALL_HEAD_RE,
BaseToolParser,
SimpleJsonToolParser,
ToolParserFactory,
_find_partial_tool_call,
_find_tool_calls,
_scan_json,
)
def test_scan_complete_simple():
end, complete = _scan_json('{"key": "value"}', 0)
assert complete is True
assert end == len('{"key": "value"}')
def test_scan_complete_nested():
text = '{"outer": {"inner": 1}}'
end, complete = _scan_json(text, 0)
assert complete is True
assert end == len(text)
def test_scan_incomplete_unclosed():
end, complete = _scan_json('{"key": "value"', 0)
assert complete is False
def test_scan_incomplete_nested():
end, complete = _scan_json('{"outer": {"inner": 1}', 0)
assert complete is False
def test_scan_string_braces_ignored():
text = '{"key": "a{b}c"} extra'
end, complete = _scan_json(text, 0)
assert complete is True
def test_scan_escaped_quote_ignored():
text = r'{"key": "a\"b"}'
end, complete = _scan_json(text, 0)
assert complete is True
def test_scan_deeply_nested():
text = '{"a": {"b": {"c": {"d": {"e": 5}}}}}'
end, complete = _scan_json(text, 0)
assert complete is True
assert end == len(text)
def test_scan_array_with_braces():
text = '{"items": [{"x": 1}, {"x": 2}]}'
end, complete = _scan_json(text, 0)
assert complete is True
assert end == len(text)
def test_scan_code_in_string():
text = '{"fn": "function() { return 1; }"}'
end, complete = _scan_json(text, 0)
assert complete is True
def test_scan_unicode_chars():
text = '{"key": "\u5317\u4eac"}'
end, complete = _scan_json(text, 0)
assert complete is True
def test_find_single_tool_call():
text = '{"name": "get_weather", "arguments": {"city": "Beijing"}}'
results = _find_tool_calls(text)
assert len(results) == 1
assert results[0]["name"] == "get_weather"
assert '"city"' in results[0]["args"]
assert results[0]["complete"] is True
def test_find_text_before_tool_call():
text = 'Some text {"name": "func", "arguments": {}}'
results = _find_tool_calls(text)
assert len(results) == 1
assert results[0]["start"] > 0
def test_find_multiple_tool_calls():
text = '{"name": "f1", "arguments": {"a": 1}}{"name": "f2", "arguments": {"b": 2}}'
results = _find_tool_calls(text)
assert len(results) == 2
assert results[0]["name"] == "f1"
assert results[1]["name"] == "f2"
def test_find_no_tool_call():
results = _find_tool_calls("Hello, how are you?")
assert len(results) == 0
def test_find_non_tool_json_skipped():
results = _find_tool_calls('{"not_a_tool": true}')
assert len(results) == 0
def test_find_no_arguments_field():
results = _find_tool_calls('{"name": "simple_func"}')
assert len(results) == 1
assert results[0]["name"] == "simple_func"
assert results[0]["args"] == ""
def test_find_deeply_nested_arguments():
text = '{"name": "deep", "arguments": {"a": {"b": {"c": {"d": 4}}}}}'
results = _find_tool_calls(text)
assert len(results) == 1
assert results[0]["name"] == "deep"
assert '"d": 4' in results[0]["args"]
def test_find_arguments_with_boolean_and_null():
text = '{"name": "flags", "arguments": {"active": true, "count": 0, "nick": null}}'
results = _find_tool_calls(text)
assert len(results) == 1
assert results[0]["name"] == "flags"
assert "true" in results[0]["args"]
assert "null" in results[0]["args"]
def test_find_arguments_with_array():
text = '{"name": "add_items", "arguments": {"items": [1, 2, 3], "name": "list"}}'
results = _find_tool_calls(text)
assert len(results) == 1
assert results[0]["name"] == "add_items"
assert "[1, 2, 3]" in results[0]["args"]
def test_find_arguments_with_nested_array_of_objects():
text = (
'{"name": "batch", '
'"arguments": {"rows": [{"id": 1, "val": "a"}, {"id": 2, "val": "b"}]}}'
)
results = _find_tool_calls(text)
assert len(results) == 1
assert '"rows"' in results[0]["args"]
assert '"id": 1' in results[0]["args"]
def test_find_arguments_as_string_not_object():
text = '{"name": "echo", "arguments": "just a string"}'
results = _find_tool_calls(text)
assert len(results) == 1
assert results[0]["name"] == "echo"
assert "just a string" in results[0]["args"]
def test_find_arguments_with_unicode():
text = (
'{"name": "translate", "arguments": {"text": "\u4f60\u597d\uff0c\u4e16\u754c"}}'
)
results = _find_tool_calls(text)
assert len(results) == 1
assert results[0]["name"] == "translate"
def test_find_arguments_with_escaped_quotes():
text = '{"name": "format", "arguments": {"template": "he said \\"hello\\""}}'
results = _find_tool_calls(text)
assert len(results) == 1
assert 'he said \\"hello\\"' in results[0]["args"]
def test_find_arguments_with_braces_in_string():
text = '{"name": "eval", "arguments": {"code": "function(x) { return x + 1; }"}}'
results = _find_tool_calls(text)
assert len(results) == 1
assert results[0]["name"] == "eval"
assert "function(x) { return x + 1; }" in results[0]["args"]
def test_find_many_properties():
args = ",".join(f'"{chr(97 + i % 26)}" : {i}' for i in range(20))
text = '{"name": "many", "arguments": {' + args + "}}"
results = _find_tool_calls(text)
assert len(results) == 1
assert results[0]["name"] == "many"
def test_find_empty_arguments():
results = _find_tool_calls('{"name": "ping", "arguments": {}}')
assert len(results) == 1
assert results[0]["name"] == "ping"
assert results[0]["args"] == ""
def test_find_extracts_correct_arg_start_position():
text = '{"name": "f", "arguments": {"x": 1}}'
results = _find_tool_calls(text)
assert len(results) == 1
json_str = text[results[0]["start"] : results[0]["end"]]
assert json_str == text
def test_partial_with_name():
result = _find_partial_tool_call('{"name": "func", "arguments": {"city"')
assert result is not None
assert result["name"] == "func"
assert result["complete"] is False
def test_partial_with_full_args():
result = _find_partial_tool_call('{"name": "func", "arguments": {"city": "BJ"}}')
assert result is not None
assert result["name"] == "func"
def test_partial_no_match():
assert _find_partial_tool_call("plain text") is None
def test_partial_no_name_yet():
assert _find_partial_tool_call('{"nam') is None
def test_partial_deeply_nested():
result = _find_partial_tool_call('{"name": "deep", "arguments": {"a": {"b": {"c": ')
assert result is not None
assert result["name"] == "deep"
assert '"a"' in result["args"]
def test_partial_array_incomplete():
result = _find_partial_tool_call('{"name": "batch", "arguments": {"items": [1, 2, ')
assert result is not None
assert result["name"] == "batch"
def test_feed_plain_text():
parser = SimpleJsonToolParser()
deltas = parser.feed("Hello")
assert len(deltas) == 1
assert deltas[0]["content"] == "Hello"
def test_feed_incremental_text():
parser = SimpleJsonToolParser()
assert parser.feed("He") == [{"content": "He"}]
assert parser.feed("Hello") == [{"content": "llo"}]
def test_feed_tool_call_name_delta():
parser = SimpleJsonToolParser()
text = '{"name": "get_weather", "arguments": {"city": "Beijing"}}'
deltas = parser.feed(text)
tc_deltas = [d for d in deltas if "tool_calls" in d]
assert len(tc_deltas) >= 1
name_delta = tc_deltas[0]["tool_calls"][0]
assert name_delta["function"]["name"] == "get_weather"
assert name_delta["type"] == "function"
assert "id" in name_delta
def test_feed_tool_call_args_streaming():
parser = SimpleJsonToolParser()
d1 = parser.feed('{"name": "f", "arguments": {"x":')
d2 = parser.feed('{"name": "f", "arguments": {"x": "1"}}')
args_deltas = [
d
for batch in (d1, d2)
for d in batch
if "tool_calls" in d
and "function" in d["tool_calls"][0]
and "arguments" in d["tool_calls"][0]["function"]
]
assert len(args_deltas) >= 1
def test_feed_text_before_tool_call():
parser = SimpleJsonToolParser()
text = 'Let me check. {"name": "func", "arguments": {"a": 1}}'
deltas = parser.feed(text)
content_deltas = [d for d in deltas if "content" in d]
assert any("Let me check" in d.get("content", "") for d in content_deltas)
def test_has_tool_calls_false_by_default():
assert SimpleJsonToolParser().has_tool_calls is False
def test_has_tool_calls_true_after_detection():
parser = SimpleJsonToolParser()
parser.feed('{"name": "f", "arguments": {}}')
assert parser.has_tool_calls is True
def test_feed_no_content_when_no_new_text():
parser = SimpleJsonToolParser()
parser.feed("Hello")
assert parser.feed("Hello") == []
def test_feed_multiple_tool_calls():
parser = SimpleJsonToolParser()
text = '{"name": "f1", "arguments": {"a": 1}}{"name": "f2", "arguments": {"b": 2}}'
deltas = parser.feed(text)
tc_deltas = [d for d in deltas if "tool_calls" in d]
names = set()
for batch in tc_deltas:
for tc in batch["tool_calls"]:
if "function" in tc and "name" in tc["function"]:
names.add(tc["function"]["name"])
assert "f1" in names
assert "f2" in names
def test_feed_with_tools_constructor():
tools = [{"type": "function", "function": {"name": "get_weather"}}]
parser = SimpleJsonToolParser(tools=tools, tool_choice="auto")
deltas = parser.feed('{"name": "get_weather", "arguments": {"city": "BJ"}}')
assert len(deltas) > 0
def test_feed_content_after_tool_call_is_not_emitted():
parser = SimpleJsonToolParser()
parser.feed('{"name": "f", "arguments": {}} trailing text')
assert parser.has_tool_calls
def _collect_args_deltas(parser):
args_parts = []
for d in parser.feed(parser._text_buffer):
if "tool_calls" in d:
for tc in d["tool_calls"]:
fn = tc.get("function", {})
if "arguments" in fn and fn["arguments"]:
args_parts.append(fn["arguments"])
return args_parts
def _simulate_streaming(parser, text):
all_delta_names = []
all_args_chunks = []
for i in range(1, len(text) + 1):
deltas = parser.feed(text[:i])
for d in deltas:
if "tool_calls" in d:
for tc in d["tool_calls"]:
fn = tc.get("function", {})
if "name" in fn:
all_delta_names.append(fn["name"])
if "arguments" in fn and fn["arguments"]:
all_args_chunks.append(fn["arguments"])
return all_delta_names, all_args_chunks
def test_streaming_token_by_token_full_build():
parser = SimpleJsonToolParser()
text = '{"name": "get_weather", "arguments": {"city": "Beijing"}}'
names, args_chunks = _simulate_streaming(parser, text)
assert "get_weather" in names
joined_args = "".join(args_chunks)
assert '"city"' in joined_args
assert "Beijing" in joined_args
def test_streaming_token_by_token_text_then_tool():
parser = SimpleJsonToolParser()
parts = [
"I'll ",
"check ",
"that. ",
'{"',
'name": "search", ',
'"arguments": {"q": "hello"}}',
]
body = ""
content_chunks = []
tool_names = []
for part in parts:
body += part
deltas = parser.feed(body)
for d in deltas:
if "content" in d:
content_chunks.append(d["content"])
if "tool_calls" in d:
for tc in d["tool_calls"]:
fn = tc.get("function", {})
if "name" in fn:
tool_names.append(fn["name"])
full_content = "".join(content_chunks)
assert "I'll check that." in full_content
assert "search" in tool_names
def test_streaming_multiple_tool_calls_incremental():
parser = SimpleJsonToolParser()
text = '{"name": "f1", "arguments": {"a": 1}}{"name": "f2", "arguments": {"b": 2}}'
names, _ = _simulate_streaming(parser, text)
assert names[0] == "f1"
assert "f2" in names
def test_streaming_deeply_nested_args():
parser = SimpleJsonToolParser()
text = '{"name": "deep", "arguments": {"a": {"b": {"c": 42}}}}'
_, args_chunks = _simulate_streaming(parser, text)
joined = "".join(args_chunks)
assert '"c": 42' in joined
def test_streaming_args_with_unicode():
parser = SimpleJsonToolParser()
text = (
'{"name": "translate", "arguments": {"text": "\u4f60\u597d\uff0c\u4e16\u754c"}}'
)
_, args_chunks = _simulate_streaming(parser, text)
joined = "".join(args_chunks)
assert "\u4f60\u597d" in joined
def test_streaming_args_with_array():
parser = SimpleJsonToolParser()
text = '{"name": "add", "arguments": {"items": [1, 2, 3]}}'
_, args_chunks = _simulate_streaming(parser, text)
joined = "".join(args_chunks)
assert "[1, 2, 3]" in joined
def test_streaming_empty_arguments():
parser = SimpleJsonToolParser()
text = '{"name": "ping", "arguments": {}}'
deltas = parser.feed(text)
tc_deltas = [d for d in deltas if "tool_calls" in d]
assert len(tc_deltas) >= 1
name_delta = tc_deltas[0]["tool_calls"][0]
assert name_delta["function"]["name"] == "ping"
assert "arguments" in name_delta["function"]
def test_streaming_args_diff_only_emits_new_bytes():
parser = SimpleJsonToolParser()
step1 = parser.feed('{"name": "f", "arguments": {"city": "Bei')
step2 = parser.feed('{"name": "f", "arguments": {"city": "Beijing"}}')
all_args = []
for step in (step1, step2):
for d in step:
if "tool_calls" in d:
for tc in d["tool_calls"]:
fn = tc.get("function", {})
if "arguments" in fn and fn["arguments"]:
all_args.append(fn["arguments"])
joined = "".join(all_args)
assert "city" in joined
assert "Beijing" in joined
assert joined.startswith('"city":')
assert all_args[0] != all_args[1]
def test_streaming_distinct_tool_call_ids():
parser = SimpleJsonToolParser()
text = '{"name": "f1", "arguments": {"a": 1}}{"name": "f2", "arguments": {"b": 2}}'
all_ids = []
for i in range(1, len(text) + 1):
deltas = parser.feed(text[:i])
for d in deltas:
if "tool_calls" in d:
for tc in d["tool_calls"]:
if "id" in tc:
all_ids.append(tc["id"])
unique = list(dict.fromkeys(all_ids))
assert len(unique) == 2
def test_parse_complete_basic():
parser = SimpleJsonToolParser()
body = '{"name": "get_weather", "arguments": {"city": "Beijing"}}'
result = parser.parse_complete(body)
assert result is not None
assert result["tool_calls"][0]["function"]["name"] == "get_weather"
assert "Beijing" in result["tool_calls"][0]["function"]["arguments"]
def test_parse_complete_no_tool_call():
assert SimpleJsonToolParser().parse_complete("Hello world") is None
def test_parse_complete_with_content():
parser = SimpleJsonToolParser()
result = parser.parse_complete('Prefix text. {"name": "f", "arguments": {}}')
assert result is not None
assert result["content"] == "Prefix text."
def test_parse_complete_multiple_tool_calls():
parser = SimpleJsonToolParser()
body = (
'{"name": "get_weather", "arguments": {"city": "Beijing"}}'
'{"name": "get_time", "arguments": {"tz": "Asia/Shanghai"}}'
)
result = parser.parse_complete(body)
assert result is not None
assert len(result["tool_calls"]) == 2
assert result["tool_calls"][0]["function"]["name"] == "get_weather"
assert result["tool_calls"][1]["function"]["name"] == "get_time"
assert "Beijing" in result["tool_calls"][0]["function"]["arguments"]
assert "Asia/Shanghai" in result["tool_calls"][1]["function"]["arguments"]
def test_parse_complete_complex_real_world():
parser = SimpleJsonToolParser()
body = (
'{"name": "send_email", '
'"arguments": {'
'"to": ["a@b.com", "c@d.com"], '
'"cc": null, '
'"subject": "Hello World", '
'"body": "This is a test email.", '
'"priority": 1, '
'"attachments": false'
"}}"
)
result = parser.parse_complete(body)
assert result is not None
tc = result["tool_calls"][0]
assert tc["function"]["name"] == "send_email"
args = tc["function"]["arguments"]
assert '"to"' in args
assert "a@b.com" in args
assert "null" in args
assert "false" in args
def test_parse_complete_content_with_multiple_tool_calls():
parser = SimpleJsonToolParser()
body = (
"I will do two things. "
'{"name": "f1", "arguments": {"a": 1}}'
'{"name": "f2", "arguments": {"b": 2}}'
)
result = parser.parse_complete(body)
assert result is not None
assert result["content"] == "I will do two things."
assert len(result["tool_calls"]) == 2
def test_parse_complete_no_arguments_field():
parser = SimpleJsonToolParser()
result = parser.parse_complete('{"name": "ping"}')
assert result is not None
assert result["tool_calls"][0]["function"]["name"] == "ping"
assert result["tool_calls"][0]["function"]["arguments"] == ""
def test_parse_complete_content_is_none_when_pure_tool_call():
parser = SimpleJsonToolParser()
result = parser.parse_complete('{"name": "f", "arguments": {"x": 1}}')
assert result is not None
assert result["content"] is None
def test_parse_complete_tool_calls_have_ids():
parser = SimpleJsonToolParser()
result = parser.parse_complete(
'{"name": "f1", "arguments": {}}{"name": "f2", "arguments": {}}'
)
assert result is not None
ids = [tc["id"] for tc in result["tool_calls"]]
assert len(ids) == 2
assert all(isinstance(i, str) and i.startswith("call_") for i in ids)
assert ids[0] != ids[1]
def test_feed_then_parse_complete_same_instance():
parser = SimpleJsonToolParser()
parser.feed('{"name": "get_weather", "arguments": {"city": "Beijing"}}')
result = parser.parse_complete(
'{"name": "get_weather", "arguments": {"city": "Beijing"}}'
)
assert result is not None
assert result["tool_calls"][0]["function"]["name"] == "get_weather"
assert parser.has_tool_calls
def test_pattern_matches_basic():
assert _TOOL_CALL_HEAD_RE.search('{"name": "f"}')
def test_pattern_matches_with_whitespace():
assert _TOOL_CALL_HEAD_RE.search('{ "name" : "f"}')
def test_pattern_no_match_without_name():
assert _TOOL_CALL_HEAD_RE.search('{"other": 1}') is None
def test_pattern_match_mid_text():
assert _TOOL_CALL_HEAD_RE.search('prefix {"name": "f", "args": {}}') is not None
def test_pattern_name_at_start():
assert _TOOL_CALL_HEAD_RE.match('{"name": "f"}')
def test_pattern_leading_whitespace():
assert _TOOL_CALL_HEAD_RE.search(' {"name": "f"}') is not None
def test_factory_register_and_create():
parser = ToolParserFactory.create("simple_json")
assert isinstance(parser, BaseToolParser)
assert isinstance(parser, SimpleJsonToolParser)
def test_factory_create_passes_tools():
parser = ToolParserFactory.create(
"simple_json", tools=[{"type": "function"}], tool_choice="required"
)
assert parser.tool_choice == "required"
def test_factory_list_registered():
assert "simple_json" in ToolParserFactory.list_registered()
def test_factory_create_with_no_extra_kwargs():
assert isinstance(ToolParserFactory.create("simple_json"), BaseToolParser)
def test_factory_create_with_tools_only():
tools = [
{
"type": "function",
"function": {"name": "test", "parameters": {"type": "object"}},
}
]
parser = ToolParserFactory.create("simple_json", tools=tools)
assert parser.tools == tools
assert parser.tool_choice == "auto"
def test_feed_accepts_token_ids_and_ignores_them():
parser = SimpleJsonToolParser()
text = '{"name": "get_weather", "arguments": {"city": "Beijing"}}'
deltas_with = parser.feed(text, current_token_ids=[123, 456], delta_token_ids=[456])
assert len(deltas_with) > 0
def test_feed_token_ids_do_not_affect_parsing():
parser_no_ids = SimpleJsonToolParser()
parser_with_ids = SimpleJsonToolParser()
text = '{"name": "get_weather", "arguments": {"city": "Beijing"}}'
result_no = parser_no_ids.feed(text)
result_with = parser_with_ids.feed(
text, current_token_ids=[1, 2, 3], delta_token_ids=[3]
)
assert len(result_no) == len(result_with)
assert len(result_no) > 0
assert (
result_no[0]["tool_calls"][0]["function"]["name"]
== result_with[0]["tool_calls"][0]["function"]["name"]
)
def test_parser_uses_token_ids_for_detection():
class TokenIdParser(BaseToolParser):
def __init__(self, tools=None, tool_choice="auto"):
super().__init__(tools, tool_choice)
self._detections = 0
def feed(self, body, current_token_ids=None, delta_token_ids=None):
if current_token_ids and 999 in current_token_ids:
self._detections += 1
return []
def parse_complete(self, body):
return None
@property
def has_tool_calls(self):
return self._detections > 0
parser = TokenIdParser()
parser.feed("hello", current_token_ids=[1, 999, 3])
assert parser.has_tool_calls

View File

@ -65,7 +65,7 @@ def create_train_config(
def scheduler_fn(optim):
return SchedulerFactory.create(
optim, "cosine", warmup_steps=10, lr_decay_steps=10, min_rate=0.05
"cosine", optim, warmup_steps=10, lr_decay_steps=10, min_rate=0.05
)
return TrainConfig(

View File

@ -102,7 +102,7 @@ def test_gradient_checkpointing_trainer_integration(base_test_env, random_datase
def scheduler_fn(optim):
return SchedulerFactory.create(
optim, "cosine", warmup_steps=10, lr_decay_steps=10, min_rate=0.05
"cosine", optim, warmup_steps=10, lr_decay_steps=10, min_rate=0.05
)
train_config = TrainConfig(
@ -136,7 +136,7 @@ def test_callback_integration(base_test_env, random_dataset):
def scheduler_fn(optim):
return SchedulerFactory.create(
optim, "cosine", warmup_steps=10, lr_decay_steps=10, min_rate=0.05
"cosine", optim, warmup_steps=10, lr_decay_steps=10, min_rate=0.05
)
train_config = TrainConfig(

View File

@ -16,7 +16,7 @@ def test_early_stopping_simulation(base_test_env, early_stopping_dataset):
def scheduler_fn(optim):
return SchedulerFactory.create(
optim, "cosine", warmup_steps=10, lr_decay_steps=10, min_rate=0.05
"cosine", optim, warmup_steps=10, lr_decay_steps=10, min_rate=0.05
)
train_config = TrainConfig(

View File

@ -36,8 +36,8 @@ def test_schedule_factory_random_configs():
min_rate = params["min_rate"]
lr_decay_steps = total_steps - warmup_steps
scheduler = SchedulerFactory.create(
optimizer,
schedule_type,
optimizer,
warmup_steps=warmup_steps,
lr_decay_steps=lr_decay_steps,
min_rate=min_rate,
@ -52,8 +52,8 @@ def test_schedule_factory_random_configs():
t_mult = params["t_mult"]
min_rate = params["min_rate"]
scheduler = SchedulerFactory.create(
optimizer,
schedule_type,
optimizer,
warmup_steps=warmup_steps,
cycle_length=cycle_length,
t_mult=t_mult,
@ -103,8 +103,8 @@ def test_schedule_factory_edge_cases():
min_rate = params["min_rate"]
lr_decay_steps = total_steps - warmup_steps
scheduler = SchedulerFactory.create(
optimizer,
"cosine",
optimizer,
warmup_steps=warmup_steps,
lr_decay_steps=lr_decay_steps,
min_rate=min_rate,
@ -129,8 +129,8 @@ def test_schedule_factory_state_persistence():
min_rate = 0.1
lr_decay_steps = total_steps - warmup_steps
scheduler = SchedulerFactory.create(
optimizer,
"cosine",
optimizer,
warmup_steps=warmup_steps,
lr_decay_steps=lr_decay_steps,
min_rate=min_rate,
@ -146,8 +146,8 @@ def test_schedule_factory_state_persistence():
# Create new scheduler with same parameters
new_scheduler = SchedulerFactory.create(
optimizer,
"cosine",
optimizer,
warmup_steps=warmup_steps,
lr_decay_steps=lr_decay_steps,
min_rate=min_rate,