Prompt-versioning a trading research assistant with Temporal
Six months ago, I deployed an LLM-based research assistant designed to ingest quarterly earnings transcripts, extract qualitative sentiment signals (e.g., subtle shifts in guidance confidence, unusual pushback from analysts on inventory backlogs), and write an automated strategy note for our long-short equity desk.
It worked spectacularly until it didn’t.
I made what I thought was a minor, cost-saving optimization to the system prompt: reducing the token footprint of our output JSON schema. The change went live on a Tuesday. By Thursday, our downstream parser began silently discarding mid-cap retail signals because the model, now constrained by the shorter prompt, stopped outputting the exact structural keys we used to calculate our leverage ratios. We missed a critical EBITDA-margin warning on a retail position, costing the book a 140-basis-point drawdown on a post-earnings gap-down.
This failure exposed three glaring vulnerabilities in our LLM-driven research stack:
1. Zero Backtest Traceability: If I modify an LLM prompt today, how do I reliably backtest our execution signal over three years of historical transcripts? I cannot trust historical backtest results unless I can guarantee the exact prompt version used to evaluate each past transcript.
2. Brittle Upstream Dependencies: Prompts are code, but we were treating them as configuration values. They drift, fail silently, and escape traditional unit tests.
3. Transient Execution Failures: LLM APIs rate-limit us, time out, and occasionally return garbled JSON. Running 500 parallel transcript analyses across multiple prompt steps without state durability means one failure at transcript 499 forces us to restart the entire batch.
To solve this, I rebuilt our research assistant around a decoupled, stateful engine using Temporal for orchestration and a strict schema-enforced DB for prompt versioning. Here is how we implemented it.
The Architecture: State-Driven Prompt Pipelines
We decoupled our system into three distinct layers:
1. The Prompt Registry: A PostgreSQL-backed schema where prompts are treated as immutable, semantic code artifacts with cryptographic hashes.
2. The Temporal Workflow Orchestrator: A durable execution engine that guarantees every execution step—fetching transcripts, querying the LLM, validating schemas, and writing database state—runs deterministically and survives system failures.
3. The Worker Pool: Ephemeral execution pods that run our model calls, handle token-bucket rate limits, and execute retry logic.
flowchart TD A["TranscriptIngest"] --> B["WorkflowStart"] B --> C["FetchPromptVersion"] C --> D["LLMExecutionActivity"] D --> E["JSONSchemaValidation"] E -->|"Success"| F["PersistSignal"] E -->|"ParseError"| G["FallbackOrRetry"]
Every signal written to our alpha database is tagged with a unique prompt_version_hash. If we need to run a backtest, we query the exact prompt version active at the time of that historical window, eliminating any backtest-to-production mismatch.
The Implementation
Let us dive straight into the code. We start with our schema definition for the prompt registry. This ensures that a prompt can never be modified in place. It is strictly append-only.
1. The Database Layer (SQL)
We track prompts using a SHA-256 hash of their raw text, system prompt, and temperature configuration.
id SERIAL PRIMARY KEY,
prompt_name VARCHAR(100) NOT NULL,
version_hash VARCHAR(64) NOT NULL UNIQUE,
system_prompt TEXT NOT NULL,
user_template TEXT NOT NULL,
temperature NUMERIC(3, 2) NOT NULL DEFAULT 0.0,
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX idx_prompt_name_hash ON prompt_registry (prompt_name, version_hash);
— Insert our baseline earnings analysis prompt
INSERT INTO prompt_registry (prompt_name, version_hash, system_prompt, user_template, temperature)
VALUES (
‘earnings_transcript_analyzer’,
‘8f4a39b2e0ad3d3f9b88930a931ef1a4ef4f1d43c2bcbe4df14a87265aef4023’,
‘You are a senior buy-side equity analyst. Analyze this earnings call transcript for qualitative risks. Output raw JSON ONLY with keys: “capex_risk” (bool), “margin_compression_risk” (bool), “management_confidence” (int 1-10), “risk_summary” (string).’,
‘Analyze the following transcript for ticker {ticker}: {transcript}’,
0.0
);
2. The Temporal Activities
Temporal workflows must be deterministic. This means you cannot execute network requests, read files, or fetch database configs directly within the workflow body. All non-deterministic operations must be wrapped inside Activities.
Here is our activities.py implementing the registry lookup and the actual LLM execution utilizing the openai and pydantic SDKs.
import hashlib
from typing import Dict, Any
from pydantic import BaseModel, Field
from temporalio import activity
import psycopg2
from openai import OpenAI
class AlphaSignalSchema(BaseModel):
capex_risk: bool = Field(description=“True if management signals rising capital expenditure risk”)
margin_compression_risk: bool = Field(description=“True if margins are expected to contract”)
management_confidence: int = Field(description=“Confidence score from 1 to 10”)
risk_summary: str = Field(description=“Brief analysis of the risk landscape discussed”)
# Define helper to compute our SHA-256 version hash
def compute_prompt_hash(system_prompt: str, user_template: str, temp: float) -> str:
payload = f”{system_prompt}|{user_template}|{temp}”
return hashlib.sha256(payload.encode(“utf-8”)).hexdigest()
@activity.defn
async def fetch_prompt_by_hash(prompt_name: str, version_hash: str) -> Dict[str, Any]:
“””Fetches a locked prompt from the database to ensure execution reproducibility.”””
conn = psycopg2.connect(os.environ[“DATABASE_URL”])
try:
with conn.cursor() as cur:
cur.execute(
“””
SELECT system_prompt, user_template, temperature
FROM prompt_registry
WHERE prompt_name = %s AND version_hash = %s
“””,
(prompt_name, version_hash)
)
row = cur.fetchone()
if not row:
raise ValueError(f”Prompt hash {version_hash} not found in registry.”)
return {
“system_prompt”: row[0],
“user_template”: row[1],
“temperature”: float(row[2]),
“version_hash”: version_hash
}
finally:
conn.close()
@activity.defn
async def execute_llm_inference(params: Dict[str, Any]) -> Dict[str, Any]:
“””Queries the LLM using the pinned prompt version parameters.”””
client = OpenAI(api_key=os.environ[“OPENAI_API_KEY”])
system_prompt = params[“prompt”][“system_prompt”]
user_template = params[“prompt”][“user_template”]
ticker = params[“ticker”]
transcript = params[“transcript”]
temperature = params[“prompt”][“temperature”]
formatted_user = user_template.format(ticker=ticker, transcript=transcript)
# We call the model, enforcing structured outputs for parser durability
response = client.beta.chat.completions.parse(
model=“gpt-4o-2024-08-06”,
messages=[
{“role”: “system”, “content”: system_prompt},
{“role”: “user”, “content”: formatted_user}
],
temperature=temperature,
response_format=AlphaSignalSchema,
timeout=45.0
)
parsed_response = response.choices[0].message.parsed
if not parsed_response:
raise ValueError(“Failed to parse response structure from LLM.”)
return {
“signal”: parsed_response.model_dump(),
“prompt_version_hash”: params[“prompt”][“version_hash”],
“model_used”: “gpt-4o-2024-08-06”,
“raw_usage”: {
“prompt_tokens”: response.usage.prompt_tokens,
“completion_tokens”: response.usage.completion_tokens
}
}
@activity.defn
async def persist_signal_to_db(result: Dict[str, Any], ticker: str) -> str:
“””Persists computed alpha signal tagged with prompt version hash.”””
conn = psycopg2.connect(os.environ[“DATABASE_URL”])
try:
with conn.cursor() as cur:
cur.execute(
“””
INSERT INTO signal_store (ticker, capex_risk, margin_compression_risk, management_confidence, risk_summary, prompt_version_hash, computed_at)
VALUES (%s, %s, %s, %s, %s, %s, NOW())
ON CONFLICT (ticker, prompt_version_hash) DO UPDATE
SET capex_risk = EXCLUDED.capex_risk,
margin_compression_risk = EXCLUDED.margin_compression_risk,
management_confidence = EXCLUDED.management_confidence,
risk_summary = EXCLUDED.risk_summary;
“””,
(
ticker,
result[“signal”][“capex_risk”],
result[“signal”][“margin_compression_risk”],
result[“signal”][“management_confidence”],
result[“signal”][“risk_summary”],
result[“prompt_version_hash”]
)
)
conn.commit()
return “SUCCESS”
except Exception as e:
conn.rollback()
raise e
finally:
conn.close()
3. The Orchestration Layer: Temporal Workflows
Here is where the durability guarantees happen. If our worker pod gets killed in the middle of executing an LLM call, Temporal’s event-sourcing re-runs the workflow up to the exact point of failure without calling the LLM API twice—saving us both cost and API rate-limit exhaustion.
from temporalio import workflow
from temporalio.common import RetryPolicy
# Import our activity signatures
with workflow.unsafe.imports_passed_through():
from activities import fetch_prompt_by_hash, execute_llm_inference, persist_signal_to_db
@workflow.defn
class TranscriptAnalysisWorkflow:
@workflow.run
async def run(self, ticker: str, transcript: str, target_prompt_hash: str) -> Dict[str, Any]:
# Configure exponential backoff for DB and external LLM interactions
llm_retry_policy = RetryPolicy(
initial_interval=timedelta(seconds=5),
backoff_coefficient=2.0,
maximum_interval=timedelta(seconds=60),
maximum_attempts=5,
non_retryable_error_types=[“ValueError”] # Do not retry bad logic or parse schema breaks
)
# Step 1: Fetch locked version of prompt
prompt_config = await workflow.execute_activity(
fetch_prompt_by_hash,
args=[“earnings_transcript_analyzer”, target_prompt_hash],
start_to_close_timeout=timedelta(seconds=10),
retry_policy=RetryPolicy(maximum_attempts=3)
)
# Step 2: Query LLM using locked parameters
inference_payload = {
“ticker”: ticker,
“transcript”: transcript,
“prompt”: prompt_config
}
inference_result = await workflow.execute_activity(
execute_llm_inference,
args=[inference_payload],
start_to_close_timeout=timedelta(seconds=90),
retry_policy=llm_retry_policy
)
# Step 3: Write signal to SQL database along with hash verification
persistence_status = await workflow.execute_activity(
persist_signal_to_db,
args=[inference_result, ticker],
start_to_close_timeout=timedelta(seconds=15),
retry_policy=RetryPolicy(maximum_attempts=5)
)
return {
“status”: persistence_status,
“ticker”: ticker,
“prompt_version”: target_prompt_hash,
“confidence_score”: inference_result[“signal”][“management_confidence”]
}
4. Running the Pipeline
To run this pipeline, start a Temporal Worker. This worker polls Temporal’s queue and handles execution.
from temporalio.client import Client
from temporalio.worker import Worker
from workflows import TranscriptAnalysisWorkflow
from activities import fetch_prompt_by_hash, execute_llm_inference, persist_signal_to_db
async def main():
# Connect to local or cloud temporal server
client = await Client.connect(“localhost:7233”)
# Instantiate the worker listening to the ‘research-tasks’ queue
worker = Worker(
client,
task_queue=“research-tasks”,
workflows=[TranscriptAnalysisWorkflow],
activities=[fetch_prompt_by_hash, execute_llm_inference, persist_signal_to_db]
)
print(“Worker pool successfully running. Listening for tasks…”)
await worker.run()
if __name__ == “__main__”:
asyncio.run(main())
To trigger a backtest or run a single analysis job deterministically:
from temporalio.client import Client
async def trigger_analysis():
client = await Client.connect(“localhost:7233”)
ticker = “WMT”
sample_transcript = “Our capital expenditure this quarter rose by 14% to support automated fulfillment. We face headwinds in consumer discretionary spend, leading to a possible 1% margin squeeze next half.”
# We specify the EXACT prompt version hash we want to evaluate this against
target_prompt_version = “8f4a39b2e0ad3d3f9b88930a931ef1a4ef4f1d43c2bcbe4df14a87265aef4023”
result = await client.execute_workflow(
“TranscriptAnalysisWorkflow”,
args=[ticker, sample_transcript, target_prompt_version],
id=f”analysis-{ticker}-{target_prompt_version}”,
task_queue=“research-tasks”
)
print(f”Workflow execution succeeded: {result}”)
if __name__ == “__main__”:
asyncio.run(trigger_analysis())
The Results: Determinism in Production
Transitioning to this architectural pattern solved our core data integrity problems overnight.
1. Reliable Backtesting and zero Signal Drift
We can now safely alter prompts for the active trading desk while historical analyses remain perfectly isolated. When we updated our earnings prompt to focus heavier on inventory levels (v2 with hash 7d2c88...), we ran parallel tests.
- We executed 500 transcripts under the old version (
8f4a39...) and new version (7d2c88...). - Because both versions were locked inside our registry database and orchestrated via explicit hash parameters, there was zero state pollution. We verified that our backtested alpha signals under version
8f4a39...achieved a 1.0 correlation with original runs made months ago.
2. Elimination of Transient Failures
Prior to using Temporal, a network timeout on transcript 42 out of 100 caused our naive loop scripts to halt entirely, requiring complex manual state management to resume without re-running (and re-paying for) the first 41 API requests.
With Temporal, we simulated a rate-limit error (HTTP 429) during a run of 15 batch files. The retry policies caught the rate-limiting exception and backed off gracefully over 30 seconds. The workflow completed successfully without losing context and with zero lines of custom retry logic inside our core LLM application code.
Workflow Execution Status: COMPLETED
Events:
1 WorkflowExecutionStarted
2 WorkflowTaskScheduled
3 WorkflowTaskStarted
4 WorkflowTaskCompleted
5 ActivityTaskScheduled [fetch_prompt_by_hash]
6 ActivityTaskStarted
7 ActivityTaskCompleted
8 ActivityTaskScheduled [execute_llm_inference]
9 ActivityTaskStarted
10 ActivityTaskFailed [HTTP 429 Rate Limit Exceeded – Attempt 1]
11 ActivityTaskScheduled [Retry Scheduled for +30s]
12 ActivityTaskStarted [Attempt 2]
13 ActivityTaskCompleted [LLM Output Returned Successfully]
14 ActivityTaskScheduled [persist_signal_to_db]
15 ActivityTaskCompleted
16 WorkflowExecutionCompleted
Lessons Learned
- Keep DB Transactions Out of Workflows: In early iterations, we fetched prompt configurations from our relational DB inside our workflow class using standard psycopg2 connectors. This broke Temporal’s core rule of determinism and caused workflow replay errors because database responses can shift over time. Always wrap database queries and network fetches in Activities.
- Commit System Prompts alongside Code: Although the SQL-registry works flawlessly, we found that committing our schema migrations as localized JSON files within our Git repos—and populating our database tables from these files during CI/CD steps—is the cleanest way to unify LLM prompt engineering with regular application development.
- Budget for Replay Storage: Temporal keeps track of every state change. For thousands of long transcripts, avoid passing huge text payloads as workflow input arguments directly. Instead, store the raw transcripts in an S3 bucket or high-performance database, and pass only safe object keys or database IDs as parameters to the workflow. Have your Activities fetch the heavy textual data when required.