From 6436db475d65262d1bcc125e9bdd8456f46579c4 Mon Sep 17 00:00:00 2001 From: Deep Koluguri Date: Fri, 22 May 2026 20:27:25 -0400 Subject: [PATCH] feat: Add CuraFlow AI Scribe and Coder agents --- k8s/curaflow-agent/deployment.yaml | 43 +++++++++++++++++++++ k8s/kustomization.yaml | 3 +- src/curaflow_agent/activities.py | 60 ++++++++++++++++++++++++++++++ src/curaflow_agent/models.py | 20 ++++++++++ src/curaflow_agent/worker.py | 41 ++++++++++++++++++++ src/curaflow_agent/workflows.py | 39 +++++++++++++++++++ 6 files changed, 205 insertions(+), 1 deletion(-) create mode 100644 k8s/curaflow-agent/deployment.yaml create mode 100644 src/curaflow_agent/activities.py create mode 100644 src/curaflow_agent/models.py create mode 100644 src/curaflow_agent/worker.py create mode 100644 src/curaflow_agent/workflows.py diff --git a/k8s/curaflow-agent/deployment.yaml b/k8s/curaflow-agent/deployment.yaml new file mode 100644 index 0000000..faf34e3 --- /dev/null +++ b/k8s/curaflow-agent/deployment.yaml @@ -0,0 +1,43 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: curaflow-agent + labels: + app.kubernetes.io/name: curaflow-agent +spec: + replicas: 1 + selector: + matchLabels: + app.kubernetes.io/name: curaflow-agent + template: + metadata: + labels: + app.kubernetes.io/name: curaflow-agent + spec: + containers: + - name: worker + image: 192.168.8.250:5000/agents-runtime:latest + imagePullPolicy: Always + command: ["python", "-m", "curaflow_agent.worker"] + env: + - name: PYTHONPATH + value: "/app/src" + - name: PYTHONUNBUFFERED + value: "1" + - name: TEMPORAL_URL + value: "temporal-frontend.ai-core.svc.cluster.local:7233" + - name: TEMPORAL_NAMESPACE + value: "default" + - name: CURAFLOW_TASK_QUEUE + value: "curaflow-tasks" + - name: OLLAMA_BASE_URL + value: "http://ollama.ai-core.svc.cluster.local:11434/v1" + - name: LLM_MODEL + value: "qwen2.5:3b" + resources: + limits: + cpu: 500m + memory: 512Mi + requests: + cpu: 100m + memory: 128Mi diff --git a/k8s/kustomization.yaml b/k8s/kustomization.yaml index b83b23e..77a018c 100644 --- a/k8s/kustomization.yaml +++ b/k8s/kustomization.yaml @@ -1,3 +1,4 @@ apiVersion: kustomize.config.k8s.io/v1beta1 kind: Kustomization -resources: [] +resources: + - curaflow-agent/ diff --git a/src/curaflow_agent/activities.py b/src/curaflow_agent/activities.py new file mode 100644 index 0000000..7ef4648 --- /dev/null +++ b/src/curaflow_agent/activities.py @@ -0,0 +1,60 @@ +import os +import json +from temporalio import activity +from langchain_openai import ChatOpenAI +from langchain_core.prompts import PromptTemplate +from .models import ClinicalNote, BillingCodesOutput + +# In the cluster, Ollama provides an OpenAI-compatible endpoint. +# We fetch model/URL from env vars so it's easy to override for testing. +OLLAMA_BASE_URL = os.environ.get("OLLAMA_BASE_URL", "http://ollama.ai-core.svc.cluster.local:11434/v1") +LLM_MODEL = os.environ.get("LLM_MODEL", "qwen2.5:3b") + +def get_llm(): + return ChatOpenAI( + model=LLM_MODEL, + api_key="ollama", # Ollama doesn't strictly need an API key + base_url=OLLAMA_BASE_URL, + temperature=0.1 + ) + +@activity.defn +async def structure_note_activity(raw_dictation: str) -> dict: + """ + Takes raw doctor's dictation and uses an LLM to extract a structured ClinicalNote payload. + """ + llm = get_llm() + structured_llm = llm.with_structured_output(ClinicalNote) + + prompt = PromptTemplate.from_template( + "You are an expert AI Medical Scribe. Extract the clinical information from the following raw dictation and structure it perfectly.\n\n" + "Dictation:\n{dictation}\n" + ) + + chain = prompt | structured_llm + result: ClinicalNote = await chain.ainvoke({"dictation": raw_dictation}) + + # Return as dict so Temporal can serialize it natively + return result.model_dump() + +@activity.defn +async def generate_billing_codes_activity(clinical_note_dict: dict) -> dict: + """ + Takes a structured ClinicalNote and uses an LLM to suggest ICD-10 and CPT codes. + """ + llm = get_llm() + structured_llm = llm.with_structured_output(BillingCodesOutput) + + # Convert dict to nicely formatted string for the prompt + note_str = json.dumps(clinical_note_dict, indent=2) + + prompt = PromptTemplate.from_template( + "You are an expert Medical Coder. Review the following structured clinical note and identify all applicable ICD-10 diagnosis codes and CPT procedure codes.\n" + "Provide a clear justification for each code based solely on the provided clinical note.\n\n" + "Clinical Note:\n{note}\n" + ) + + chain = prompt | structured_llm + result: BillingCodesOutput = await chain.ainvoke({"note": note_str}) + + return result.model_dump() diff --git a/src/curaflow_agent/models.py b/src/curaflow_agent/models.py new file mode 100644 index 0000000..f7893a5 --- /dev/null +++ b/src/curaflow_agent/models.py @@ -0,0 +1,20 @@ +from pydantic import BaseModel, Field +from typing import List, Optional + +class ClinicalNote(BaseModel): + chief_complaint: str = Field(description="The primary reason for the patient's visit.") + history_of_present_illness: str = Field(description="Detailed narrative of the patient's current symptoms and history.") + past_medical_history: List[str] = Field(description="List of known past medical conditions.") + medications: List[str] = Field(description="List of current medications the patient is taking.") + allergies: List[str] = Field(description="List of known allergies.") + assessment: str = Field(description="The physician's diagnosis or assessment of the patient's condition.") + plan: str = Field(description="The proposed treatment plan or next steps.") + +class BillingCode(BaseModel): + code_type: str = Field(description="Either 'ICD-10' or 'CPT'") + code: str = Field(description="The specific alphanumeric code (e.g., J45.909, 99213)") + description: str = Field(description="Brief description of what the code represents") + justification: str = Field(description="Explanation of why this code was selected based on the clinical note") + +class BillingCodesOutput(BaseModel): + codes: List[BillingCode] = Field(description="List of all applicable billing codes for the visit.") diff --git a/src/curaflow_agent/worker.py b/src/curaflow_agent/worker.py new file mode 100644 index 0000000..af2eb80 --- /dev/null +++ b/src/curaflow_agent/worker.py @@ -0,0 +1,41 @@ +import asyncio +import os +import sys +import logging + +from temporalio.client import Client +from temporalio.worker import Worker + +# Configure basic logging so we can see output in Kubernetes logs +logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") +logger = logging.getLogger(__name__) + +from curaflow_agent.workflows import ClinicalIntakeWorkflow +from curaflow_agent.activities import structure_note_activity, generate_billing_codes_activity + +async def main(): + temporal_url = os.environ.get("TEMPORAL_URL", "temporal-frontend.ai-core.svc.cluster.local:7233") + temporal_namespace = os.environ.get("TEMPORAL_NAMESPACE", "default") + task_queue = os.environ.get("CURAFLOW_TASK_QUEUE", "curaflow-tasks") + + logger.info(f"Connecting to Temporal cluster at {temporal_url} (Namespace: {temporal_namespace})") + + try: + client = await Client.connect(temporal_url, namespace=temporal_namespace) + logger.info("Successfully connected to Temporal!") + except Exception as e: + logger.error(f"Failed to connect to Temporal: {e}") + sys.exit(1) + + worker = Worker( + client, + task_queue=task_queue, + workflows=[ClinicalIntakeWorkflow], + activities=[structure_note_activity, generate_billing_codes_activity], + ) + + logger.info(f"Starting CuraFlow agent worker on queue '{task_queue}'...") + await worker.run() + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/curaflow_agent/workflows.py b/src/curaflow_agent/workflows.py new file mode 100644 index 0000000..18e52ab --- /dev/null +++ b/src/curaflow_agent/workflows.py @@ -0,0 +1,39 @@ +from datetime import timedelta +from temporalio import workflow + +with workflow.unsafe.imports_passed_through(): + from .activities import structure_note_activity, generate_billing_codes_activity + +@workflow.defn +class ClinicalIntakeWorkflow: + @workflow.run + async def run(self, raw_dictation: str) -> dict: + """ + Takes raw doctor's dictation, extracts structured clinical data, + and generates associated billing codes. + Returns the combined payload. + """ + # Step 1: AI Scribe (Structure the dictation) + structured_note = await workflow.execute_activity( + structure_note_activity, + raw_dictation, + start_to_close_timeout=timedelta(minutes=10) + ) + + # Step 2: Medical Coder (Generate ICD-10/CPT codes) + billing_codes = await workflow.execute_activity( + generate_billing_codes_activity, + structured_note, + start_to_close_timeout=timedelta(minutes=10) + ) + + # Step 3: Return the final, combined object + # Note: We are returning it directly here so it's visible in the Temporal UI. + # In a real integration, we would trigger a 3rd activity here to POST this + # JSON payload to the client's BYOD (Bring Your Own Database) API endpoint. + + return { + "status": "success", + "clinical_note": structured_note, + "billing_codes": billing_codes + }