gary-boon commited on
Commit
3506846
·
1 Parent(s): b5e4add

Derive aggregate-row num_heads from request metadata, not live model

Browse files

The /attention/row aggregate endpoint was reading num_attention_heads
from manager.model.config at request time. After a model switch, that
returns the *new* model's head count even when serving rows from the
*previous* model's cached attention. The per-head row data is correct
(stored under request_id keys) but the aggregation averaged the wrong
number of heads, silently producing wrong values.

Record per-request metadata (num_heads, num_layers, model_id) at
analysis time on the MatrixCache. get_aggregate_row derives num_heads
from this metadata; the explicit argument is retained as an optional
override for tests. The endpoint no longer touches manager.model
when serving aggregate rows.

clear_request and clear_old_requests drop metadata alongside the
per-head cache entries.

Files changed (1) hide show
  1. backend/model_service.py +80 -12
backend/model_service.py CHANGED
@@ -71,6 +71,12 @@ class MatrixCache:
71
  self._cache: Dict[str, Dict] = {}
72
  self._timestamps: Dict[str, float] = {}
73
  self._request_ids: set = set() # Track active request IDs
 
 
 
 
 
 
74
  self._lock = Lock()
75
  self._ttl = ttl_seconds
76
 
@@ -83,6 +89,7 @@ class MatrixCache:
83
  if k in self._timestamps:
84
  del self._timestamps[k]
85
  self._request_ids.discard(request_id)
 
86
  if keys_to_delete:
87
  logger.info(f"MatrixCache: cleared {len(keys_to_delete)} entries for request {request_id[:8]}")
88
 
@@ -99,6 +106,10 @@ class MatrixCache:
99
  del self._timestamps[k]
100
  total_cleared += len(keys_to_delete)
101
  self._request_ids = {keep_request_id} if keep_request_id else set()
 
 
 
 
102
  if total_cleared:
103
  logger.info(f"MatrixCache: cleared {total_cleared} entries from old requests")
104
  # Force garbage collection to release memory back to system
@@ -118,6 +129,29 @@ class MatrixCache:
118
  self._timestamps[key] = time_now()
119
  self._request_ids.add(request_id)
120
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
121
  def get(self, request_id: str, step: int, layer: int, head: int) -> Optional[dict]:
122
  """Retrieve matrix data, returning None if expired or not found."""
123
  key = f"{request_id}:{step}:{layer}:{head}"
@@ -169,20 +203,33 @@ class MatrixCache:
169
  return list(last_row)
170
 
171
  def get_aggregate_row(self, request_id: str, step: int, layer: int,
172
- num_heads: int, mode: str = "mean") -> Optional[list]:
 
173
  """
174
  Compute aggregated attention row across all heads for a layer.
175
 
 
 
 
 
 
 
176
  Args:
177
  request_id: UUID from analysis
178
  step: Generation step
179
  layer: Layer index
180
- num_heads: Number of attention heads in model
181
  mode: Aggregation mode - "mean" or "max"
 
182
 
183
  Returns:
184
  List of aggregated attention weights, or None if data unavailable
 
185
  """
 
 
 
 
 
186
  rows = []
187
  for h in range(num_heads):
188
  row = self.get_attention_row(request_id, step, layer, h)
