diagnostics

Modules:

• core –

Classes:

• DiagnosticsSkill –

Functions:

• differential –
• exams –

DiagnosticsSkill

DiagnosticsSkill()

Bases: BaseSkill

Methods:

• check_requirements –
• execute –
• post –
• pre –

Source code in src/hiperhealth/skills/diagnostics/core.py

def __init__(self) -> None:
    """
    title: Initialize the built-in diagnostics skill metadata.
    """
    super().__init__(
        SkillMetadata(
            name='hiperhealth.diagnostics',
            version='0.4.0',
            stages=(Stage.DIAGNOSIS, Stage.EXAM),
            description=(
                'Core differential diagnosis and exam suggestion.'
            ),
        )
    )

check_requirements

check_requirements(
    stage: str, ctx: PipelineContext
) -> list[Inquiry]

Sends the current patient data to the LLM and asks what additional information would improve the given stage. Fields already present in ctx.patient are filtered out. parameters: stage: type: str ctx: type: PipelineContext returns: type: list[Inquiry]

Source code in src/hiperhealth/skills/diagnostics/core.py

def check_requirements(
    self, stage: str, ctx: PipelineContext
) -> list[Inquiry]:
    """
    title: Use the LLM to identify missing clinical information.
    summary: |-
      Sends the current patient data to the LLM and asks what
      additional information would improve the given stage.
      Fields already present in ctx.patient are filtered out.
    parameters:
      stage:
        type: str
      ctx:
        type: PipelineContext
    returns:
      type: list[Inquiry]
    """
    run_kwargs = ctx.extras.get('_run_kwargs', {})
    llm = run_kwargs.get('llm')
    llm_settings = run_kwargs.get('llm_settings')

    system_prompt = _requirements_prompt(stage, ctx.language)

    extra = ctx.extras.get('prompt_fragments', {}).get(
        f'{stage}_requirements', ''
    )
    if extra:
        system_prompt = f'{system_prompt}\n\n{extra}'

    result = chat_structured(
        system_prompt,
        json.dumps(ctx.patient, ensure_ascii=False),
        LLMInquiryList,
        session_id=ctx.session_id,
        llm=llm,
        llm_settings=llm_settings,
    )

    existing_fields = set(ctx.patient.keys())
    return [
        Inquiry(
            skill_name=self.metadata.name,
            stage=stage,
            field=item.field,
            label=item.label,
            description=item.description,
            priority=item.priority,
            input_type=item.input_type,
            choices=item.choices,
        )
        for item in result.inquiries
        if item.field not in existing_fields
    ]

execute

execute(
    stage: str, ctx: PipelineContext
) -> PipelineContext

Source code in src/hiperhealth/skills/diagnostics/core.py

def execute(self, stage: str, ctx: PipelineContext) -> PipelineContext:
    """
    title: Run differential diagnosis or exam suggestions.
    parameters:
      stage:
        type: str
      ctx:
        type: PipelineContext
    returns:
      type: PipelineContext
    """
    run_kwargs = ctx.extras.get('_run_kwargs', {})
    llm = run_kwargs.get('llm')
    llm_settings = run_kwargs.get('llm_settings')

    if stage == Stage.DIAGNOSIS:
        prompt = _diagnosis_prompt(ctx.language)
        extra = ctx.extras.get('prompt_fragments', {}).get('diagnosis', '')
        if extra:
            prompt = f'{prompt}\n\n{extra}'

        result = chat(
            prompt,
            json.dumps(ctx.patient, ensure_ascii=False),
            session_id=ctx.session_id,
            llm=llm,
            llm_settings=llm_settings,
        )
        ctx.results[Stage.DIAGNOSIS] = result

    elif stage == Stage.EXAM:
        diagnosis = ctx.results.get(Stage.DIAGNOSIS)
        if not diagnosis:
            return ctx
        options = (
            diagnosis.options
            if hasattr(diagnosis, 'options')
            else diagnosis.get('options', [])
        )
        selected = (
            options
            if isinstance(options, list)
            else list(options.keys())
            if isinstance(options, dict)
            else []
        )

        prompt = _exam_prompt(ctx.language)
        extra = ctx.extras.get('prompt_fragments', {}).get('exam', '')
        if extra:
            prompt = f'{prompt}\n\n{extra}'

        result = chat(
            prompt,
            json.dumps(selected, ensure_ascii=False),
            session_id=ctx.session_id,
            llm=llm,
            llm_settings=llm_settings,
        )
        ctx.results[Stage.EXAM] = result

    return ctx

