From 5a253c49ee75be4d724b3f284b895d9629d42544 Mon Sep 17 00:00:00 2001 From: Maira Salazar Date: Mon, 6 Jul 2026 15:20:00 +0200 Subject: [PATCH 1/4] refactor(extraction): move shared LLM agent builder Move the LLM agent construction logic from `extract_metadata` into a reusable `_llm.build_agent` helper, and use it in the existing metadata extraction activity. --- app/activities/_llm.py | 66 ++++++++++++++++++++++++++++++ app/activities/extract_metadata.py | 60 +-------------------------- 2 files changed, 68 insertions(+), 58 deletions(-) create mode 100644 app/activities/_llm.py diff --git a/app/activities/_llm.py b/app/activities/_llm.py new file mode 100644 index 0000000..aeff55a --- /dev/null +++ b/app/activities/_llm.py @@ -0,0 +1,66 @@ +# SPDX-FileCopyrightText: 2026 CERN. +# SPDX-License-Identifier: MIT + +"""Shared LLM agent builder.""" + +from typing import TypeVar + +from pydantic import BaseModel +from pydantic_ai import Agent, PromptedOutput +from pydantic_ai.models.openai import OpenAIChatModel, OpenAIChatModelSettings +from pydantic_ai.providers.litellm import LiteLLMProvider +from pydantic_ai.providers.ollama import OllamaProvider + +from app.config import get_settings + +T = TypeVar("T", bound=BaseModel) + +# Extra instruction pieces for prompted output (models without native tool calls): +# cap reasoning, then force a single JSON object. +_REASONING_LOW = "Reasoning: low" +_JSON_ONLY = ( + "Respond immediately; do not deliberate. " + "Reply with exactly one JSON object matching the schema and nothing else." +) + + +def _parse_llm(llm: str) -> tuple[str, str]: + provider, sep, model_name = llm.partition("/") + if not sep: + raise ValueError("Invalid LLM; expected '/'") + provider = provider.strip().lower() + model_name = model_name.strip() + if provider not in {"litellm", "ollama"}: + raise ValueError("Invalid LLM; provider must be 'litellm' or 'ollama'") + if not model_name: + raise ValueError("Invalid LLM; model name is missing") + return provider, model_name + + +def build_agent(llm: str, output_type: type[T], instructions: str) -> Agent[None, T]: + """Build the extraction agent for `llm` from the configured settings.""" + settings = get_settings() + cfg = settings.llm_settings + provider_name, model_name = _parse_llm(llm) + + if provider_name == "ollama": + provider = OllamaProvider( + base_url=settings.ollama_base_url, api_key=settings.ollama_api_key + ) + else: + provider = LiteLLMProvider( + api_base=settings.litellm_api_base, api_key=settings.litellm_api_key + ) + + model = OpenAIChatModel( + model_name=model_name, + provider=provider, + settings=OpenAIChatModelSettings(**cfg.model), + ) + if cfg.output == "prompted": + return Agent[None, T]( + model, + instructions=[_REASONING_LOW, instructions, _JSON_ONLY], + output_type=PromptedOutput(output_type), + ) + return Agent[None, T](model, instructions=instructions, output_type=output_type) diff --git a/app/activities/extract_metadata.py b/app/activities/extract_metadata.py index 2f245bf..50cb989 100644 --- a/app/activities/extract_metadata.py +++ b/app/activities/extract_metadata.py @@ -10,13 +10,10 @@ from idutils.normalizers import normalize_doi from idutils.validators import is_doi from pydantic import BaseModel, Field -from pydantic_ai import Agent, PromptedOutput -from pydantic_ai.models.openai import OpenAIChatModel, OpenAIChatModelSettings -from pydantic_ai.providers.litellm import LiteLLMProvider -from pydantic_ai.providers.ollama import OllamaProvider from temporalio import activity from temporalio.common import RetryPolicy +from app.activities._llm import build_agent from app.config import get_settings from app.observability import propagate_langfuse_context from app.schemas.metadata_suggestions import ExtractedMetadata, MetadataSuggestions @@ -30,19 +27,6 @@ ) -def _parse_llm(llm: str) -> tuple[str, str]: - provider, sep, model_name = llm.partition("/") - if not sep: - raise ValueError("Invalid LLM; expected '/'") - provider = provider.strip().lower() - model_name = model_name.strip() - if provider not in {"litellm", "ollama"}: - raise ValueError("Invalid LLM; provider must be 'litellm' or 'ollama'") - if not model_name: - raise ValueError("Invalid LLM; model name is missing") - return provider, model_name - - class ExtractMetadataRequest(BaseModel): """Request to generate metadata suggestions from document text.""" @@ -109,46 +93,6 @@ def _clear_absent_fields(output: ExtractedMetadata, text: str) -> None: output.doi = None -# Extra instruction pieces for prompted output (no native tool calls): cap -# reasoning, then force a single JSON object. -_REASONING_LOW = "Reasoning: low" -_JSON_ONLY = ( - "Respond immediately; do not deliberate. " - "Reply with exactly one JSON object matching the schema and nothing else." -) - - -def _build_agent(llm: str) -> Agent[None, ExtractedMetadata]: - """Build the extraction agent for `llm` from the configured settings.""" - settings = get_settings() - cfg = settings.llm_settings - provider_name, model_name = _parse_llm(llm) - - if provider_name == "ollama": - provider = OllamaProvider( - base_url=settings.ollama_base_url, api_key=settings.ollama_api_key - ) - else: - provider = LiteLLMProvider( - api_base=settings.litellm_api_base, api_key=settings.litellm_api_key - ) - - model = OpenAIChatModel( - model_name=model_name, - provider=provider, - settings=OpenAIChatModelSettings(**cfg.model), - ) - if cfg.output == "prompted": - return Agent[None, ExtractedMetadata]( - model, - instructions=[_REASONING_LOW, INSTRUCTIONS, _JSON_ONLY], - output_type=PromptedOutput(ExtractedMetadata), - ) - return Agent[None, ExtractedMetadata]( - model, instructions=INSTRUCTIONS, output_type=ExtractedMetadata - ) - - @activity.defn async def extract_metadata_with_llm( request: ExtractMetadataRequest, @@ -159,7 +103,7 @@ async def extract_metadata_with_llm( # No usable text: skip the LLM entirely rather than let it fabricate. return MetadataSuggestions(suggestions=[]) - agent = _build_agent(get_settings().llm) + agent = build_agent(get_settings().llm, ExtractedMetadata, INSTRUCTIONS) with propagate_langfuse_context(context, trace_name="extract_metadata"): result = await agent.run(request.text) From f82351304526b98886c3c8ca97a3be13d054f4f7 Mon Sep 17 00:00:00 2001 From: Maira Salazar Date: Mon, 6 Jul 2026 15:19:19 +0200 Subject: [PATCH 2/4] feat(workflow): add a funding check workflow Add a new `check_funding_relevance` activity. The new activity assesses whether a record's title and description match a grant description, guided by a caller-supplied rule. Wires up the corresponding Temporal workflow and registers it under the `check_funding_relevance` key. --- app/activities/__init__.py | 3 + app/activities/check_funding_relevance.py | 42 +++++++++ .../check_funding_relevance_workflow.py | 90 +++++++++++++++++++ app/workflows/registry.py | 10 +++ 4 files changed, 145 insertions(+) create mode 100644 app/activities/check_funding_relevance.py create mode 100644 app/workflows/check_funding_relevance_workflow.py diff --git a/app/activities/__init__.py b/app/activities/__init__.py index 8ec6acc..f33c71b 100644 --- a/app/activities/__init__.py +++ b/app/activities/__init__.py @@ -3,6 +3,7 @@ """Activities for the Orcha application.""" +from .check_funding_relevance import check_funding_relevance from .extract_metadata import extract_metadata_with_llm from .extract_pdf_content import extract_pdf_text from .update_workflow import update_workflow @@ -11,6 +12,7 @@ extract_pdf_text, extract_metadata_with_llm, update_workflow, + check_funding_relevance, ] __all__ = [ @@ -18,4 +20,5 @@ "extract_pdf_text", "extract_metadata_with_llm", "update_workflow", + "check_funding_relevance", ] diff --git a/app/activities/check_funding_relevance.py b/app/activities/check_funding_relevance.py new file mode 100644 index 0000000..1224ed5 --- /dev/null +++ b/app/activities/check_funding_relevance.py @@ -0,0 +1,42 @@ +# SPDX-FileCopyrightText: 2026 CERN. +# SPDX-License-Identifier: MIT + +"""LLM-based funding relevance check activity.""" + +from pydantic import BaseModel, Field +from temporalio import activity + +from app.activities._llm import build_agent +from app.config import get_settings + + +class CheckFundingRelevanceRequest(BaseModel): + """Request to check if record metadata matches an award description.""" + + award_description: str = Field(description="Official EU grant description") + metadata: dict[str, object] = Field(description="Record metadata") + rule: str = Field(description="Instructions to decide if there is a match") + + +class CheckFundingRelevanceResponse(BaseModel): + """Result of the funding relevance check.""" + + match: bool = Field(description="Whether the record is relevant to the grant") + message: str = Field(description="Explanation of the decision") + + +@activity.defn +async def check_funding_relevance( + request: CheckFundingRelevanceRequest, +) -> CheckFundingRelevanceResponse: + """Use an LLM to assess if a record's metadata matches a grant description.""" + agent = build_agent(get_settings().llm, CheckFundingRelevanceResponse, request.rule) + + prompt = ( + f"Grant description:\n{request.award_description}\n\n" + f"Record title:\n{request.metadata.get('title')}\n\n" + f"Record description:\n{request.metadata.get('description')}" + ) + + result = await agent.run(prompt) + return result.output diff --git a/app/workflows/check_funding_relevance_workflow.py b/app/workflows/check_funding_relevance_workflow.py new file mode 100644 index 0000000..53cfe12 --- /dev/null +++ b/app/workflows/check_funding_relevance_workflow.py @@ -0,0 +1,90 @@ +# SPDX-FileCopyrightText: 2026 CERN. +# SPDX-License-Identifier: MIT + +from datetime import timedelta + +from temporalio import workflow + +from app.activities.check_funding_relevance import ( + CheckFundingRelevanceRequest, + CheckFundingRelevanceResponse, + check_funding_relevance, +) +from app.activities.update_workflow import ( + UPDATE_WORKFLOW_RETRY_POLICY, + WorkflowUpdateRequest, + update_workflow, +) +from app.database.models import WorkflowStatus +from app.workflows.specs import WorkflowContext, WorkflowParams + + +class CheckFundingRelevanceParams(WorkflowParams): + """User-provided params for the check_funding_relevance workflow.""" + + metadata: dict[str, object] + award_description: str + rule: str + + +@workflow.defn +class CheckFundingRelevance: + """Workflow that checks if a record's metadata matches an EU grant description.""" + + @workflow.run + async def run( + self, + context: WorkflowContext, + params: CheckFundingRelevanceParams, + ) -> CheckFundingRelevanceResponse: + """Execute the funding relevance check.""" + try: + await workflow.execute_activity( + update_workflow, + WorkflowUpdateRequest( + public_id=context.workflow_id, + tenant_id=context.tenant_id, + start_time=workflow.now(), + ), + start_to_close_timeout=timedelta(minutes=1), + retry_policy=UPDATE_WORKFLOW_RETRY_POLICY, + ) + + result = await workflow.execute_activity( + check_funding_relevance, + CheckFundingRelevanceRequest( + metadata=params.metadata, + award_description=params.award_description, + rule=params.rule, + ), + start_to_close_timeout=timedelta(minutes=2), + ) + except Exception: + await workflow.execute_activity( + update_workflow, + WorkflowUpdateRequest( + workflow_id=context.workflow_id, + tenant_id=context.tenant_id, + status=WorkflowStatus.ERROR, + result=None, + end_time=workflow.now(), + ), + start_to_close_timeout=timedelta(minutes=1), + retry_policy=UPDATE_WORKFLOW_RETRY_POLICY, + ) + raise + + await workflow.execute_activity( + update_workflow, + WorkflowUpdateRequest( + public_id=context.workflow_id, + tenant_id=context.tenant_id, + status=WorkflowStatus.SUCCESS, + result=result.model_dump(), + end_time=workflow.now(), + ), + start_to_close_timeout=timedelta(minutes=1), + retry_policy=UPDATE_WORKFLOW_RETRY_POLICY, + ) + + return result diff --git a/app/workflows/registry.py b/app/workflows/registry.py index d9c8a3a..93dcbc5 100644 --- a/app/workflows/registry.py +++ b/app/workflows/registry.py @@ -5,6 +5,10 @@ from __future__ import annotations +from app.workflows.check_funding_relevance_workflow import ( + CheckFundingRelevance, + CheckFundingRelevanceParams, +) from app.workflows.extract_metadata_workflow import ( ExtractMetadata, ExtractMetadataParams, @@ -19,6 +23,12 @@ task_queue=DEFAULT_TASK_QUEUE, id_prefix="extract-metadata", ), + "check_funding_relevance": WorkflowSpec( + workflow_cls=CheckFundingRelevance, + params_model=CheckFundingRelevanceParams, + task_queue=DEFAULT_TASK_QUEUE, + id_prefix="check-funding-relevance", + ), } From 148e5671ef53fa6dd0bd374058038696fec0f173 Mon Sep 17 00:00:00 2001 From: Pablo Tamarit Date: Tue, 14 Jul 2026 15:46:53 +0200 Subject: [PATCH 3/4] fix(workflow): pass public_id instead of workflow_id to WorkflowUpdateRequest --- app/workflows/check_funding_relevance_workflow.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/workflows/check_funding_relevance_workflow.py b/app/workflows/check_funding_relevance_workflow.py index 53cfe12..7cd6c7d 100644 --- a/app/workflows/check_funding_relevance_workflow.py +++ b/app/workflows/check_funding_relevance_workflow.py @@ -63,7 +63,7 @@ async def run( await workflow.execute_activity( update_workflow, WorkflowUpdateRequest( - workflow_id=context.workflow_id, + public_id=context.workflow_id, tenant_id=context.tenant_id, status=WorkflowStatus.ERROR, result=None, From 7299e81c4a9774fe218d469b39c978065601d706 Mon Sep 17 00:00:00 2001 From: Maira Salazar Date: Tue, 21 Jul 2026 17:40:21 +0200 Subject: [PATCH 4/4] fix(workflow): add propagate_lagfuse_context Adds the context and the propagate_lagfuse_context to the check funding relevance activity, passing the context from the check funding relevance workflow. Also add a guard for too sparse metadata, bypassing the LLM calls if the title + description are below 30 characters, and a test for this guard. --- app/activities/check_funding_relevance.py | 22 ++++++++-- .../check_funding_relevance_workflow.py | 13 +++--- tests/test_check_funding_relevance.py | 40 +++++++++++++++++++ 3 files changed, 67 insertions(+), 8 deletions(-) create mode 100644 tests/test_check_funding_relevance.py diff --git a/app/activities/check_funding_relevance.py b/app/activities/check_funding_relevance.py index 1224ed5..2618fe3 100644 --- a/app/activities/check_funding_relevance.py +++ b/app/activities/check_funding_relevance.py @@ -8,6 +8,8 @@ from app.activities._llm import build_agent from app.config import get_settings +from app.observability import propagate_langfuse_context +from app.workflows.specs import WorkflowContext class CheckFundingRelevanceRequest(BaseModel): @@ -25,18 +27,32 @@ class CheckFundingRelevanceResponse(BaseModel): message: str = Field(description="Explanation of the decision") +# Below this many non-whitespace-stripped chars, there is insufficient data +# for performing a relevance check. +MIN_METADATA_CHARS = 30 + + @activity.defn async def check_funding_relevance( request: CheckFundingRelevanceRequest, + context: WorkflowContext, ) -> CheckFundingRelevanceResponse: """Use an LLM to assess if a record's metadata matches a grant description.""" agent = build_agent(get_settings().llm, CheckFundingRelevanceResponse, request.rule) + title = str(request.metadata.get("title", "")) + description = str(request.metadata.get("description", "")) + if len(title.strip()) + len(description.strip()) < MIN_METADATA_CHARS: + return CheckFundingRelevanceResponse( + match=False, + message="Insufficient metadata to check funding relevance.", + ) prompt = ( f"Grant description:\n{request.award_description}\n\n" - f"Record title:\n{request.metadata.get('title')}\n\n" - f"Record description:\n{request.metadata.get('description')}" + f"Record title:\n{title}\n\n" + f"Record description:\n{description}" ) - result = await agent.run(prompt) + with propagate_langfuse_context(context, trace_name="check_funding_relevance"): + result = await agent.run(prompt) return result.output diff --git a/app/workflows/check_funding_relevance_workflow.py b/app/workflows/check_funding_relevance_workflow.py index 7cd6c7d..978c28e 100644 --- a/app/workflows/check_funding_relevance_workflow.py +++ b/app/workflows/check_funding_relevance_workflow.py @@ -52,11 +52,14 @@ async def run( result = await workflow.execute_activity( check_funding_relevance, - CheckFundingRelevanceRequest( - metadata=params.metadata, - award_description=params.award_description, - rule=params.rule, - ), + args=[ + CheckFundingRelevanceRequest( + metadata=params.metadata, + award_description=params.award_description, + rule=params.rule, + ), + context, + ], start_to_close_timeout=timedelta(minutes=2), ) except Exception: diff --git a/tests/test_check_funding_relevance.py b/tests/test_check_funding_relevance.py new file mode 100644 index 0000000..e248ea7 --- /dev/null +++ b/tests/test_check_funding_relevance.py @@ -0,0 +1,40 @@ +# SPDX-FileCopyrightText: 2026 CERN. +# SPDX-License-Identifier: MIT + +"""Tests for the check_funding_relevance activity's deterministic guard. + +Covers the insufficient-metadata gate (no LLM call). +""" + +import asyncio + +import pytest + +from app.activities.check_funding_relevance import ( + CheckFundingRelevanceRequest, + check_funding_relevance, +) +from app.workflows.specs import WorkflowContext + +CONTEXT = WorkflowContext(workflow_id="wf-test", tenant_id="tenant-1") +AWARD = "This project investigates LLM-based metadata extraction." +RULE = "Return match=true if the record is about the same topic as the grant." + + +@pytest.mark.parametrize( + ("title", "description"), + [ + ("", ""), + ("short title", "description"), + ], +) +def test_insufficient_metadata_returns_no_match(title, description): + """Below the char threshold the activity returns the fixed response, no LLM call.""" + request = CheckFundingRelevanceRequest( + award_description=AWARD, + metadata={"title": title, "description": description}, + rule=RULE, + ) + result = asyncio.run(check_funding_relevance(request, CONTEXT)) + assert result.match is False + assert result.message == "Insufficient metadata to check funding relevance."