phenoforge.agent

The LangGraph agent: decompose → check curated first → generate only for unresolved terms → human confirmation gate on anything generated → assemble. See phenoforge.agent.graph’s own docstring for the exact interrupt/resume contract.

phenoforge.agent.state

Pydantic state schema for the cohort-assembly LangGraph.

Fields multiple nodes append to across steps (seed_terms, term_results, pending_candidates) use Annotated[..., operator.add] reducers. Without one, LangGraph replaces a Pydantic state field wholesale on each node’s partial-update return rather than merging, so a later node’s return would silently clobber an earlier node’s contribution instead of accumulating. Each of these fields is single-writer in this graph (one node returns the whole list in one shot) — the reducer costs nothing now and keeps the door open for a later per-term fan-out (LangGraph’s Send API) without a state-shape change. confirmed_codes and final_concept_set are set exactly once each, by a single node, so they need no reducer.

class phenoforge.agent.state.SeedTermResult(**data)[source]

Bases: BaseModel

One seed clinical term’s curated-lookup outcome.

Variables:

  • term – The seed term as decomposed from the population description.

  • curated – Result of checking this term against the curated OHDSI Phenotype Library — set by check_curated().

  • resolved – True only if curated.concepts is non-empty. A curated miss (even one carrying its own explanatory UnmappableTerm) still leaves this False so the term falls through to generate().

Parameters:
term: str
curated: ConceptSet | None
resolved: bool
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class phenoforge.agent.state.PendingCandidate(**data)[source]

Bases: BaseModel

One generated-tier concept awaiting human confirmation.

Variables:

  • term – The seed term this candidate was generated for.

  • concept – The candidate concept itself.

Parameters:
term: str
concept: ConceptWithProvenance
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class phenoforge.agent.state.CohortAssemblyState(**data)[source]

Bases: BaseModel

Full state threaded through the cohort-assembly graph.

Variables:

  • population_description – The original free-text input, e.g. “adults with type 2 diabetes and diabetic nephropathy”. Set once by the caller.

  • seed_terms – Seed clinical terms extracted by decompose().

  • term_results – Per-term curated-lookup results, written by check_curated().

  • pending_candidates – generated-tier concepts awaiting confirmation, staged by generate().

  • confirmed_codes – concept_code values the human accepted during confirm() — cross-referenced against pending_candidates by assemble().

  • unmappable – Terms/candidates that ended up unresolved anywhere in the pipeline, accumulated across nodes.

  • final_concept_set – The assembled result, set once by assemble().

Parameters:
population_description: str
seed_terms: Annotated[list[str], <built-in function add>]
term_results: SeedTermResult], <built-in function add>]
pending_candidates: PendingCandidate], <built-in function add>]
confirmed_codes: list[str]
unmappable: UnmappableTerm], <built-in function add>]
final_concept_set: ConceptSet | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

phenoforge.agent.nodes

Graph node functions for cohort assembly: decompose, curated-check, generate, confirm, assemble.

Every node is a plain function taking CohortAssemblyState and returning a partial state update (a dict of the fields it changed), per LangGraph’s node contract. DB/retriever/LLM dependencies are bound at graph-build time via functools.partial (see phenoforge.agent.graph) rather than threaded through state, since state must stay Pydantic-serializable for the checkpointer — a duckdb.DuckDBPyConnection can’t cross it.

The decomposition LLM call is injectable (DecomposeFn), mirroring engine/dense.py’s EmbedFn pattern, so tests never make a real Anthropic API call. Unlike dense.py’s sentence_transformers, langgraph itself is a hard import here (this module is graph node code, always needed) — only langchain_anthropic is deferred into default_decomposer(), since that’s the piece a caller can avoid entirely by supplying their own decompose_fn.

class phenoforge.agent.nodes.DecomposedTerms(**data)[source]

Bases: BaseModel

Structured-output schema the decomposition LLM call populates.

Variables:

terms – Distinct seed clinical terms extracted from the population description, each independently searchable.

Parameters:

terms (list[str])

terms: list[str]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

phenoforge.agent.nodes.default_decomposer(model='claude-sonnet-5')[source]

Build a decomposition function backed by a real Claude call.

Imports langchain_anthropic on call, not at module import time, so import phenoforge.agent.nodes never requires the agent extra to be installed (mirrors default_embedder()). Defaults to a quality-over-cost model since decomposition accuracy is the capability this version exists to add (see phenoforge.agent).

Parameters:

model (str) – Anthropic model id.

Returns:

A function mapping a population description to a list of seed clinical terms (see DecomposeFn).

Return type:

DecomposeFn

Raises:

ImportError – If langchain-anthropic is not installed.

phenoforge.agent.nodes.decompose(state, *, decompose_fn)[source]

Extract distinct seed clinical terms from the population description.

Parameters:
Returns:

Partial state update: {"seed_terms": [...]}.

Return type:

dict[str, object]

phenoforge.agent.nodes.check_curated(state, *, con, library_dir, bm25, dense)[source]

Check every seed term against the curated OHDSI Phenotype Library.

