feat: 5-Agent Agentic RAG 리포트 생성 (Strands Graph)

.env.example

@@ -27,6 +27,7 @@ SQS_AUDIO_PREPROCESS_QUEUE_URL=
 SQS_GPU_INFERENCE_QUEUE_URL=
 SQS_REPORT_ANALYSIS_QUEUE_URL=
 SQS_RAG_INGEST_QUEUE_URL=
+SQS_REPORT_GENERATION_QUEUE_URL=
 
 BEDROCK_REGION=ap-northeast-2
 BEDROCK_REPORT_MODEL_ID=anthropic.claude-haiku-4-5-20251001-v1:0

app/agents/report_editing_chatbot/validators.py

@@ -9,6 +9,11 @@
     r"(장애로\s*(판단|확인|진단))",
     r"(진단\s*결과\s*[^\(]*장애)",
     r"(\[진단\]|\[확진\])",
+    # 장애/진단명 자체를 "의심"과 묶어 쓰면 진단 시사로 읽혀 금지 대상이다
+    # (01_production_5_agent_agentic_rag_design.md §8.2 대표 예시: "자폐가 의심됩니다").
+    # "발달 지연이 의심됩니다" 같은 일반 관찰 문구는 임상적으로 허용되는 완곡
+    # 표현이라 여기서 걸러지지 않도록, 진단명/장애명이 명시된 경우만 잡는다.
+    r"(자폐|ADHD|ADD|지적장애|조음장애|언어장애|발달장애|주의력결핍|과잉행동장애|증후군)\s*(이|가)?\s*(으로)?\s*의심",
 ]
 
 # 근거 없는 단정 개선/악화 표현

app/agents/report_generation/clinical_context/agent.py

@@ -0,0 +1,81 @@
+# Clinical Context Agent. tool 없음 — 모든 입력은 LOAD_CONTEXT가 준비한다 (§8.2).
+# 참고: 01_production_5_agent_agentic_rag_design.md §8
+from __future__ import annotations
+
+from loguru import logger
+from strands import Agent
+from strands.models import BedrockModel
+
+from app.agents.report_generation.clinical_context.prompt import CLINICAL_CONTEXT_SYSTEM_PROMPT
+from app.agents.report_generation.clinical_context.schemas import ClinicalContextOutput
+from app.agents.report_generation.config import report_generation_settings as cfg
+from app.agents.report_generation.metric_trends import compute_trend_direction
+from app.agents.report_generation.models.common import CurrentSession, PatientHistory
+
+# 반응 지연은 낮을수록 좋은 지표라 추세 방향을 뒤집는다.
+_LOWER_IS_BETTER_METRICS = {"response_latency_avg_ms"}
+
+
+def build_clinical_context_agent() -> Agent:
+    model = BedrockModel(
+        model_id=cfg.context_agent_model_id,
+        region_name="ap-northeast-2",
+        temperature=cfg.context_agent_temperature,
+        max_tokens=4096,
+    )
+    return Agent(model=model, system_prompt=CLINICAL_CONTEXT_SYSTEM_PROMPT)
+
+
+def _compute_trend_facts(
+    current_session: CurrentSession, patient_history: PatientHistory
+) -> dict[str, str]:
+    """지표별 trend_direction을 결정론적으로 미리 계산해 Agent에게 사실로 준다."""
+    facts: dict[str, str] = {}
+    current_metrics = current_session.metrics.model_dump()
+    for metric, values in patient_history.metric_trends.items():
+        current_value = current_metrics.get(metric)
+        series = [*values, current_value] if current_value is not None else values
+        facts[metric] = compute_trend_direction(
+            [float(v) for v in series if v is not None],
+            higher_is_better=metric not in _LOWER_IS_BETTER_METRICS,
+        )
+    return facts
+
+
+def analyze_clinical_context(
+    current_session: CurrentSession,
+    patient_history: PatientHistory,
+    max_retries: int = 1,
+) -> ClinicalContextOutput:
+    """실패 시 한 번 재시도하고, 그래도 실패하면 예외를 낸다 (§8.6).
+
+    호출자(Phase 4 Graph 노드)가 이 예외를 FAILED_CONTEXT_ANALYSIS로 변환한다.
+    """
+    trend_facts = _compute_trend_facts(current_session, patient_history)
+
+    message = (
+        f"[선택 분석 항목]\n{current_session.selected_analysis}\n\n"
+        f"[현재 세션 지표]\n{current_session.metrics.model_dump_json(indent=2)}\n\n"
+        f"[환자 히스토리]\n{patient_history.model_dump_json(indent=2)}\n\n"
+        f"[지표 추세 - 이미 계산됨, 그대로 사용]\n{trend_facts}\n\n"
+        f"[세션 조건]\n{current_session.session_conditions}\n\n"
+        "위 정보를 바탕으로 핵심 clinical finding을 구조화하라."
+    )
+
+    agent = build_clinical_context_agent()
+    last_error: Exception | None = None
+    for attempt in range(max_retries + 1):
+        try:
+            result = agent(
+                message,
+                structured_output_model=ClinicalContextOutput,
+                limits={"turns": cfg.context_agent_max_turns},
+            )
+            if result.structured_output is not None:
+                return result.structured_output
+            last_error = ValueError("structured_output이 비어 있음")
+        except Exception as exc:  # noqa: BLE001 - 재시도 후 상위에서 FailureCode로 변환
+            last_error = exc
+        logger.warning(f"[clinical_context_agent] 시도 {attempt + 1} 실패: {last_error}")
+
+    raise last_error or RuntimeError("clinical context analysis failed")

