A compositional pipeline for generating sell-side research reports from structured inputs. Owns context assembly → LLM generation → validation. Data fetching and material retrieval live in sibling services and reach this pipeline via API.
The design goal is simple: let an LLM write the prose while the pipeline keeps the numbers grounded, the economic logic honest, and the style consistent across report types.
Brief + DataPack + MaterialPack + HouseView
↓
Context Builder ← declarative recipes, reusable blocks
↓
Generator ← single LLM call via OpenRouter
↓
Validation ← (next) numeric grounding, structure,
↓ logic consistency, house-view reconciliation
Report + ValidationTrace
Three layers:
- Blocks — stateless units that render a slice of context (
fact_table,derived_metrics,framework,house_view,exemplars,style_guide, …). Each declares its required inputs; blocks whose inputs are absent are skipped and recorded in the trace. - Recipes — YAML specs listing which blocks (with what config) compose a given report type. Adding a new report type = new YAML, no code.
- Params — runtime knobs (
language,reader_tier) threaded through every block. Language affects glossary scope and exemplar selection; it does not create parallel code paths.
Three recipes ship out of the box: brief_comment, deep_research,
trading_daily. Context is rendered as XML (Claude responds better to tagged
structure than to prose or JSON for long context).
One LLM call, whole report out. Style coherence matters more on sell-side than composition convenience, so the generator does not interleave templated prose with LLM prose.
Anti-hallucination is enforced at the edges, not the middle:
- The prompt ships a
FactTablewith stable IDs; every numeric claim in the output must carry[F-xxx]. - A post-processor (next in the pipeline) validates every citation, rejects invented IDs or uncited numbers, and injects the static disclaimer.
- Validators catch the rest.
The provider is OpenRouter. Any model you can route through OpenRouter is
callable via GeneratorParams(model="...").
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"Python 3.11+.
Build a context for a CPI print:
python examples/smoke_test.pyGenerate a full report (requires OPENROUTER_API_KEY):
export OPENROUTER_API_KEY=sk-or-...
python examples/generate_sample.py brief_comment anthropic/claude-sonnet-4.5Run the test suite:
pytestfrom datetime import datetime, timezone
from researchflow.context import (
BlockInputs, Brief, ContextParams, DataPack, HouseView, build,
)
from researchflow.generation import GeneratorParams, generate
brief = Brief(
event_id="us_cpi_2026_03",
event_name="US CPI — March 2026",
release_time=datetime(2026, 4, 10, 8, 30, tzinfo=timezone.utc),
report_type="us_cpi",
)
data_pack = DataPack(
event_id=brief.event_id,
payload={"facts": [
{"id": "F-CPI-HEAD-YOY", "label": "Headline CPI YoY",
"unit": "%", "actual": 3.1, "consensus": 3.2, "prior": 3.0, "tier": 1},
]},
)
house_view = HouseView(
version="2026-04-15",
as_of=datetime(2026, 4, 15, tzinfo=timezone.utc),
content={"base_case": "Fed cuts 75bp in 2026.", "tone_lean": "modestly_dovish"},
)
ctx = build(
"brief_comment",
ContextParams(language="en", reader_tier="pm"),
BlockInputs(brief=brief, data_pack=data_pack, house_view=house_view,
extras={"framework": {}, "exemplar_pool": []}),
)
report = generate(ctx, GeneratorParams(model="anthropic/claude-sonnet-4.5"))
print(report.raw_text)
print(report.fact_citations)- Add
src/researchflow/context/recipes/<name>.yaml. - Pick blocks and config. Example minimal recipe:
name: my_report
token_budget: 6000
renderer: xml
blocks:
- brief
- fact_table: { include_components: minimal }
- derived_metrics: { include_components: minimal }
- framework: { glossary: minimal, sign_map: true }
- house_view: { depth: summary }
- exemplars: { k: 2 }
- style_guide:
length_words: [300, 500]
sections: [bottom_line, analysis, risks]- Call
build("my_report", params, inputs).
No code changes needed unless you want a new block type.
Implement the ContextBlock protocol and decorate with @register_block:
from typing import Any, ClassVar
from researchflow.context.blocks.base import register_block, rough_tokens
from researchflow.context.contracts import BlockInputs, ContextParams, RenderedBlock
@register_block
class CrossAssetBlock:
name: ClassVar[str] = "cross_asset"
required_inputs: ClassVar[set[str]] = {"data_pack"}
def render(self, inputs: BlockInputs, params: ContextParams,
config: dict[str, Any]) -> RenderedBlock:
content = {"snapshot": inputs.data_pack.payload.get("cross_asset", {})}
return RenderedBlock(name=self.name, content=content,
token_estimate=rough_tokens(content))Import it from researchflow/context/blocks/__init__.py so the registration
fires on package import. It is now usable in any recipe.
src/researchflow/
├── context/
│ ├── contracts.py pydantic models
│ ├── blocks/ one module per block + base registry
│ ├── renderers/xml.py default renderer
│ ├── recipe_loader.py YAML → Recipe
│ ├── recipes/ brief_comment, deep_research, trading_daily
│ └── assembler.py build(recipe, params, inputs) → Context
├── generation/
│ ├── contracts.py GeneratorParams, Report, GenerationTrace
│ ├── prompts.py system prompt
│ ├── provider.py OpenRouter client
│ └── generator.py generate(context, params) → Report
├── validation/
│ ├── contracts.py Severity, ValidationIssue, ValidationReport
│ ├── postprocessor.py post_process(report, disclaimer=...)
│ ├── judge.py shared LLM-judge helper (OpenAI-compatible client)
│ ├── validators/ numeric_grounding, structure, citation_integrity,
│ │ logic_consistency, house_view_reconciliation
│ └── pipeline.py validate(report, context, judge_client=...)
│ → ValidationReport
├── eval/
│ ├── contracts.py Fixture, Expected, StageScore, FixtureScorecard, RunSummary
│ ├── fixtures.py load fixtures from eval/fixtures/*/fixture.yaml
│ ├── mock_client.py QueueClient for deterministic replay
│ ├── runners.py one runner per stage, returns (output, artifacts)
│ ├── scorers.py one scorer per stage, compares to fixture.expected
│ ├── harness.py run_fixture / run_all — writes per-stage artifacts
│ └── cli.py `python -m researchflow.eval run|list ...`
├── flow.py run_research_flow(block_inputs, params, recipe, ...)
│ → FlowResult; optionally writes live audit tree
├── resolve.py TopicResolver + LocalRegistry (topic string → Brief)
├── clients/ httpx clients for macro-data-service, rag-service,
│ plus HouseViewLoader (YAML on disk)
└── server/
├── main.py FastAPI app + Dependencies wiring
├── config.py pydantic-settings loaded from env / .env
├── auth.py optional Bearer-token dependency
├── schemas.py ResearchRequest, ResolveRequest, responses
├── routes.py endpoint handlers (thin; wiring only)
├── resources.py framework + exemplar + disclaimer loaders
├── persistence.py read artifacts back, whitelist path-guarded
└── __main__.py `researchflow-server` console script → uvicorn
The harness lives outside the production flow. It runs the same context → generate → post-process → validate pipeline over fixtures and writes, for every stage, the inputs that entered the stage, the outputs that left it, and a trace. That's what "every step auditable" means here: a human or a script can reconstruct exactly what the pipeline did for any fixture on any run.
A fixture is a YAML file that carries the complete input state for a scheduled event plus declared expectations per stage:
id: us_cpi_2026_03
recipe: brief_comment
params: { language: en, reader_tier: pm }
inputs:
brief: {...}
data_pack: {event_id, payload: {facts: [...]}}
house_view: {version, as_of, content: {base_case, tone_lean}}
extras: {framework: {...}, exemplar_pool: [...]}
expected:
context: { blocks_rendered: [...], must_contain_fact_ids: [...], max_token_estimate: 4000 }
report: { must_cite: [...], word_count: [150, 250], must_have_sections: [...] }
postprocess: { disclaimer_injected: true }
validation: { passed: true, max_errors: 0, require_codes: [], forbid_codes: [] }
mock_responses: # consumed in order: generator, then each LLM judge
- "## Bottom line\n...{{DISCLAIMER}}"
- '{"violations": []}'
- '{"violations": []}'Any expected.* block is optional; the harness scores only the stages
that have declared expectations.
# List available fixtures
python -m researchflow.eval list
# Run everything (uses mock_responses; no network, deterministic)
python -m researchflow.eval run
# Single fixture, explicit run id
python -m researchflow.eval run --fixture us_cpi_2026_03 --run-id smoke
# Live-mode (no mock_responses; real OpenRouter calls):
# leave mock_responses: [] in the fixture and pass a real client in Python.Exit code is non-zero if any fixture fails — friendly to CI.
Every run writes an immutable tree under eval/runs/<run_id>/:
eval/runs/<run_id>/
├── manifest.json run id, git commit, fixture list, pass rate
├── summary.json full RunSummary
└── <fixture_id>/
├── 01_context/
│ ├── inputs.json brief + data_pack + house_view + extras
│ ├── params.json ContextParams
│ ├── output.xml rendered context (what the LLM sees)
│ ├── blocks.json per-block name / token estimate / trace
│ └── trace.json ContextTrace
├── 02_generation/
│ ├── prompt_user.xml user-message content
│ ├── output.md raw LLM output
│ ├── fact_citations.json extracted [F-xxx] ids
│ └── trace.json GenerationTrace (model, timing, usage)
├── 03_postprocess/
│ ├── input.md pre-disclaimer-injection
│ └── output.md post-disclaimer-injection
├── 04_validation/
│ ├── input.md text validators ran against
│ └── report.json ValidationReport (issues, run/skipped)
└── scorecard.json per-stage pass/fail + metrics
manifest.json captures the git commit SHA so any run can be
reproduced at the exact code state that produced it.
| Stage | Metrics |
|---|---|
| context | blocks rendered, fact-ids present, token estimate within budget |
| generation | required citations present, word count bounds, required sections |
| postprocess | disclaimer placeholder replaced |
| validation | passed flag matches, error-count bound, required/forbidden issue codes |
Adding a new metric is one function in scorers.py and one expected key
in the fixture — no changes to the harness or runners.
The same pipeline runs as a FastAPI service. Callers send the minimum
they naturally know — a raw topic string and a recipe name; the service
resolves the topic to a canonical Brief, fetches the DataPack from
macro-data-service and the MaterialPack from rag-service, calls
OpenRouter, post-processes, validates, and returns the report plus a
stage-by-stage summary. Every run writes the same audit tree the eval
harness produces.
pip install -e ".[service]"Pulls fastapi, uvicorn[standard], httpx, pydantic-settings.
| Var | Purpose |
|---|---|
RESEARCHFLOW_API_TOKEN |
Optional inbound bearer token; unset = open |
ANALYST_MACRO_DATA_BASE_URL / _API_TOKEN |
macro-data-service |
ANALYST_RAG_BASE_URL / _API_TOKEN |
rag-service |
OPENROUTER_API_KEY |
required to generate reports |
OPENROUTER_GENERATOR_MODEL |
default anthropic/claude-sonnet-4.5 |
OPENROUTER_JUDGE_MODEL |
default anthropic/claude-haiku-4.5 |
RESEARCHFLOW_RUNS_DIR |
live audit tree (default eval/runs/live) |
HOUSE_VIEW_PATH |
YAML (default config/house_view.yaml) |
TOPIC_REGISTRY_PATH |
YAML (default config/topics.yaml) |
FRAMEWORK_DIR |
per-report-type YAMLs (default config/frameworks) |
EXEMPLAR_DIR |
per-report-type JSON files (default config/exemplars) |
DISCLAIMER_PATH |
plain-text disclaimer; optional |
| Method + path | What |
|---|---|
POST /v1/research |
Topic → validated report + stage summary |
POST /v1/topics/resolve |
Topic → canonical Brief (for disambiguation) |
GET /v1/recipes |
List packaged recipes |
GET /v1/research/runs/{run_id} |
Per-run flow summary JSON |
GET /v1/research/runs/{run_id}/{stage}/{artifact} |
One audit artifact |
GET /v1/health |
Liveness + per-dependency readiness |
curl -X POST http://localhost:8000/v1/research \
-H 'Content-Type: application/json' \
-H 'Authorization: Bearer <optional>' \
-d '{
"topic": "US CPI March 2026",
"recipe": "brief_comment",
"params": {"language": "en", "reader_tier": "pm"}
}'Response carries { run_id, resolved_brief, report, validation, stage_summary, run_dir }.
All stage artifacts are available at /v1/research/runs/{run_id}/... and on disk
under ${RESEARCHFLOW_RUNS_DIR}/{run_id}/.
When the caller already has a fully-assembled BlockInputs (e.g. replaying
a fixture, or testing without sibling services), send it under
inputs_override — the service skips topic resolution and sibling fetches:
{
"topic": "n/a",
"recipe": "brief_comment",
"inputs_override": {
"brief": {...},
"data_pack": {"event_id": "...", "payload": {"facts": [...]}},
"house_view": {...},
"extras": {"framework": {...}, "exemplar_pool": []}
}
}researchflow-server # console script → uvicorn on 0.0.0.0:8000
# or:
uvicorn researchflow.server.main:app --reload
# or:
docker build -t researchflow-api . && \
docker run -p 8000:8000 --env-file .env researchflow-apiFour layers, ordered cheapest-first, each producing structured issues
(error hard-fails the report; warning flags for human review):
| Validator | Kind | Catches |
|---|---|---|
numeric_grounding |
deterministic | decimal numbers that aren't cited, cited numbers that don't match the fact, citations to unknown fact ids |
structure |
deterministic | missing or out-of-order sections, length bounds, un-injected disclaimer placeholder |
citation_integrity |
deterministic | [F-xxx] citations that don't resolve to any fact in the context |
logic_consistency |
LLM judge | directional claims that violate the framework sign_map |
house_view_reconciliation |
LLM judge | contradictions of the house view that aren't explicitly flagged as divergences |
LLM judges use an OpenAI-compatible client (OpenRouter by default) and are
skipped unless a judge_client is passed to validate(). They return
WARNING-level issues rather than hard-failing; human reviewers triage.
from researchflow.generation.provider import openrouter_client
from researchflow.validation import post_process, validate
# 1. Inject disclaimer (deterministic, safe)
finalized = post_process(report, disclaimer="For institutional use only...")
# 2. Run validators. Pass judge_client to enable LLM judges.
vr = validate(finalized, context, judge_client=openrouter_client())
if not vr.passed:
for issue in vr.errors():
print(f"[{issue.validator}:{issue.code}] {issue.message}")
for warn in vr.warnings():
print(f"[WARN {warn.validator}:{warn.code}] {warn.message}")- Context builder — implemented, 5 tests.
- Generator — implemented, 4 tests.
- Post-processor — disclaimer injection landed (canonicalization deferred).
- Validation pipeline — 3 deterministic validators + 2 LLM judges, 40 tests.
- Eval harness — per-stage runners, scorers, artifacts, CLI.
- HTTP service — FastAPI wrapper, sibling clients, topic resolver, auth, Docker. 93 tests across all layers.
MIT. See LICENSE.