perf: PagedCache 持久前缀缓存 + LRU 逐出

- astrai/inference/cache.py: refcount 归零时保留 hash 映射，页加入 LRU evictable 池
- alloc() 无空闲页时从 LRU 逐出，优先释放 _free_mask
- lookup_prefix/inc_ref 触发 _touch 更新 LRU 序
- record_page 设置 pin 标记并从 LRU 移除

astrai/inference/cache.py

@@ -22,14 +22,16 @@ def page_hash(token_ids: List[int], page_idx: int, page_size: int) -> int:
 
 
 class PagedCache:
-    """Paged KV cache with page-table-indirected read/write.
+    """Paged KV cache with page-table-indirected read/write and persistent prefix caching.
 
     Combines:
       - Page pool (ref-counted alloc/free via bitmask)
       - KV tensor storage (k_cache, v_cache)
       - Prefix-cache hash lookup (page_content_hash -> physical_page_idx)
+      - LRU eviction for persistent cross-batch prefix caching
 
-    Call :meth:`bind` to obtain a batch view for the attention layers.
+    Pages with recorded hashes persist after refcount reaches 0 (pinned).
+    They are evicted via LRU only when alloc() finds no free pages.
     """
 
     def __init__(
@@ -57,6 +59,24 @@ class PagedCache:
         )
         self._page_to_hash: Dict[int, int] = {}
         self._hash_to_page: Dict[int, int] = {}
+        self._lru: List[int] = []
+        self._pin: List[bool] = [False] * n_pages
+
+    def _touch(self, idx: int) -> None:
+        if self._refs[idx] == 0 and idx in self._lru:
+            self._lru.remove(idx)
+            self._lru.append(idx)
+
+    def _evict_one(self) -> int:
+        while self._lru:
+            idx = self._lru.pop(0)
+            h = self._page_to_hash.pop(idx, None)
+            if h is not None:
+                self._hash_to_page.pop(h, None)
+            self._pin[idx] = False
+            self._refs[idx] = 1
+            return idx
+        return -1
 
     def record_page(
         self, page_idx: int, token_ids: List[int], logical_page_idx: int
@@ -67,6 +87,9 @@ class PagedCache:
             self._hash_to_page.pop(old_h, None)
         self._page_to_hash[page_idx] = h
         self._hash_to_page[h] = page_idx
+        self._pin[page_idx] = True
+        if page_idx in self._lru:
+            self._lru.remove(page_idx)
 
     def lookup_prefix(self, token_ids: List[int]) -> List[int]:
         full_pages = len(token_ids) // self.page_size
@@ -76,20 +99,25 @@ class PagedCache:
             p = self._hash_to_page.get(h)
             if p is None:
                 break
+            self._touch(p)
             hits.append(p)
         return hits
 
     def inc_ref(self, idx: int) -> None:
         self._refs[idx] += 1
+        if self._refs[idx] == 1 and idx in self._lru:
+            self._lru.remove(idx)
 
     def alloc(self) -> int:
-        lsb = self._free_mask & -self._free_mask
+        if self._free_mask:
-        if lsb == 0:
+            lsb = self._free_mask & -self._free_mask
-            return -1
+            idx = lsb.bit_length() - 1
-        idx = lsb.bit_length() - 1
+            self._free_mask ^= lsb
-        self._free_mask ^= lsb
+            self._refs[idx] = 1
-        self._refs[idx] = 1
+            if idx in self._lru:
-        return idx
+                self._lru.remove(idx)
+            return idx
+        return self._evict_one()
 
     def alloc_n(self, n: int) -> List[int]:
         pages = [self.alloc() for _ in range(n)]
@@ -103,10 +131,15 @@ class PagedCache:
     def free(self, idx: int) -> None:
         self._refs[idx] -= 1
         if self._refs[idx] == 0:
-            self._free_mask |= 1 << idx
+            h = self._page_to_hash.get(idx)
-            h = self._page_to_hash.pop(idx, None)
+            if h is not None and self._pin[idx]:
-            if h is not None:
+                self._lru.append(idx)
-                self._hash_to_page.pop(h, None)
+            else:
+                self._free_mask |= 1 << idx
+                h = self._page_to_hash.pop(idx, None)
+                if h is not None:
+                    self._hash_to_page.pop(h, None)
+                self._pin[idx] = False
 
     def bind(self, page_table: Tensor, total_len: int = 0) -> "CacheView":
         return CacheView(self, page_table, total_len)

astrai/inference/engine.py

@@ -11,7 +11,7 @@ import asyncio
 import gc
 import threading
 from dataclasses import dataclass
-from typing import Any, AsyncGenerator, Dict, Generator, List, Optional, Union
+from typing import Any, AsyncGenerator, Dict, Generator, List, Optional, Tuple, Union
 
 import torch
 import torch.nn as nn
@@ -126,7 +126,7 @@ class _Result:
             idx: Index of the generation task this token belongs to.
         """
         with self._cond:
-            self.tokens.append(token)
+            self.tokens.append((idx, token))
             if token is not STOP:
                 self.results[idx] += token
             else:
@@ -136,11 +136,11 @@ class _Result:
                     self._cond.notify_all()
             self._event.set()
 
-    def pop_all(self) -> List[str]:
+    def pop_all(self) -> List[Tuple[int, str]]:
-        """Returns and clears all accumulated tokens.
+        """Returns and clears all accumulated (idx, token) pairs.
 
         Returns:
-            List of token strings since the last call.
+            List of (index, token_string) tuples since the last call.
         """
         with self._cond:
             out = self.tokens.copy()
@@ -238,20 +238,22 @@ class InferenceEngine:
         temperature: float = 1.0,
         top_p: float = 1.0,
         top_k: int = 50,
-    ) -> Union[Generator[str, None, None], str, List[str]]:
+    ) -> Union[Generator, str, List[str]]:
         """Generates text from a prompt.
 
         Args:
             prompt: Single string or list of strings for batch generation.
-            stream: If True, returns a generator yielding tokens one by one.
+            stream: If True, returns a generator yielding tokens.
             max_tokens: Maximum number of tokens to generate.
             temperature: Sampling temperature.
             top_p: Nucleus sampling probability threshold.
             top_k: Top-k sampling count (0 disables).
 
         Returns:
-            Generator (stream=True), single string (non-stream, single prompt),
+            stream=False, single prompt: str
-            or list of strings (non-stream, batch prompts).
+            stream=False, batch: List[str]
+            stream=True, single prompt: Generator[str, None, None]
+            stream=True, batch: Generator[Tuple[int, str], None, None]
         """
         is_batch = isinstance(prompt, list)
         prompts = prompt if is_batch else [prompt]
@@ -348,49 +350,60 @@ class InferenceEngine:
         temperature: float,
         top_p: float,
         top_k: int,
-    ) -> Generator[str, None, None]:
+    ) -> Generator:
         """Internal streaming generator.
 
         Polls the _Result accumulator in a loop, yielding tokens as they arrive.
-        Cleans up the scheduler task on GeneratorExit.
+        Single prompt yields raw token strings; batch yields (idx, token) tuples.
 
         Args:
-            prompts: List of prompts (only first is used; batch not yet supported).
+            prompts: List of prompts.
-            is_batch: If True, raises NotImplementedError.
+            is_batch: If True, yields (idx, token) tuples; else yields raw tokens.
             max_tokens: Maximum tokens to generate.
             temperature: Sampling temperature.
             top_p: Nucleus sampling threshold.
             top_k: Top-k sampling count.
 
         Yields:
-            Decoded token strings.
+            Single prompt: decoded token strings.
+            Batch: (sequence_index, token_string) tuples.
         """
-        if is_batch:
+        n = len(prompts)
-            raise NotImplementedError("Batch streaming not yet supported")
+        result = _Result(count=n)
+        task_ids = []
 
-        result = _Result()
+        for i, p in enumerate(prompts):
+            task_id = self.scheduler.add_task(
+                prompt=p,
+                max_tokens=max_tokens,
+                temperature=temperature,
+                top_p=top_p,
+                top_k=top_k,
+                stream_callback=lambda tok, idx=i: result.append(tok, idx),
+            )
+            task_ids.append(task_id)
 
-        task_id = self.scheduler.add_task(
+        remaining = n
-            prompt=prompts[0],
+        finished = [False] * n
-            max_tokens=max_tokens,
-            temperature=temperature,
-            top_p=top_p,
-            top_k=top_k,
-            stream_callback=lambda tok: result.append(tok, 0),
-        )
 
         def gen():
+            nonlocal remaining
             try:
-                while True:
+                while remaining > 0:
-                    tokens = result.pop_all()
+                    items = result.pop_all()
-                    for token in tokens:
+                    for idx, token in items:
                         if token is STOP:
-                            return
+                            if not finished[idx]:
-                        yield token
+                                finished[idx] = True
-                    if not result.wait(timeout=0.05):
+                                remaining -= 1
-                        pass
+                        else:
+                            yield (idx, token) if is_batch else token
+                    if remaining > 0:
+                        if not result.wait(timeout=0.05):
+                            pass
             finally:
-                self.scheduler.remove_task(task_id)
+                for tid in task_ids:
+                    self.scheduler.remove_task(tid)
 
         return gen()