my prefect server setup
prefect-metrics.waow.tech
python
orchestration
cache compact write task to skip redundant embed+upsert
when observations haven't changed, synthesize_summary returns a cached
result — but write_summary_to_turbopuffer was still making an OpenAI
embedding call + TurboPuffer upsert with identical data every cycle.
add BySummaryContent cache policy to the write task, keyed on
handle + summary text hash. now both the LLM call and the write are
skipped when nothing has changed.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
+23
-1
+23
-1
flows/compact.py
+23
-1
flows/compact.py
···
164
164
return f"summary-{clean_handle(handle)}"
165
165
166
166
167
-
@task
167
+
class BySummaryContent(CachePolicy):
168
+
"""Cache write by handle + summary text hash. Skips embed+upsert when unchanged."""
169
+
170
+
def compute_key(
171
+
self,
172
+
task_ctx: TaskRunContext,
173
+
inputs: dict,
174
+
flow_parameters: dict,
175
+
**kwargs,
176
+
) -> str | None:
177
+
handle = inputs.get("handle")
178
+
summary = inputs.get("summary")
179
+
if not handle or not summary:
180
+
return None
181
+
h = hashlib.md5(summary.encode()).hexdigest()[:12]
182
+
return f"compact-write/{handle}/{h}"
183
+
184
+
185
+
@task(
186
+
cache_policy=BySummaryContent(),
187
+
cache_expiration=timedelta(hours=4),
188
+
persist_result=True,
189
+
)
168
190
def write_summary_to_turbopuffer(
169
191
tpuf_key: str,
170
192
openai_key: str,