@@ -1799,6 +1846,17 @@ async def analyze_research_attention(request: Dict[str, Any], authenticated: boo
1799
  # Clear old cached matrices to free memory before starting new analysis
1800
  matrix_cache.clear_old_requests(request_id)
1801
 
 
 
 
 
 
 
 
 
 
 
 
1802
  # Get parameters
1803
  prompt = request.get("prompt", "def quicksort(arr):")
1804
  max_tokens = request.get("max_tokens", 8)
@@ -2586,6 +2644,17 @@ async def analyze_research_attention_stream(request: Dict[str, Any], authenticat
2586
  # Clear old cached matrices to free memory before starting new analysis
2587
  matrix_cache.clear_old_requests(request_id)
2588
 
 
 
 
 
 
 
 
 
 
 
 
2589
  # Get parameters
2590
  prompt = request.get("prompt", "def quicksort(arr):")
2591
  max_tokens = request.get("max_tokens", 8)
@@ -3845,15 +3914,13 @@ async def get_attention_row(
3845
  - head: Head index (null if aggregated)
3846
  - aggregate_mode: Mode used if aggregated (null otherwise)
3847
  """
3848
- # Get number of heads from model config
3849
- if not manager.model:
3850
- raise HTTPException(status_code=503, detail="Model not loaded")
3851
-
3852
- config = manager.model.config
3853
- num_heads = getattr(config, 'num_attention_heads', getattr(config, 'n_head', 16))
3854
-
3855
  if head is not None:
3856
- # Fetch specific head
3857
  attention_row = matrix_cache.get_attention_row(request_id, step, layer, head)
3858
  if attention_row is None:
3859
  logger.warning(f"Attention row cache miss: request_id={request_id}, step={step}, layer={layer}, head={head}")
@@ -3870,9 +3937,10 @@ async def get_attention_row(
3870
  "aggregate_mode": None
3871
  }
3872
  else:
3873
- # Aggregate across all heads
 
3874
  attention_row = matrix_cache.get_aggregate_row(
3875
- request_id, step, layer, num_heads, aggregate_mode
3876
  )
3877
  if attention_row is None:
3878
  logger.warning(f"Attention row aggregate cache miss: request_id={request_id}, step={step}, layer={layer}")
 
71
  self._cache: Dict[str, Dict] = {}
72
  self._timestamps: Dict[str, float] = {}
73
  self._request_ids: set = set() # Track active request IDs
74
+ # Per-request metadata captured at analysis time. Aggregation across
75
+ # heads (e.g. mean/max in /attention/row) needs num_heads, and that
76
+ # number belongs to the model that produced the cached attention,
77
+ # not whichever model is loaded when the row is later retrieved.
78
+ # Storing it here decouples retrieval from `manager.model` state.
79
+ self._request_meta: Dict[str, Dict] = {}
80
  self._lock = Lock()
81
  self._ttl = ttl_seconds
82
 
 
89
  if k in self._timestamps:
90
  del self._timestamps[k]
91
  self._request_ids.discard(request_id)
92
+ self._request_meta.pop(request_id, None)
93
  if keys_to_delete:
94
  logger.info(f"MatrixCache: cleared {len(keys_to_delete)} entries for request {request_id[:8]}")
95
 
 
106
  del self._timestamps[k]
107
  total_cleared += len(keys_to_delete)
108
  self._request_ids = {keep_request_id} if keep_request_id else set()
109
+ # Drop metadata for cleared requests; keep only the surviving one.
110
+ for rid in list(self._request_meta.keys()):
111
+ if rid != keep_request_id:
112
+ self._request_meta.pop(rid, None)
113
  if total_cleared:
114
  logger.info(f"MatrixCache: cleared {total_cleared} entries from old requests")
115
  # Force garbage collection to release memory back to system
 
129
  self._timestamps[key] = time_now()
130
  self._request_ids.add(request_id)
131
 
132
+ def set_request_metadata(self, request_id: str, num_heads: int,
133
+ num_layers: Optional[int] = None,
134
+ model_id: Optional[str] = None):
135
+ """
136
+ Record metadata for the model that produced this request's cached
137
+ matrices. Aggregation operations (e.g. mean/max across heads) read
138
+ num_heads from here rather than from the live `manager.model`, so
139
+ that switching models mid-session cannot corrupt aggregate results
140
+ for older requests.
141
+ """
142
+ with self._lock:
143
+ self._request_meta[request_id] = {
144
+ "num_heads": num_heads,
145
+ "num_layers": num_layers,
146
+ "model_id": model_id,
147
+ }
148
+
149
+ def get_request_metadata(self, request_id: str) -> Optional[dict]:
150
+ """Return per-request metadata recorded at analysis time, or None."""
151
+ with self._lock:
152
+ meta = self._request_meta.get(request_id)
153
+ return dict(meta) if meta else None
154
+
155
  def get(self, request_id: str, step: int, layer: int, head: int) -> Optional[dict]:
156
  """Retrieve matrix data, returning None if expired or not found."""
157
  key = f"{request_id}:{step}:{layer}:{head}"
 
203
  return list(last_row)
204
 
205
  def get_aggregate_row(self, request_id: str, step: int, layer: int,
206
+ mode: str = "mean",
207
+ num_heads: Optional[int] = None) -> Optional[list]:
208
  """
209
  Compute aggregated attention row across all heads for a layer.
210
 
211
+ num_heads is derived from per-request metadata recorded at analysis
212
+ time so the aggregation always uses the head count of the model that
213
+ produced the cached attention. The explicit `num_heads` argument is
214
+ retained for callers that want to override (e.g. tests); when None,
215
+ falls back to the request's stored metadata.
216
+
217
  Args:
218
  request_id: UUID from analysis
219
  step: Generation step
220
  layer: Layer index
 
221
  mode: Aggregation mode - "mean" or "max"
222
+ num_heads: Optional override; otherwise derived from request meta
223
 
224
  Returns:
225
  List of aggregated attention weights, or None if data unavailable
226
+ or request metadata is missing.
227
  """
228
+ if num_heads is None:
229
+ meta = self.get_request_metadata(request_id)
230
+ if meta is None or meta.get("num_heads") is None:
231
+ return None
232
+ num_heads = meta["num_heads"]
233
  rows = []
234
  for h in range(num_heads):
235
  row = self.get_attention_row(request_id, step, layer, h)
 
1846
  # Clear old cached matrices to free memory before starting new analysis
1847
  matrix_cache.clear_old_requests(request_id)
1848
 
1849
+ # Record per-request model metadata so aggregate-row lookups can
1850
+ # derive num_heads from this request rather than from the live
1851
+ # `manager.model` (which may have been switched in the meantime).
1852
+ if manager.adapter is not None:
1853
+ matrix_cache.set_request_metadata(
1854
+ request_id,
1855
+ num_heads=manager.adapter.get_num_heads(),
1856
+ num_layers=manager.adapter.get_num_layers(),
1857
+ model_id=manager.model_id,
1858
+ )
1859
+
1860
  # Get parameters
1861
  prompt = request.get("prompt", "def quicksort(arr):")
1862
  max_tokens = request.get("max_tokens", 8)
 
2644
  # Clear old cached matrices to free memory before starting new analysis
2645
  matrix_cache.clear_old_requests(request_id)
2646
 
2647
+ # Record per-request model metadata so aggregate-row lookups can
2648
+ # derive num_heads from this request rather than from the live
2649
+ # `manager.model` (which may have been switched in the meantime).
2650
+ if manager.adapter is not None:
2651
+ matrix_cache.set_request_metadata(
2652
+ request_id,
2653
+ num_heads=manager.adapter.get_num_heads(),
2654
+ num_layers=manager.adapter.get_num_layers(),
2655
+ model_id=manager.model_id,
2656
+ )
2657
+
2658
  # Get parameters
2659
  prompt = request.get("prompt", "def quicksort(arr):")
2660
  max_tokens = request.get("max_tokens", 8)
 
3914
  - head: Head index (null if aggregated)
3915
  - aggregate_mode: Mode used if aggregated (null otherwise)
3916
  """
3917
+ # Aggregation parameters (num_heads in particular) come from per-request
3918
+ # metadata recorded at analysis time, not from the live model config.
3919
+ # This keeps aggregate rows correct even after the user switches models
3920
+ # mid-session — the row is averaged over the heads of the model that
3921
+ # actually produced the cached attention.
 
 
3922
  if head is not None:
3923
+ # Fetch specific head — no aggregation, no metadata lookup needed.
3924
  attention_row = matrix_cache.get_attention_row(request_id, step, layer, head)
3925
  if attention_row is None:
3926
  logger.warning(f"Attention row cache miss: request_id={request_id}, step={step}, layer={layer}, head={head}")
 
3937
  "aggregate_mode": None
3938
  }
3939
  else:
3940
+ # Aggregate across all heads. num_heads is derived inside
3941
+ # get_aggregate_row from the request's stored metadata.
3942
  attention_row = matrix_cache.get_aggregate_row(
3943
+ request_id, step, layer, aggregate_mode
3944
  )
3945
  if attention_row is None:
3946
  logger.warning(f"Attention row aggregate cache miss: request_id={request_id}, step={step}, layer={layer}")