Delegates to find_curated_definition() per term, which matches on the bundled cohorts’ own resolved code content (via the same bm25/dense retrievers generate() uses) rather than their display names. A term resolves (SeedTermResult.resolved = True) only if the curated lookup returns at least one concept; a curated miss (even one carrying its own explanatory UnmappableTerm) leaves it unresolved for generate() to pick up.

Parameters:

  • state (CohortAssemblyState) – Current graph state; reads seed_terms.

  • con (DuckDBPyConnection) – Open connection to vocab.duckdb.

  • library_dir (Path) – Fetched phenotype library directory.

  • bm25 (BM25Retriever) – Built BM25 retriever, shared with generate().

  • dense (DenseRetriever | None) – Built dense retriever, or None to degrade to BM25-only, shared with generate().

Returns:

Partial state update: {"term_results": [...]}.

Return type:

dict[str, object]

phenoforge.agent.nodes.generate(state, *, bm25, dense, k=10)[source]

Generate candidate concepts (hybrid search) for every unresolved seed term.

Only runs for terms check_curated() left unresolved. Every candidate is generated-tier by construction — that is hybrid_search()’s contract — and is staged into pending_candidates for human confirmation rather than folded directly into the result, since that’s exactly the tier that requires it.

Parameters:
Returns:

Partial state update: {"pending_candidates": [...], "unmappable": [...]}.

Return type:

dict[str, object]

phenoforge.agent.nodes.confirm(state)[source]

Pause the graph for human review of pending generated candidates.

Calls langgraph.types.interrupt() with the pending payload (see phenoforge.agent.graph’s module docstring for the exact payload/resume shape); the caller resumes with Command(resume=<list of accepted concept_codes>). A no-op if there is nothing pending — lets build_graph()’s conditional edge route straight to assemble in the all-curated case without ever reaching this node, but staying defensive here too.

Parameters:

state (CohortAssemblyState) – Current graph state; reads pending_candidates.

Returns:

Partial state update: {"confirmed_codes": [...]}.

Return type:

dict[str, object]

phenoforge.agent.nodes.assemble(state)[source]

Assemble the final ConceptSet from curated results and confirmed candidates.

Curated concepts from every resolved term are included unconditionally, along with every curated per-term unmappable entry — a resolved term’s curated match can itself carry unmappable SNOMED items (e.g. high-fanout-skipped), and those must not be silently dropped just because the term overall resolved. Generated candidates are included only if their concept_code is in confirmed_codes; rejected candidates become UnmappableTerm entries. Deduplicated by concept_id.

Parameters:

state (CohortAssemblyState) – Current graph state; reads term_results, pending_candidates, confirmed_codes, unmappable.

Returns:

Partial state update: {"final_concept_set": ConceptSet(...)}.

Return type:

dict[str, object]

phenoforge.agent.graph

LangGraph wiring for tiered-trust cohort assembly.

Node functions in agent/nodes.py are pure-ish (state in, partial state out); this module is the only place DB/retriever/LLM dependencies get bound to them (via functools.partial) and the only place graph topology lives.

Topology: decompose -> check_curated -> [generate -> confirm] -> assemble, with the bracketed segment skipped via conditional edges when every seed term resolved via curated matching.

Interrupt/resume payload shape (the contract between confirm() and any caller, e.g. scripts/run_agent.py):

  • Outbound (what graph.invoke(...) returns via result["__interrupt__"][0].value when paused): {"pending": [{"term": str, "concept_code": str, "concept_name": str, "source": str}, ...]}.

  • Inbound (what Command(resume=...) carries to resume): list[str] — the concept_code values the human accepted, a subset of the outbound payload’s codes. Chosen over a per-concept accept/reject dict because the CLI’s natural input (a comma-separated list of accepted codes, or “all”/”none”) maps directly to it with no intermediate structure, and assemble only needs to know “accepted or not” per candidate.

phenoforge.agent.graph.build_graph(con, library_dir, decompose_fn=None, dense=None, checkpointer=None)[source]

Build and compile the cohort-assembly graph.

Parameters:

  • con (duckdb.DuckDBPyConnection) – Open connection to vocab.duckdb, bound into check_curated() and (via a fresh BM25Retriever) generate().

  • library_dir (Path) – Fetched phenotype library directory, bound into check_curated().

  • decompose_fn (DecomposeFn | None) – Decomposition function; defaults to default_decomposer() (real Claude call). Tests inject a fake returning deterministic terms.

  • dense (DenseRetriever | None) – Pre-built dense retriever, or None to degrade generate to BM25-only, matching search_concepts’s existing fallback behavior.

  • checkpointer (BaseCheckpointSaver[Any] | None) – Checkpointer backing interrupt/ Command(resume=...). Required for the confirm node’s pause/resume to work at all — defaults to an in-memory InMemorySaver (with this project’s Pydantic types allowlisted for its serializer — see _CHECKPOINT_ALLOWED_TYPES) if not given, fine for a single-process CLI run but does not persist across process restarts.

Returns:

A compiled graph, ready for .invoke(state, config={"configurable": {"thread_id": ...}}).

Return type:

CompiledStateGraph