post

post(stage: str, ctx: PipelineContext) -> PipelineContext

Source code in src/hiperhealth/pipeline/skill.py

def post(self, stage: str, ctx: PipelineContext) -> PipelineContext:
    """
    title: Called after the stage's main execution.
    parameters:
      stage:
        type: str
      ctx:
        type: PipelineContext
    returns:
      type: PipelineContext
    """
    return ctx

pre

pre(stage: str, ctx: PipelineContext) -> PipelineContext

Source code in src/hiperhealth/pipeline/skill.py

def pre(self, stage: str, ctx: PipelineContext) -> PipelineContext:
    """
    title: Called before the stage's main execution.
    parameters:
      stage:
        type: str
      ctx:
        type: PipelineContext
    returns:
      type: PipelineContext
    """
    return ctx

differential

differential(
    patient: dict[str, Any],
    language: str = 'en',
    session_id: str | None = None,
    llm: StructuredLLM | None = None,
    llm_settings: LLMSettings | None = None,
) -> LLMDiagnosis

Source code in src/hiperhealth/skills/diagnostics/core.py

def differential(
    patient: dict[str, Any],
    language: str = 'en',
    session_id: str | None = None,
    llm: StructuredLLM | None = None,
    llm_settings: LLMSettings | None = None,
) -> LLMDiagnosis:
    """
    title: Return summary + list of differential diagnoses.
    parameters:
      patient:
        type: dict[str, Any]
        description: Value for patient.
      language:
        type: str
        description: Value for language.
      session_id:
        type: str | None
        description: Value for session_id.
      llm:
        type: StructuredLLM | None
        description: Value for llm.
      llm_settings:
        type: LLMSettings | None
        description: Value for llm_settings.
    returns:
      type: LLMDiagnosis
      description: Return value.
    """
    prompt = _diagnosis_prompt(language)
    chat_kwargs: dict[str, Any] = {'session_id': session_id}
    if llm is not None:
        chat_kwargs['llm'] = llm
    if llm_settings is not None:
        chat_kwargs['llm_settings'] = llm_settings
    return chat(
        prompt,
        json.dumps(patient, ensure_ascii=False),
        **chat_kwargs,
    )

exams

exams(
    selected_dx: list[str],
    language: str = 'en',
    session_id: str | None = None,
    llm: StructuredLLM | None = None,
    llm_settings: LLMSettings | None = None,
) -> LLMDiagnosis

Source code in src/hiperhealth/skills/diagnostics/core.py

def exams(
    selected_dx: list[str],
    language: str = 'en',
    session_id: str | None = None,
    llm: StructuredLLM | None = None,
    llm_settings: LLMSettings | None = None,
) -> LLMDiagnosis:
    """
    title: Return summary + list of suggested examinations.
    parameters:
      selected_dx:
        type: list[str]
        description: Value for selected_dx.
      language:
        type: str
        description: Value for language.
      session_id:
        type: str | None
        description: Value for session_id.
      llm:
        type: StructuredLLM | None
        description: Value for llm.
      llm_settings:
        type: LLMSettings | None
        description: Value for llm_settings.
    returns:
      type: LLMDiagnosis
      description: Return value.
    """
    prompt = _exam_prompt(language)
    chat_kwargs: dict[str, Any] = {'session_id': session_id}
    if llm is not None:
        chat_kwargs['llm'] = llm
    if llm_settings is not None:
        chat_kwargs['llm_settings'] = llm_settings
    return chat(
        prompt,
        json.dumps(selected_dx, ensure_ascii=False),
        **chat_kwargs,
    )