app/agents/report_generation/clinical_context/prompt.py

@@ -0,0 +1,21 @@
+CLINICAL_CONTEXT_SYSTEM_PROMPT = """너는 언어재활 리포트 생성을 위한 임상 맥락 분석 AI다.
+
+역할:
+- 현재 세션 지표와 환자 longitudinal context를 분석해 리포트가 다뤄야 할 핵심
+  clinical finding을 구조화한다.
+- 이 단계에서 SOAP 문장을 작성하지 않는다.
+
+절대 지켜야 할 규칙:
+- 진단하지 않는다.
+- 현재 지표에 없는 내용을 추정하지 않는다.
+- 각 지표의 trend_direction은 [지표 추세] 블록에 이미 계산되어 주어진 값을 그대로
+  사용한다 — 직접 계산하거나 다르게 판단하지 않는다.
+- 한 회기의 현상과 반복 패턴을 구분한다.
+- INSUFFICIENT_DATA를 정상적인 결과로 허용한다 (오류가 아니다).
+- 문헌 지식을 직접 만들어내지 않는다 (그건 Evidence Research/Critic Agent의 역할이다).
+- selected_analysis에 없는 지표를 핵심 finding으로 확정하지 않는다.
+
+출력은 ClinicalContextOutput 스키마를 따른다: primary_findings, secondary_findings,
+contextual_conditions, longitudinal_patterns, clinical_priorities, evidence_needs,
+insufficient_data_flags.
+"""

app/agents/report_generation/clinical_context/schemas.py

@@ -0,0 +1,15 @@
+# Clinical Context Agent의 structured output 계약.
+# 참고: 01_production_5_agent_agentic_rag_design.md §8.4
+from pydantic import BaseModel
+
+from app.agents.report_generation.models.report import MetricFinding
+
+
+class ClinicalContextOutput(BaseModel):
+    primary_findings: list[MetricFinding]
+    secondary_findings: list[MetricFinding] = []
+    contextual_conditions: list[str] = []
+    longitudinal_patterns: list[str] = []
+    clinical_priorities: list[str] = []
+    evidence_needs: list[str] = []
+    insufficient_data_flags: list[str] = []

app/agents/report_generation/clinical_report/agent.py

@@ -0,0 +1,61 @@
+# Clinical Report Agent. 원칙적으로 tool 없음 — 검색/DB조회/저장 tool을 주지 않는다 (§11.2).
+# 참고: 01_production_5_agent_agentic_rag_design.md §11
+from __future__ import annotations
+
+from loguru import logger
+from strands import Agent
+from strands.models import BedrockModel
+
+from app.agents.report_generation.clinical_context.schemas import ClinicalContextOutput
+from app.agents.report_generation.clinical_report.prompt import CLINICAL_REPORT_SYSTEM_PROMPT
+from app.agents.report_generation.clinical_report.schemas import ClinicalReportOutput
+from app.agents.report_generation.config import report_generation_settings as cfg
+from app.agents.report_generation.evidence_critic.schemas import EvidenceCriticOutput
+from app.agents.report_generation.models.common import CurrentSession, PatientHistory
+
+
+def build_clinical_report_agent() -> Agent:
+    model = BedrockModel(
+        model_id=cfg.report_agent_model_id,
+        region_name="ap-northeast-2",
+        temperature=cfg.report_agent_temperature,
+        max_tokens=8192,
+    )
+    return Agent(model=model, system_prompt=CLINICAL_REPORT_SYSTEM_PROMPT)
+
+
+def write_report(
+    clinical_context: ClinicalContextOutput,
+    current_session: CurrentSession,
+    patient_history: PatientHistory,
+    critic_result: EvidenceCriticOutput,
+    rewrite_feedback: str | None = None,
+) -> ClinicalReportOutput:
+    accepted_ids = {e.evidence_id for e in critic_result.accepted_evidence}
+
+    message = (
+        f"[Clinical Context]\n{clinical_context.model_dump_json(indent=2)}\n\n"
+        f"[현재 세션]\n{current_session.model_dump_json(indent=2)}\n\n"
+        f"[환자 히스토리]\n{patient_history.model_dump_json(indent=2)}\n\n"
+        f"[승인된 문헌 근거 evidence_id 목록 - 이것만 인용 가능]\n{sorted(accepted_ids)}\n\n"
+        f"[승인된 근거 상세]\n"
+        + "\n".join(
+            f"- {e.evidence_id}: {e.supported_claim_scope} (directness={e.directness})"
+            for e in critic_result.accepted_evidence
+        )
+    )
+    if rewrite_feedback:
+        message += f"\n\n[이전 검증/QA 피드백 - 반드시 반영해 재작성]\n{rewrite_feedback}"
+    message += "\n\n위 정보로 SOAP 리포트 초안을 작성하라."
+
+    agent = build_clinical_report_agent()
+    result = agent(
+        message,
+        structured_output_model=ClinicalReportOutput,
+        limits={"turns": cfg.report_agent_max_turns},
+    )
+    output = result.structured_output
+    if output is None:
+        logger.warning("[clinical_report_agent] structured_output이 비어 있음")
+        raise ValueError("clinical report generation returned no structured output")
+    return output

