// Coverage grammars and multi-component routing, including time-major streams. #include #include #include #include #include #include #include #include #include "parallel.hpp" #include "token_streams.hpp" namespace py=pybind11; namespace { using State=std::vector; using States=std::vector; class Grammar { public: int64_t horizon,vocabulary; virtual ~Grammar()=default; virtual State initial()const=0; virtual State advance(const State&,int64_t)const=0; virtual int64_t covered(const State&)const=0; virtual int64_t next_dimension(const State&)const=0; virtual bool allowed(const State&,int64_t)const=0; virtual std::shared_ptr clone(int64_t)const=0; py::array_t mask(const State& state)const { py::array_t result(vocabulary); py::gil_scoped_release release; for(int64_t token=0;token>>; std::shared_ptr> roots_; std::shared_ptr patterns_; std::shared_ptr> scope_; public: SetGrammar(int64_t steps,uint64_t full,std::vector roots, std::vector>> patterns) :full_(full),roots_(std::make_shared>(std::move(roots))), patterns_(std::make_shared(std::move(patterns))) { if(steps<1 || !full || roots_->size()!=patterns_->size())throw std::invalid_argument("invalid grammar metadata"); horizon=steps;vocabulary=roots_->size(); std::vector scopes; for(const auto& pattern:*patterns_) { bool scope=true; for(auto[offset,support]:pattern) { if(offset<0)throw std::invalid_argument("negative grammar pattern offset"); scope=scope && !(support & ~full_); } scopes.push_back(scope); } scope_=std::make_shared>(std::move(scopes)); } std::shared_ptr clone(int64_t steps)const override { if(steps<1)throw std::invalid_argument("horizon must be positive"); auto result=std::make_shared(*this);result->horizon=steps;return result; } State initial()const override{return State(horizon,0);} int64_t covered(const State& state)const override { if(static_cast(state.size())!=horizon)throw std::invalid_argument("invalid grammar state"); for(int64_t i=0;i(state[i])!=full_)return i; return horizon; } int64_t next_dimension(const State& state)const override { int64_t front=covered(state);if(front==horizon)return -1; uint64_t missing=full_ & ~static_cast(state[front]); int64_t d=0;while(!(missing & 1)){missing>>=1;++d;}return d; } bool allowed(const State& state,int64_t token)const override { if(token<0 || token>=vocabulary || !(*scope_)[token])return false; const int64_t front=covered(state); if(front==horizon || (*roots_)[token]!=next_dimension(state))return false; for(auto[offset,support]:(*patterns_)[token]) { if(front+offset>=horizon || (static_cast(state[front+offset]) & support))return false; } return true; } State advance(const State& state,int64_t token)const override { if(!allowed(state,token))return {}; State result=state;int64_t front=covered(state); for(auto[offset,support]:(*patterns_)[token])result[front+offset]|=static_cast(support); return result; } }; class PlainGrammar:public Grammar { std::shared_ptr> lengths_,dims_; std::shared_ptr> phase_; public: PlainGrammar(int64_t steps,std::vector lengths,std::vector dims, py::array_t phase) :lengths_(std::make_shared>(std::move(lengths))), dims_(std::make_shared>(std::move(dims))) { horizon=steps;vocabulary=lengths_->size(); if(steps<1 || dims_->empty() || phase.ndim()!=2 || phase.shape(0)!=static_cast(dims_->size()) || phase.shape(1)!=vocabulary) throw std::invalid_argument("invalid plain-BPE grammar metadata"); for(auto value:*lengths_)if(value<1)throw std::invalid_argument("invalid token expansion"); phase_=std::make_shared>(phase.data(),phase.data()+phase.size()); } std::shared_ptr clone(int64_t steps)const override { if(steps<1)throw std::invalid_argument("horizon must be positive"); auto result=std::make_shared(*this);result->horizon=steps;return result; } State initial()const override{return {0};} int64_t covered(const State& state)const override { if(state.size()!=1 || state[0]<0)throw std::invalid_argument("invalid grammar state"); return state[0]/dims_->size(); } int64_t next_dimension(const State& state)const override{return (*dims_)[state[0]%dims_->size()];} bool allowed(const State& state,int64_t token)const override { return token>=0 && token(dims_->size()) && (*phase_)[(state[0]%dims_->size())*vocabulary+token]; } State advance(const State& state,int64_t token)const override { if(!allowed(state,token))return {}; return {state[0]+(*lengths_)[token]}; } }; using Grammars=std::vector>; int64_t next_component(const Grammars& grammars,const States& states,bool time_major) { int64_t selected=grammars.size(),minimum=INT64_MAX; for(size_t i=0;icovered(states[i]); if(front>=grammars[i]->horizon)continue; if(!time_major)return i; if(frontinitial());return states; } bool advance(const Grammars& grammars,const std::vector& offsets,States& states,int64_t index,int64_t token) { if(index<0 || index>=static_cast(grammars.size()))return false; State next=grammars[index]->advance(states[index],token-offsets[index]); if(next.empty())return false; states[index]=std::move(next); return true; } py::tuple state_tuple(const State& state) { py::tuple result(state.size());for(size_t i=0;i& offsets, bool time_major,int64_t index,States states,int64_t token) { if(states.size()!=grammars.size() || offsets.size()!=grammars.size())throw std::invalid_argument("invalid routed metadata"); if(!advance(grammars,offsets,states,index,token))return py::none(); py::tuple local(states.size());for(size_t i=0;i interleave(const Grammars& grammars,const std::vector& offsets, bool time_major,const std::vector>& streams) { if(streams.size()!=grammars.size() || offsets.size()!=grammars.size())throw std::invalid_argument("one stream per component is required"); States states=initial(grammars);std::vector cursor(streams.size());std::vector output; int64_t index; while((index=next_component(grammars,states,time_major))(grammars.size())) { if(cursor[index]==streams[index].size())throw std::invalid_argument("component stream ends before complete coverage"); const int64_t token=streams[index][cursor[index]++]+offsets[index]; if(!advance(grammars,offsets,states,index,token))throw std::invalid_argument("invalid routed component stream"); output.push_back(token); } for(size_t i=0;i> split(const Grammars& grammars,const std::vector& offsets, bool time_major,const std::vector& tokens) { if(offsets.size()!=grammars.size())throw std::invalid_argument("invalid routed offsets"); States states=initial(grammars);std::vector> output(grammars.size()); for(int64_t token:tokens) { int64_t index=next_component(grammars,states,time_major); if(!advance(grammars,offsets,states,index,token))throw std::invalid_argument("invalid routed token coverage"); output[index].push_back(token-offsets[index]); } if(next_component(grammars,states,time_major)!=static_cast(grammars.size())) throw std::invalid_argument("Incomplete action token stream"); return output; } using Streams=std::vector>; using ComponentStreams=std::vector; Streams interleave_batch(const Grammars& grammars,const std::vector& offsets, bool time_major,const ComponentStreams& streams) { if(streams.size()!=grammars.size() || streams.empty() || streams.front().empty()) throw std::invalid_argument("one non-empty batch per component is required"); const int64_t batch=streams.front().size(); for(const auto& rows:streams)if(static_cast(rows.size())!=batch) throw std::invalid_argument("component batches must have the same size"); Streams output(batch); py::gil_scoped_release release; ac2::parallel_for(batch,0,[&](int64_t b) { Streams local;for(const auto& rows:streams)local.push_back(rows[b]); output[b]=interleave(grammars,offsets,time_major,local); }); return output; } py::tuple decode_batch(const Grammars& prototypes,const std::vector& offsets, bool time_major,const py::iterable& streams, py::array_t counts, int64_t cells_per_step) { if(prototypes.empty() || counts.ndim()!=1 || cells_per_step<1) throw std::invalid_argument("invalid routed batch metadata"); const auto rows=ac2::checked_token_streams(streams,counts.size(),true); int64_t steps=-1; ComponentStreams output(prototypes.size(),Streams(rows.size())); { py::gil_scoped_release release; for(const auto& row:rows) { int64_t cells=0; for(int64_t token:row) { if(counts.data()[token]<1)throw std::invalid_argument("token is outside the routed profile vocabularies"); cells+=counts.data()[token]; } if(cells%cells_per_step || cells/cells_per_step<1) throw std::invalid_argument("Incomplete action token stream: " + std::to_string(cells_per_step-cells%cells_per_step) + " more coordinate values are needed to finish a step. " "Constrain generation with codec.grammar(H, fps=...).next_token_mask(prefix) " "and check is_complete(tokens) for the intended H before decoding. " "Use codec.batch_decode(..., errors='return') to isolate invalid rows; " "do not pad truncated tokens with zeros."); if(steps>=0 && steps!=cells/cells_per_step) throw std::invalid_argument("all token streams must decode to the same length; use codec.batch_decode(...) for independent variable-length outputs"); steps=cells/cells_per_step; } Grammars grammars;for(const auto& prototype:prototypes)grammars.push_back(prototype->clone(steps)); ac2::parallel_for(rows.size(),0,[&](int64_t b) { auto local=split(grammars,offsets,time_major,rows[b]); for(size_t i=0;i complete(const Grammars& grammars,const std::vector& offsets, bool time_major,std::vector tokens,int64_t steps, const std::vector>& zero_atoms) { if(grammars.empty() || offsets.size()!=grammars.size() || zero_atoms.size()!=grammars.size() || steps<1 || steps>grammars.front()->horizon)throw std::invalid_argument("invalid prefix completion metadata"); py::gil_scoped_release release; States states=initial(grammars); for(int64_t token:tokens) { int64_t index=next_component(grammars,states,time_major); if(!advance(grammars,offsets,states,index,token))throw std::invalid_argument("invalid routed token prefix"); } for(size_t i=0;icovered(states[i])(grammars.size())) { int64_t dimension=grammars[index]->next_dimension(states[index]); if(dimension<0 || dimension>=static_cast(zero_atoms[index].size())) throw std::invalid_argument("missing zero-motion primitive atom"); const int64_t token=offsets[index]+zero_atoms[index][dimension]; if(!advance(grammars,offsets,states,index,token))throw std::invalid_argument("invalid zero-motion suffix"); tokens.push_back(token); } return tokens; } } // namespace void register_grammar(py::module_& module) { py::class_>(module,"CoverageGrammar") .def("advance",[](const Grammar& g,const State& s,int64_t token)->py::object { State next=g.advance(s,token);return next.empty()?py::object(py::none()):py::object(state_tuple(next)); }) .def("mask",&Grammar::mask).def("covered",&Grammar::covered).def("clone",&Grammar::clone); py::class_>(module,"SetGrammar") .def(py::init,std::vector>>>()); py::class_>(module,"PlainGrammar") .def(py::init,std::vector,py::array_t>()); module.def("route_advance",&route_advance); module.def("route_interleave",&interleave,py::call_guard()); module.def("route_split",&split,py::call_guard()); module.def("route_interleave_batch",&interleave_batch); module.def("route_decode_batch",&decode_batch); module.def("route_complete",&complete); }