Rotate and offset a single placeholder/prefix segment for kv_reuse mode.
(
self,
request_uid: str,
message: str,
m: Dict[str, Any],
anchors_for_ph: List[Dict],
)
| 1266 | } |
| 1267 | |
| 1268 | def update_kv_cache_segment( |
| 1269 | self, |
| 1270 | request_uid: str, |
| 1271 | message: str, |
| 1272 | m: Dict[str, Any], |
| 1273 | anchors_for_ph: List[Dict], |
| 1274 | ) -> Tuple[int, DynamicCache, Dict[str, torch.Tensor]]: |
| 1275 | """Rotate and offset a single placeholder/prefix segment for kv_reuse mode.""" |
| 1276 | new_ph, new_pf = self._rotate_segment_caches(m) |
| 1277 | new_ph, new_pf = self.offset_kv_cache_pair( |
| 1278 | m["ph_id"], message, request_uid, new_ph, new_pf, anchors_for_ph, temperature=1 |
| 1279 | ) |
| 1280 | |
| 1281 | seg_cache = new_ph.concat_([new_pf]) |
| 1282 | seg_token_ids = concat(self.trim_token_ids(m["ph_cache_ids"], m["drop_num"]), m["pf_ids"]) |
| 1283 | |
| 1284 | return m["idx"], seg_cache, seg_token_ids |
| 1285 | |
| 1286 | def process_anchor( |
| 1287 | self, |
nothing calls this directly
no test coverage detected