app/agents/report_generation/clinical_report/prompt.py

@@ -0,0 +1,35 @@
+CLINICAL_REPORT_SYSTEM_PROMPT = """너는 언어재활 SOAP 리포트 초안을 작성하는 AI다.
+tool은 없다 — 필요한 입력은 모두 프롬프트로 주어진다.
+
+두 단계 출력을 한 번에 생성한다: reasoning_plan(SOAP 섹션별 근거 연결 계획) →
+soap(실제 문장). 두 출력은 스키마에서 명확히 분리되지만 서로 일치해야 한다.
+
+SOAP 섹션 규칙:
+
+[S] 허용: 보호자/치료사 진술, 세션에서 직접 전달된 주관 정보, 관찰 조건.
+    금지: 측정 수치 해석, 문헌 지식, 진단 추론.
+[O] 허용: 관찰 사실, transcript 근거, 측정 지표, 선택 분석 결과.
+    금지: 치료 효과 평가, 진단, 논문 기반 일반화.
+[A] 허용: 현재 지표 해석, 최근 환자 히스토리 변화, 승인된 문헌 근거와의 연결,
+    제한점을 포함한 임상적 해석.
+    금지: 진단 확정, 데이터에 없는 개선/악화, 문헌보다 강한 표현,
+    상관관계를 인과관계로 변경.
+[P] 허용: 다음 회기 목표, 치료사가 검토할 중재 후보, 승인된 근거가 있는 지원 방향,
+    측정 가능한 follow-up 항목.
+    금지: 치료 효과 보장, 처방처럼 단정, 다른 전문 영역의 확정 판단,
+    환자/보호자 책임 귀속.
+
+claim trace 규칙:
+- A/P의 모든 핵심 문장은 claims 목록에 ReportClaim으로 포함되어야 한다.
+- 환자 변화에 관한 문장은 patient_evidence_refs가 필수다
+  (형식: "metric_trend:{METRIC}:recent_N" 또는 "finding:{finding_id}").
+- 문헌 기반 해석/중재 문장은 literature_evidence_refs가 필수다
+  (accepted_evidence의 evidence_id만 사용 — 승인되지 않은 근거를 인용하지 않는다).
+- 둘 다 사용하는 문장은 두 ref 모두 채운다.
+- 근거 ref 없는 A/P 핵심 claim은 QA에서 실패 처리된다는 것을 알고 작성하라.
+
+이전 QA/deterministic validation 피드백이 주어지면 반드시 반영해 재작성한다.
+
+출력은 ClinicalReportOutput 스키마를 따른다: reasoning_plan, soap, claims,
+limitations, clinician_review_points.
+"""

app/agents/report_generation/clinical_report/schemas.py

@@ -0,0 +1,14 @@
+# Clinical Report Agent의 structured output 계약.
+# reasoning_plan과 soap를 한 번에 생성하되 schema에서 명확히 분리한다.
+# 참고: 01_production_5_agent_agentic_rag_design.md §11.4
+from pydantic import BaseModel
+
+from app.agents.report_generation.models.report import ReportClaim, SoapDraft, SoapReasoningPlan
+
+
+class ClinicalReportOutput(BaseModel):
+    reasoning_plan: SoapReasoningPlan
+    soap: SoapDraft
+    claims: list[ReportClaim]
+    limitations: list[str] = []
+    clinician_review_points: list[str] = []

app/agents/report_generation/config.py

@@ -0,0 +1,57 @@
+# Agent별 모델/temperature/retry-budget/QA threshold 설정.
+# 참고: 01_production_5_agent_agentic_rag_design.md §15, §27
+from pydantic_settings import BaseSettings
+
+
+class ReportGenerationSettings(BaseSettings):
+    graph_version: str = "v1"
+    prompt_version: str = "v1"
+    schema_version: str = "v1"
+
+    # 프로젝트 기본 리포트 생성 모델(app.config.settings.bedrock_report_model_id)과
+    # 동일한 값을 기본값으로 쓰되, Agent별로 독립 override 가능하게 한다.
+    context_agent_model_id: str = "global.anthropic.claude-haiku-4-5-20251001-v1:0"
+    research_agent_model_id: str = "global.anthropic.claude-haiku-4-5-20251001-v1:0"
+    critic_agent_model_id: str = "global.anthropic.claude-haiku-4-5-20251001-v1:0"
+    report_agent_model_id: str = "global.anthropic.claude-haiku-4-5-20251001-v1:0"
+    qa_agent_model_id: str = "global.anthropic.claude-haiku-4-5-20251001-v1:0"
+
+    context_agent_temperature: float = 0.1
+    research_agent_temperature: float = 0.2
+    critic_agent_temperature: float = 0.0
+    report_agent_temperature: float = 0.2
+    qa_agent_temperature: float = 0.0
+
+    max_research_rounds: int = 3
+    max_report_revisions: int = 2
+    max_qa_routes: int = 3
+    max_total_model_calls: int = 20
+
+    # Agent 호출 1회당 최대 turn 수 (strands Agent의 limits={"turns": N}).
+    # 이 상한이 없으면 structured output이 스키마 검증에 계속 실패할 때 Strands가
+    # 내부적으로 무한(또는 매우 큰) 재시도를 해 비용이 통제 없이 쌓일 수 있다
+    # (로컬 데모에서 EvidenceCriticOutput 생성이 148회 연속 실패한 채 안 끝난 사례로 확인).
+    # tool을 쓰는 Agent(Research/Critic)는 정상 작업만으로도 turn 수가 많이 나오므로
+    # tool이 없는 Agent보다 넉넉하게 잡는다.
+    context_agent_max_turns: int = 6
+    research_agent_max_turns: int = 40
+    # 근거 후보가 많을 때 fetch_document_section/verify_source_metadata 호출이
+    # 늘어나 30으로는 부족한 사례를 실전에서 확인해 50으로 상향했다.
+    critic_agent_max_turns: int = 50
+    report_agent_max_turns: int = 6
+    qa_agent_max_turns: int = 6
+
+    qa_groundedness_threshold: float = 0.90
+    qa_citation_threshold: float = 0.95
+    qa_context_fidelity_threshold: float = 0.95
+    qa_longitudinal_threshold: float = 0.95
+    qa_soap_threshold: float = 0.90
+    qa_overall_threshold: float = 0.92
+
+    enable_bedrock_guardrails: bool = False
+    bedrock_guardrail_id: str = ""
+
+    model_config = {"env_file": ".env", "extra": "ignore"}
+
+
+report_generation_settings = ReportGenerationSettings()

app/agents/report_generation/context_loader.py

@@ -0,0 +1,35 @@
+# LOAD_CONTEXT deterministic node의 핵심 로직.
+# Phase 4에서 Strands Graph 노드로 얇게 감싸질 함수 — 여기서는 Graph에 의존하지 않는다.
+# 참고: 01_production_5_agent_agentic_rag_design.md §7.1
+#
+# 중복 실행 방지(§7.1 항목 8)는 여기서 하지 않는다 — Phase 4의
+# run_report_generation()이 report_generation_runs row를 생성하기 *전에* 확인한다.
+# 이 함수 안에서 확인하면 그 시점엔 이미 자기 자신의 run이 QUEUED로 존재해
+# 항상 자기 자신을 중복으로 오판한다.
+from app.agents.report_generation.models.common import CurrentSession, PatientHistory
+from app.repositories.patient_history_repository import PatientHistoryRepository
+from app.repositories.session_repository import SessionRepository
+
+
+async def load_context(
+    *,
+    session_id: str,
+    patient_ref_id: str,
+    therapist_id: str,
+    selected_analysis: list[str],
+    history_session_count: int,
+    session_repo: SessionRepository,
+    patient_history_repo: PatientHistoryRepository,
+) -> tuple[CurrentSession, PatientHistory]:
+    """§7.1의 9개 책임 중 권한 검증/확정 데이터 조회를 수행한다.
+
+    민감정보 최소화(§7.1 항목 9)는 CurrentSession/PatientHistory 스키마 자체가
+    필요한 필드만 노출하도록 설계되어 있어 이 함수가 별도로 마스킹하지 않는다.
+    """
+    current_session = await session_repo.get_current_session(
+        session_id, therapist_id, selected_analysis
+    )
+    patient_history = await patient_history_repo.get_patient_history(
+        patient_ref_id, therapist_id, history_session_count
+    )
+    return current_session, patient_history