From 47f9136807e98e3ba2ce3af67fe392d4a24ce400 Mon Sep 17 00:00:00 2001 From: KimDongAn Date: Tue, 29 Sep 2026 20:31:43 +0900 Subject: [PATCH] =?UTF-8?q?perf(safety):=20=ED=98=BC=EC=9E=AC=20=ED=8C=90?= =?UTF-8?q?=EC=A0=95=20=EA=B7=B8=EB=9E=98=ED=94=84=EB=A5=BC=20=EB=A9=94?= =?UTF-8?q?=EB=AA=A8=EB=A6=AC=20=EC=82=AC=EB=B3=B8=EC=9C=BC=EB=A1=9C,=20?= =?UTF-8?q?=ED=99=94=EB=AC=BC=EB=B3=84=20=ED=8C=90=EC=A0=95=C2=B7LLM=20?= =?UTF-8?q?=ED=98=B8=EC=B6=9C=EC=9D=84=20=EB=8F=99=EC=8B=9C=EC=97=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 운영 Neo4j(Aura)는 조회 한 번이 약 300ms 라 근거 패널 한 번이 약 20초 걸렸다(화물 6종 배). - 판정이 읽는 그래프 조회 9개는 기준 데이터라 메모리 사본(graph_snapshot, 10분마다 갱신)으로 바꿨다. 같은 답을 내는지 로컬 그래프 전 화학물질로 Cypher 와 대조한다(test_graph_snapshot). - 화물별 판정·같은 선박 비교를 화물마다 동시에 돌린다(DB 해석은 먼저 순서대로). - 혼재 서술 LLM 을 종합 문장 LLM 과 함께 기다린다(종합 문장은 규칙 결과만 본다). - 기동 시 사본을 미리 읽는다. 로컬 접안 선박 10척 옛/새 코드 연속 판정 — 등급·근거 동일. Co-Authored-By: Claude Opus 5.5 --- app/agents/orchestrator/service.py | 16 +- app/agents/safety/graph_snapshot.py | 187 +++++++++++++++++++++ app/agents/safety/service.py | 185 ++++++++++---------- app/agents/tools.py | 4 + app/main.py | 13 ++ tests/test_concurrent_judgment_20260929.py | 83 +++++++++ tests/test_graph_snapshot_20260929.py | 53 ++++++ 7 files changed, 449 insertions(+), 92 deletions(-) create mode 100644 app/agents/safety/graph_snapshot.py create mode 100644 tests/test_concurrent_judgment_20260929.py create mode 100644 tests/test_graph_snapshot_20260929.py diff --git a/app/agents/orchestrator/service.py b/app/agents/orchestrator/service.py index de0a69a..1e17b46 100644 --- a/app/agents/orchestrator/service.py +++ b/app/agents/orchestrator/service.py @@ -17,7 +17,9 @@ 추천·'적합 선석 없음')을 보여 주고 있었다. 정박지는 위치로 관찰만 한다(/dashboard/anchorages). """ +import asyncio import logging +from collections.abc import Awaitable from datetime import timedelta, timezone from neo4j import AsyncDriver @@ -145,9 +147,14 @@ async def orchestrate( """summarize=False 면 종합 문장을 템플릿으로 쓴다(안전 서술 LLM 은 그대로 부른다). 판정 잡이 쓴다 — 종합 문장은 판정 기록에 저장되지 않아, 잡에서 LLM 으로 쓰면 버려진다. + + [2026-09-29] 혼재 서술 LLM(체크리스트)은 판정 중에 기다리지 않고, 종합 문장 LLM 과 함께 기다린다. + 종합 문장은 규칙 결과(근거·확인 필요)만 보므로 서술을 기다릴 필요가 없다 — 두 호출이 차례로 약 6초였다. """ - result = await _judge(db, neo4j_driver, llm_client, request) - return await _finish(result, request, llm_client if summarize else None) + pending: list[Awaitable[None]] = [] + result = await _judge(db, neo4j_driver, llm_client, request, defer=pending) + result, *_ = await asyncio.gather(_finish(result, request, llm_client if summarize else None), *pending) + return result async def _judge( @@ -155,8 +162,10 @@ async def _judge( neo4j_driver: AsyncDriver, llm_client: LLMClient, request: OrchestratorRequest, + *, + defer: list[Awaitable[None]] | None = None, ) -> OrchestratorResult: - """판정 본체 — 종합 문장(summary)은 비워 두고 _finish 가 채운다.""" + """판정 본체 — 종합 문장(summary)은 비워 두고 _finish 가 채운다. defer 는 assess_safety 참고.""" # 전역 폴백 판정(__GLOBAL_DEFAULT__ 임계값). 선석을 확인하지 못했거나 혼재로 부적합인 # 응답의 weather_assessment 를 채우는 용도다 — 통과/차단은 아래 선석별 임계값으로 한다. # [2026-09-29] 판정 대상 시설 이름을 넘긴다 — 파고 적용 규칙(wave_applies_to)이 이름으로 정해진다. @@ -270,6 +279,7 @@ async def _alternatives() -> tools.Alternatives: cargos=request.cargos, adjacent_cargos=candidate.adjacent_cargos, call_sign=request.vessel.call_sign, + defer=defer, ) opinions.append(tools.segregation_opinion(safety_result)) diff --git a/app/agents/safety/graph_snapshot.py b/app/agents/safety/graph_snapshot.py new file mode 100644 index 0000000..3102436 --- /dev/null +++ b/app/agents/safety/graph_snapshot.py @@ -0,0 +1,187 @@ +"""혼재 판정 기준 그래프의 메모리 사본 (2026-09-29). + +판정(_compute_verdict)이 Neo4j 에서 읽는 것은 화물 대 화물 **기준 데이터**뿐이다 — MSDS 혼재금지 +카테고리, IMDG Class·격리표, 46 CFR 150 호환성 그룹·예외. 로더를 다시 돌리지 않는 한 바뀌지 않는다. +그런데 판정 한 번에 이 조회가 화물 × 9번 나갔고, 운영 Neo4j(Aura)는 조회 한 번이 약 300ms 라 +화물 6종 배의 규칙 판정이 12초 넘게 걸렸다(9/29 실측). 그래서 통째로 한 번 읽어 두고, 같은 질문에 +graph_queries 의 Cypher 와 **같은 답**을 메모리에서 낸다(tests/test_graph_snapshot_20260929.py 가 대조). + +사본은 REFRESH_SECONDS 마다 다시 읽는다 — 로더를 돌리면 늦어도 그만큼 뒤에 반영된다. +""" + +import asyncio +import logging +import time +from collections import defaultdict +from dataclasses import dataclass + +from neo4j import AsyncDriver + +logger = logging.getLogger(__name__) + +REFRESH_SECONDS = 600 + +_QUERIES = { + "chemicals": "MATCH (c:Chemical) RETURN c.id AS c, c.name_ko AS name", + "avoids": """MATCH (c:Chemical)-[:INCOMPATIBLE_WITH]->(m) + RETURN c.id AS c, elementId(m) AS m, m.name AS name, 'IncompatibleMaterial' IN labels(m) AS im""", + "classified": """MATCH (c:Chemical)-[:IS_CLASSIFIED_AS]->(m) + RETURN c.id AS c, elementId(m) AS m, m.name AS name, 'IncompatibleMaterial' IN labels(m) AS im""", + "imdg": "MATCH (c:Chemical)-[:HAS_IMDG_CLASS]->(k:ImdgClass) RETURN c.id AS c, elementId(k) AS k, k.code AS code", + "segregate": """MATCH (a:ImdgClass)-[s:SEGREGATE]->(b:ImdgClass) + RETURN elementId(a) AS a, elementId(b) AS b, s.code AS code""", + "no_seg": """MATCH (a:ImdgClass)-[:NO_SEGREGATION_REQUIRED]->(b:ImdgClass) + RETURN elementId(a) AS a, elementId(b) AS b""", + "groups": """MATCH (c:Chemical)-[:IN_COMPATIBILITY_GROUP]->(g:CompatibilityGroup) + RETURN c.id AS c, elementId(g) AS g, g.group_no AS no, g.group_type AS type, g.name AS name""", + "group_incompat": """MATCH (a:CompatibilityGroup)-[:INCOMPATIBLE_WITH_GROUP]->(b:CompatibilityGroup) + RETURN elementId(a) AS a, elementId(b) AS b""", + "safe_exc": "MATCH (a:Chemical)-[:BULK_COMPAT_SAFE_EXCEPTION]->(b:Chemical) RETURN a.id AS a, b.id AS b", + "blocked_exc": "MATCH (a:Chemical)-[:BULK_COMPAT_BLOCKED_EXCEPTION]->(b:Chemical) RETURN a.id AS a, b.id AS b", +} + + +@dataclass +class GraphSnapshot: + names: dict[str, str | None] # Chemical id -> name_ko + avoids: dict[str, list[tuple[str, str, bool]]] # id -> [(카테고리 노드, 이름, IncompatibleMaterial?)] + classified: dict[str, list[tuple[str, str, bool]]] + imdg: dict[str, list[tuple[str, str]]] # id -> [(Class 노드, code)] + segregate: dict[tuple[str, str], list[str]] # (Class 노드, Class 노드) -> [격리 code] + no_seg: set[tuple[str, str]] + groups: dict[str, list[tuple[str, int, str, str]]] # id -> [(그룹 노드, 번호, 구분, 이름)] + group_incompat: dict[tuple[str, str], int] # (그룹 노드, 그룹 노드) -> 관계 수 + safe_exc: set[tuple[str, str]] + blocked_exc: set[tuple[str, str]] + live_categories: set[str] + + # ── graph_queries 의 같은 이름 함수와 같은 답 ──────────────────────────────── + + def incompatible_conflicts(self, target: str, adjacent: list[str]) -> list[dict]: + if not adjacent or target not in self.names: + return [] + rows: set[tuple] = set() + target_cls = {m for m, _, _ in self.classified.get(target, [])} + for b in adjacent: + if b not in self.names: + continue + b_cls = {m for m, _, _ in self.classified.get(b, [])} + for m, name, _ in self.avoids.get(target, []): + if m in b_cls: + rows.add((b, self.names[b], name, "target_incompatible_with_adjacent")) + for m, name, _ in self.avoids.get(b, []): + if m in target_cls: + rows.add((b, self.names[b], name, "adjacent_incompatible_with_target")) + return [dict(chem_id=r[0], name_ko=r[1], category=r[2], direction=r[3]) for r in rows] + + def imdg_segregation_conflicts(self, target: str, adjacent: list[str]) -> list[dict]: + if not adjacent: + return [] + return [ + dict(chem_id=b, name_ko=self.names[b], target_class=ca_code, adjacent_class=cb_code, segregation_code=code) + for ca, ca_code in self.imdg.get(target, []) + for b in adjacent + for cb, cb_code in self.imdg.get(b, []) + for code in self.segregate.get((ca, cb), []) + ] + + def imdg_no_segregation_required(self, target: str, adjacent: list[str]) -> set[str]: + return { + b + for ca, _ in self.imdg.get(target, []) + for b in adjacent + for cb, _ in self.imdg.get(b, []) + if (ca, cb) in self.no_seg + } + + def imdg_classes(self, chem_ids: list[str]) -> dict[str, str]: + return {c: code for c in dict.fromkeys(chem_ids) for _, code in self.imdg.get(c, [])} + + def bulk_group_conflicts(self, target: str, adjacent: list[str]) -> list[dict]: + if not adjacent: + return [] + return [ + dict(chem_id=b, name_ko=self.names[b]) + for ga, *_ in self.groups.get(target, []) + for b in adjacent + for gb, *_ in self.groups.get(b, []) + for _ in range(self.group_incompat.get((ga, gb), 0)) + ] + + def bulk_groups(self, chem_ids: list[str]) -> dict[str, tuple[int, str, str]]: + return {c: (no, type_, name) for c in dict.fromkeys(chem_ids) for _, no, type_, name in self.groups.get(c, [])} + + def bulk_exceptions(self, target: str, adjacent: list[str]) -> tuple[set[str], set[str]]: + return ( + {b for b in adjacent if (target, b) in self.safe_exc}, + {b for b in adjacent if (target, b) in self.blocked_exc}, + ) + + def assessability_facts(self, chem_ids: list[str]) -> dict[str, tuple[set[str], set[str]]]: + return { + c: ( + {name for _, name, im in self.avoids.get(c, []) if im and name}, + {name for _, name, im in self.classified.get(c, []) if im and name}, + ) + for c in chem_ids if c in self.names + } + + +async def load_snapshot(driver: AsyncDriver) -> GraphSnapshot: + async def run(query: str) -> list[dict]: + async with driver.session() as session: + async def _tx(tx): + return [record.data() async for record in await tx.run(query)] + return await session.execute_read(_tx) + + keys = list(_QUERIES) + raw = dict(zip(keys, await asyncio.gather(*(run(_QUERIES[k]) for k in keys)))) + + def by_chem(rows, *cols): + out = defaultdict(list) + for r in rows: + out[r["c"]].append(tuple(r[c] for c in cols)) + return dict(out) + + segregate = defaultdict(list) + for r in raw["segregate"]: + segregate[(r["a"], r["b"])].append(r["code"]) + group_incompat = defaultdict(int) + for r in raw["group_incompat"]: + group_incompat[(r["a"], r["b"])] += 1 + classified = by_chem(raw["classified"], "m", "name", "im") + return GraphSnapshot( + names={r["c"]: r["name"] for r in raw["chemicals"]}, + avoids=by_chem(raw["avoids"], "m", "name", "im"), + classified=classified, + imdg=by_chem(raw["imdg"], "k", "code"), + segregate=dict(segregate), + no_seg={(r["a"], r["b"]) for r in raw["no_seg"]}, + groups=by_chem(raw["groups"], "g", "no", "type", "name"), + group_incompat=dict(group_incompat), + safe_exc={(r["a"], r["b"]) for r in raw["safe_exc"]}, + blocked_exc={(r["a"], r["b"]) for r in raw["blocked_exc"]}, + # find_live_categories 와 같다 — Chemical 이 IS_CLASSIFIED_AS 로 가리키는 IncompatibleMaterial 이름 + live_categories={name for rows in classified.values() for _, name, im in rows if im and name}, + ) + + +_snapshot: GraphSnapshot | None = None +_loaded_at = 0.0 +_lock: asyncio.Lock | None = None + + +async def get_snapshot(driver: AsyncDriver) -> GraphSnapshot: + """사본을 돌려준다. 없거나 REFRESH_SECONDS 가 지났으면 먼저 다시 읽는다(동시 요청은 한 번만 읽는다).""" + global _snapshot, _loaded_at, _lock + if _snapshot is not None and time.monotonic() - _loaded_at < REFRESH_SECONDS: + return _snapshot + if _lock is None: + _lock = asyncio.Lock() + async with _lock: + if _snapshot is None or time.monotonic() - _loaded_at >= REFRESH_SECONDS: + started = time.monotonic() + _snapshot = await load_snapshot(driver) + _loaded_at = time.monotonic() + logger.info("혼재 판정 그래프 사본 적재: 화학물질 %d종, %.1fs", len(_snapshot.names), _loaded_at - started) + return _snapshot diff --git a/app/agents/safety/service.py b/app/agents/safety/service.py index 011bf95..abb868c 100644 --- a/app/agents/safety/service.py +++ b/app/agents/safety/service.py @@ -17,6 +17,7 @@ import asyncio import re +from collections.abc import Awaitable from dataclasses import dataclass, field from neo4j import AsyncDriver @@ -27,17 +28,7 @@ from app.models import MsdsChemical from .bulk_compatibility import build_bulk_conflicts -from .graph_queries import ( - find_assessability_facts, - find_bulk_exceptions, - find_bulk_group_conflicts, - find_bulk_groups, - find_imdg_classes, - find_imdg_no_segregation_required, - find_imdg_segregation_conflicts, - find_incompatible_conflicts, - find_live_categories, -) +from .graph_snapshot import get_snapshot from .msds_context import ghs_classification, resolve_cargo, summarize_hazard_sections from .prompt import SYSTEM_PROMPT, build_user_prompt from .stabilized_cargo import with_stabilized_checklist @@ -208,15 +199,24 @@ def add(a: str, b: str, basis: str, detail: str) -> None: basis=basis, detail=detail, )) - for v in verdicts: + # 화물 행은 이미 해석돼 있다 — DB 없이 화물마다 동시에 돌린다(_compute_verdicts 와 같은 이유, 2026-09-29). + def onboard_verdict(v: "_Verdict"): a = v.target_row.chem_id - others = [ - AdjacentCargo(berth_name=_ONBOARD_LABEL, cargo=CargoRef(chem_id=o.target_row.chem_id)) - for o in verdicts if o.target_row.chem_id != a - ] - ov = await _compute_verdict( - db, neo4j_driver, SafetyAssessmentRequest(target_cargo=CargoRef(chem_id=a), adjacent_cargos=others), + others = [o for o in verdicts if o.target_row.chem_id != a] + request = SafetyAssessmentRequest( + target_cargo=CargoRef(chem_id=a), + adjacent_cargos=[ + AdjacentCargo(berth_name=_ONBOARD_LABEL, cargo=CargoRef(chem_id=o.target_row.chem_id)) + for o in others + ], ) + return _compute_verdict( + db, neo4j_driver, request, + resolved=(v.target_row, [(_ONBOARD_LABEL, None, o.target_row) for o in others]), + ) + + for v, ov in zip(verdicts, await asyncio.gather(*(onboard_verdict(v) for v in verdicts))): + a = v.target_row.chem_id for c in ov.bulk_compatibility_conflicts: groups = sorted([(c.target_group, c.target_group_name), (c.adjacent_group, c.adjacent_group_name)]) add(a, c.adjacent_chem_id, "46 CFR 150", " ↔ ".join(f"{n}({g})" for g, n in groups)) @@ -263,16 +263,24 @@ async def _compute_verdicts( 배의 등급(네 번째 반환값)을 최소 '주의'로 올린다 — 격리 적재 확인 필요이지 배정불가가 아니다(IBC Code 3.1.3). 화물별 등급(cargo_verdicts)은 인접 선석 기준 그대로다. """ - verdicts: list[_Verdict] = [] - seen_chem: set[str] = set() methods = await _unload_methods(db, request.call_sign) if request.call_sign else {} + # [2026-09-29] DB 해석은 먼저 순서대로 끝내고, 그래프 조회만 화물마다 동시에 돌린다. 운영 Neo4j(Aura)는 + # 조회 한 번이 약 300ms 라, 화물을 하나씩 기다리면 6종 배가 규칙 판정에만 약 13초 걸렸다. + adjacent_resolved = await _resolve_adjacent_cargos(db, request.adjacent_cargos) + targets: list[tuple[CargoRef, MsdsChemical]] = [] + seen_chem: set[str] = set() for cargo in request.all_targets(): - verdict = await _compute_verdict(db, neo4j_driver, request, target_cargo=cargo, unload_methods=methods) + row = await resolve_cargo(db, cargo) # 같은 물질이 다른 키(cas_no 만 / chem_id)로 두 번 올 수 있다 — 해석된 chem_id 로 한 번만. - if verdict.target_row.chem_id in seen_chem: + if row.chem_id in seen_chem: continue - seen_chem.add(verdict.target_row.chem_id) - verdicts.append(verdict) + seen_chem.add(row.chem_id) + targets.append((cargo, row)) + verdicts: list[_Verdict] = list(await asyncio.gather(*( + _compute_verdict(db, neo4j_driver, request, target_cargo=cargo, unload_methods=methods, + resolved=(row, adjacent_resolved)) + for cargo, row in targets + ))) governing = _pick_governing(verdicts) summaries = [ CargoVerdictSummary( @@ -324,14 +332,17 @@ async def _compute_verdict( *, target_cargo: CargoRef | None = None, unload_methods: dict[str, str] | None = None, + resolved: tuple[MsdsChemical, list[tuple[str, float | None, MsdsChemical]]] | None = None, ) -> _Verdict: """화물 하나에 대해 LLM을 부르지 않고 등급과 근거를 확정한다. unload_methods — 이번 입항 건 신고의 chem_id별 하역방식(_unload_methods). 요청에 하역방식이 없는 화물만 채운다. 해석된 chem_id 로 찾으므로 CAS 로만 온 화물도 맞는다. + resolved — 호출부가 미리 해석한 (대상 화물, 인접 화물). 주면 DB 를 쓰지 않아 화물 여럿을 + 동시에 돌릴 수 있다(AsyncSession 은 동시 사용 불가, 2026-09-29). """ target_cargo = target_cargo or request.target_cargo - target_row = await resolve_cargo(db, target_cargo) + target_row = resolved[0] if resolved else await resolve_cargo(db, target_cargo) if not target_cargo.unload_method_name and (unload_methods or {}).get(target_row.chem_id): target_cargo = target_cargo.model_copy(update={"unload_method_name": unload_methods[target_row.chem_id]}) @@ -340,42 +351,24 @@ async def _compute_verdict( # 나와(대한유화부두 후보) 왕복이 그대로 쌓였다 — 실측 34.2ms -> 12.3ms(-64%). # chem_id만 있는 화물은 한 번의 IN 조회로 끝내고, cas_no로 들어온 화물만 # 기존 lazy-fetch 경로(resolve_cargo -> 없으면 KOSHA API 호출)를 태운다. - adjacent_resolved = await _resolve_adjacent_cargos(db, request.adjacent_cargos) + adjacent_resolved = resolved[1] if resolved else await _resolve_adjacent_cargos(db, request.adjacent_cargos) adjacent_chem_ids = list({row.chem_id for _, _, row in adjacent_resolved}) - # [2026-08-23] 그래프 조회를 병렬로 묶었다. 일곱 개 Cypher가 서로의 결과를 - # 보지 않는 독립 조회인데 순차 await 하고 있었다 — 실측 71.4ms -> 34.1ms(-52%). - # Neo4j 드라이버는 조회마다 자체 세션을 열므로 동시 사용에 문제가 없다 - # (PostgreSQL 세션과 달리 공유 상태가 없다). - ( - raw_conflicts, - raw_imdg_conflicts, - imdg_classes, - confirmed_no_segregation_ids, - group_conflict_rows, - groups_by_chem_id, - bulk_exception_ids, - live_categories, - facts, - ) = await asyncio.gather( - find_incompatible_conflicts( - neo4j_driver, target_chem_id=target_row.chem_id, adjacent_chem_ids=adjacent_chem_ids), - find_imdg_segregation_conflicts( - neo4j_driver, target_chem_id=target_row.chem_id, adjacent_chem_ids=adjacent_chem_ids), - find_imdg_classes( - neo4j_driver, chem_ids=[target_row.chem_id, *adjacent_chem_ids]), - find_imdg_no_segregation_required( - neo4j_driver, target_chem_id=target_row.chem_id, adjacent_chem_ids=adjacent_chem_ids), - find_bulk_group_conflicts( - neo4j_driver, target_chem_id=target_row.chem_id, adjacent_chem_ids=adjacent_chem_ids), - find_bulk_groups( - neo4j_driver, chem_ids=[target_row.chem_id, *adjacent_chem_ids]), - find_bulk_exceptions( - neo4j_driver, target_chem_id=target_row.chem_id, adjacent_chem_ids=adjacent_chem_ids), - find_live_categories(neo4j_driver), - find_assessability_facts( - neo4j_driver, chem_ids=[target_row.chem_id, *adjacent_chem_ids]), - ) + # [2026-09-29] 그래프 조회 9개를 메모리 사본(graph_snapshot)으로 바꿨다. 기준 데이터라 판정마다 다시 + # 읽을 이유가 없고, 운영 Neo4j(Aura)는 조회 한 번이 약 300ms 라 화물 6종 배가 규칙 판정에만 12초 넘게 + # 걸렸다. 답은 graph_queries 의 같은 이름 Cypher 와 같다(tests/test_graph_snapshot_20260929.py). + graph = await get_snapshot(neo4j_driver) + target_id = target_row.chem_id + all_ids = [target_id, *adjacent_chem_ids] + raw_conflicts = graph.incompatible_conflicts(target_id, adjacent_chem_ids) + raw_imdg_conflicts = graph.imdg_segregation_conflicts(target_id, adjacent_chem_ids) + imdg_classes = graph.imdg_classes(all_ids) + confirmed_no_segregation_ids = graph.imdg_no_segregation_required(target_id, adjacent_chem_ids) + group_conflict_rows = graph.bulk_group_conflicts(target_id, adjacent_chem_ids) + groups_by_chem_id = graph.bulk_groups(all_ids) + bulk_exception_ids = graph.bulk_exceptions(target_id, adjacent_chem_ids) + live_categories = graph.live_categories + facts = graph.assessability_facts(all_ids) bulk_safe_exception_ids, bulk_blocked_exception_ids = bulk_exception_ids # 충돌 여부와 무관하게 대상·인접 화물 각자의 Class 자체를 별도로 조회한다 — # find_imdg_segregation_conflicts는 SEGREGATE 관계(=충돌)가 있을 때만 Class 값을 @@ -895,12 +888,17 @@ async def assess_safety( neo4j_driver: AsyncDriver, llm_client: LLMClient, request: SafetyAssessmentRequest, + *, + defer: list[Awaitable[None]] | None = None, ) -> SafetyAssessmentResult: """판정 + LLM 서술. 응답 형태는 종전과 같고 cargo_verdicts 만 늘었다. LLM 은 한 번만 부른다 — 화물마다 부르면 호출 비용이 화물 수만큼 늘어난다. [2026-09-29] 서술 대상은 대표 화물 하나가 아니라 _focus 다: 안전이 아닌 화물만, 모두 안전이면 전 화물을 묶어서. target_cargo_name 도 그 화물들 이름이다(화면의 "○○ 기준"). + + defer — 목록을 주면 LLM 서술(체크리스트·위험성 보충)을 기다리지 않고 그 목록에 넣는다. 등급·근거는 + 이미 확정돼 있으므로 호출부가 다른 LLM 호출과 동시에 기다릴 수 있다(오케스트레이터 종합 문장, 2026-09-29). """ v, verdicts, summaries, onboard, ship_level = await _compute_verdicts(db, neo4j_driver, request) needs_check = _needs_check(verdicts) @@ -908,26 +906,37 @@ async def assess_safety( focus = _focus(verdicts, onboard) methods = {x.unload_method_name for x in focus} - llm_result = await llm_client.generate_structured( - system_prompt=SYSTEM_PROMPT, - user_prompt=build_user_prompt( - target_cargo_name=_focus_name(focus), - hazard_summary=_merged_hazards(focus), - conflicts=[c for x in focus for c in x.conflicts], - bulk_compatibility_conflicts=[c for x in focus for c in x.bulk_compatibility_conflicts], - packaging_violations=[p for x in focus for p in x.packaging_violations], - unassessed_pairs=[p for x in focus for p in x.unassessed_pairs], - adjacent_count=len(request.adjacent_cargos), - # 하나라도 신고가 없으면 '검사하지 않음'으로 넘긴다 — 본 것처럼 쓰지 않게. - unload_method_name="·".join(sorted(methods)) if None not in methods else None, - onboard_conflicts=onboard, - needs_check=needs_check, - # 모든 화물의 충돌을 화물 이름과 함께 넘긴다 — 대표 화물 것만 넘기면 나머지 화물의 - # 충돌이 체크리스트에서 빠졌다(2026-09-29). - conflict_checks=[b for b in basis if b.startswith("이웃 화물 충돌") or "포장·하역방식 부적합" in b], - ), - schema=LLMAssessment, - ) + def _narrated_checklist(items: list[str]) -> list[str]: + # 중합성 화물(IMDG ", STABILIZED")이면 탱크 온도·억제제 확인 항목을 규칙으로 + # 맨 앞에 붙인다 — LLM 이 뽑을지에 맡기지 않는다(stabilized_cargo.py, 2026-09-17). + return with_stabilized_checklist( + _drop_generic_checklist(items, [_cargo_display_name(x.target_row) for x in verdicts]), + v.target_row.un_no, v.target_row.cas_no, + ) + + async def narrate() -> None: + llm_result = await llm_client.generate_structured( + system_prompt=SYSTEM_PROMPT, + user_prompt=build_user_prompt( + target_cargo_name=_focus_name(focus), + hazard_summary=_merged_hazards(focus), + conflicts=[c for x in focus for c in x.conflicts], + bulk_compatibility_conflicts=[c for x in focus for c in x.bulk_compatibility_conflicts], + packaging_violations=[p for x in focus for p in x.packaging_violations], + unassessed_pairs=[p for x in focus for p in x.unassessed_pairs], + adjacent_count=len(request.adjacent_cargos), + # 하나라도 신고가 없으면 '검사하지 않음'으로 넘긴다 — 본 것처럼 쓰지 않게. + unload_method_name="·".join(sorted(methods)) if None not in methods else None, + onboard_conflicts=onboard, + needs_check=needs_check, + # 모든 화물의 충돌을 화물 이름과 함께 넘긴다 — 대표 화물 것만 넘기면 나머지 화물의 + # 충돌이 체크리스트에서 빠졌다(2026-09-29). + conflict_checks=[b for b in basis if b.startswith("이웃 화물 충돌") or "포장·하역방식 부적합" in b], + ), + schema=LLMAssessment, + ) + result.checklist = _narrated_checklist(llm_result.checklist) + result.key_hazards = result.key_hazards or llm_result.key_hazards # [2026-08-23] 등급은 규칙엔진이 확정한다 — LLM은 서술만 담당한다. # 이전: max_risk_level(llm_result.risk_level, rule_engine_floor). LLM이 하한 @@ -940,22 +949,15 @@ async def assess_safety( # 네 등급이 모두 의미 있는 크기를 갖는다. **배정불가 62건은 불변이므로 # 자동배정에서 막히는 조합은 하나도 달라지지 않는다**(오케스트레이터는 # BLOCKED만 후보 탈락시킴) — 바뀌는 것은 화면에 표시되는 경고 등급뿐이다. - return SafetyAssessmentResult( + result = SafetyAssessmentResult( target_cargo_name=_focus_name(focus), risk_level=ship_level, - # 중합성 화물(IMDG ", STABILIZED")이면 탱크 온도·억제제 확인 항목을 규칙으로 - # 맨 앞에 붙인다 — LLM 이 뽑을지에 맡기지 않는다(stabilized_cargo.py, 2026-09-17). - checklist=with_stabilized_checklist( - _drop_generic_checklist( - llm_result.checklist, [_cargo_display_name(x.target_row) for x in verdicts], - ), - v.target_row.un_no, v.target_row.cas_no, - ), + checklist=_narrated_checklist([]), # LLM 서술 전 — narrate() 가 채운다 # [2026-09-29] 주요 위험성은 MSDS GHS 분류에서 코드가 만든다(_hazard_lines). 분류가 없는 화물뿐일 # 때만 LLM 것을 쓴다. LLM 화물 설명(cargo_profile)은 쓰지 않는다 — 여러 화물이면 가장 센 물질의 # 특성을 전부에 씌웠고(케로젠·디젤을 '고인화성'), 한 화물이어도 분류에 없는 위험을 썼다 # (가솔린 '피부에 심각한 자극' — 피부 자극 분류 없음). 사용자 결정 9/29. - key_hazards=_hazard_lines(focus) or llm_result.key_hazards, + key_hazards=_hazard_lines(focus), # reasoning 은 종전 소비처(챗봇 근거·옛 화면)용으로 남긴다 — 등급의 이유(코드) + 화물 특성(LLM). reasoning="\n".join( s for s in (_checked_facts(focus, len(request.adjacent_cargos), onboard, ship_level),) if s @@ -979,3 +981,8 @@ async def assess_safety( cargo_verdicts=summaries, onboard_conflicts=onboard, ) + if defer is not None: + defer.append(narrate()) + else: + await narrate() + return result diff --git a/app/agents/tools.py b/app/agents/tools.py index d2d916b..01ebda8 100644 --- a/app/agents/tools.py +++ b/app/agents/tools.py @@ -22,6 +22,7 @@ 등급 · 확인한 것 · 못 본 것)이 된다(B단계). """ +from collections.abc import Awaitable from datetime import datetime, timedelta, timezone from typing import Literal @@ -140,10 +141,12 @@ async def check_segregation( cargos: list[CargoRef], adjacent_cargos: list[AdjacentCargo], call_sign: str | None = None, + defer: list[Awaitable[None]] | None = None, ) -> SafetyAssessmentResult: """이 배의 화물과 이웃 화물의 혼재 등급. 등급은 규칙 하한이 정하고 LLM 은 설명만 쓴다. call_sign 을 주면 이번 입항 건 화물 신고의 하역방식을 채운다(안전 에이전트가 조회, 2026-09-29). + defer 를 주면 LLM 서술을 기다리지 않고 그 목록에 넣는다(assess_safety 참고). """ return await assess_safety( db, @@ -152,6 +155,7 @@ async def check_segregation( SafetyAssessmentRequest( target_cargo=cargo, target_cargos=cargos, adjacent_cargos=adjacent_cargos, call_sign=call_sign, ), + defer=defer, ) diff --git a/app/main.py b/app/main.py index 796a18f..0f741c4 100644 --- a/app/main.py +++ b/app/main.py @@ -22,6 +22,7 @@ from app.api.v1.scheduling import router as scheduling_router from app.api.v1.twin import router as twin_router from app.api.v1.weather import router as weather_router +from app.agents.safety.graph_snapshot import get_snapshot from app.config import get_settings from app.core.logging import configure_logging from app.database import AsyncSessionFactory @@ -50,10 +51,22 @@ async def _warm_up_postgres() -> None: logging.getLogger(__name__).warning("PostgreSQL 워밍업 실패 — 첫 요청이 느릴 수 있습니다", exc_info=True) +async def _warm_up_graph_snapshot() -> None: + """혼재 판정 그래프 사본을 미리 읽는다(2026-09-29) — 첫 판정이 적재를 기다리지 않게. + + 실패해도 앱을 막지 않는다. 사본은 첫 판정 때 다시 읽는다. + """ + try: + await get_snapshot(neo4j_client.driver) + except Exception: # noqa: BLE001 - 워밍업 실패는 기동을 막지 않는다 + logging.getLogger(__name__).warning("혼재 판정 그래프 사본 워밍업 실패 — 첫 판정이 느릴 수 있습니다", exc_info=True) + + @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: await neo4j_client.driver.verify_connectivity() await _warm_up_postgres() + await _warm_up_graph_snapshot() # 08_스케줄링_전면재설계_자동배정_설계문서.md §5.1·§5.5 — 입항 자동추천·출항 # 자동해제 백그라운드 잡. 둘 다 실패해도 API 자체는 계속 떠 있어야 하므로 # 앱 시작을 막지 않는다(스케줄러 자체 등록 실패만 여기서 전파됨). diff --git a/tests/test_concurrent_judgment_20260929.py b/tests/test_concurrent_judgment_20260929.py new file mode 100644 index 0000000..fb79ba7 --- /dev/null +++ b/tests/test_concurrent_judgment_20260929.py @@ -0,0 +1,83 @@ +"""판정 한 번의 대기 줄이기 (2026-09-29). + +운영 Neo4j(Aura)는 조회 한 번이 약 300ms 라, 화물 6종 배의 근거 패널이 약 20초 걸렸다. + · 화물별 판정·같은 선박 비교를 화물마다 동시에 돌린다(12번 차례 → 2번). + · 혼재 서술 LLM(체크리스트)을 종합 문장 LLM 과 함께 기다린다(차례로 2번 → 동시에). +로컬 DB·Neo4j 를 쓴다(conftest). 화물 2종 이상인 접안 선박이 없으면 건너뛴다. +""" +import asyncio +from datetime import datetime, timedelta, timezone + +import pytest +from sqlalchemy import text + +from app.agents.orchestrator.schemas import LLMSummary, OrchestratorRequest +from app.agents.orchestrator.service import orchestrate +from app.agents.safety.schemas import CargoRef, LLMAssessment +from app.agents.scheduling.schemas import VesselSpec +from app.llm.null_client import NullLLMClient + +DELAY = 0.5 + +_Q_TARGET = text(""" + SELECT vc.callsgn, fa.wharf_name, array_agg(DISTINCT cm.chem_id) AS chems + FROM mart.vessel_presence p + JOIN mart.vessel_current_call vc ON vc.callsgn = upper(btrim(p.callsgn)) + JOIN mart.cargo_msds cm ON cm.port_call_key = vc.port_call_key AND cm.chem_id IS NOT NULL + JOIN mart.facility_alias fa ON fa.source_name = p.berth_name AND fa.facility_type = 'BERTH' + WHERE p.presence_zone = 'BERTH' + GROUP BY 1, 2 HAVING count(DISTINCT cm.chem_id) >= 2 + ORDER BY 1 LIMIT 1 +""") + + +class _SlowLLM: + """호출마다 DELAY 만큼 기다리고, 동시에 몇 개가 떠 있었는지 센다.""" + + def __init__(self): + self.active = 0 + self.max_active = 0 + + async def generate_structured(self, *, system_prompt, user_prompt, schema): + self.active += 1 + self.max_active = max(self.max_active, self.active) + try: + await asyncio.sleep(DELAY) + finally: + self.active -= 1 + if schema is LLMAssessment: + empty = await NullLLMClient().generate_structured(system_prompt="", user_prompt="", schema=schema) + return empty.model_copy(update={"checklist": ["시험용 확인 항목"]}) + if schema is LLMSummary: + return LLMSummary(summary="시험용 종합 문장입니다.") + raise AssertionError(schema) + + +async def _request(db): + row = (await db.execute(_Q_TARGET)).first() + if row is None: + pytest.skip("화물 2종 이상인 접안 선박이 로컬 DB 에 없다") + cs, wharf, chems = row + chems = sorted(chems) + now = datetime.now(timezone.utc) + return OrchestratorRequest( + vessel=VesselSpec(draught_m=6.0, dwt_t=None, name_hint=cs, call_sign=cs), + cargo=CargoRef(chem_id=chems[0]), cargos=[CargoRef(chem_id=c) for c in chems[1:]], + window_start=now, window_end=now + timedelta(hours=8), + assigned_wharf_name=wharf, target_source="AIS", + ) + + +@pytest.mark.asyncio +async def test_safety_narration_and_summary_llm_overlap(db, neo4j): + request = await _request(db) + llm = _SlowLLM() + result = await orchestrate(db, neo4j, llm, request) + if result.safety_assessment is None: + pytest.skip("선석·기상 단계에서 끝나 혼재 심사까지 가지 않았다") + # 두 LLM 이 동시에 떠 있었다 + assert llm.max_active == 2 + # 나중에 채운 체크리스트가 결과에 실렸다(결과가 혼재 판정 객체를 복사하지 않는다) + assert "시험용 확인 항목" in result.safety_assessment.checklist + assert result.summary == "시험용 종합 문장입니다." + diff --git a/tests/test_graph_snapshot_20260929.py b/tests/test_graph_snapshot_20260929.py new file mode 100644 index 0000000..f67239c --- /dev/null +++ b/tests/test_graph_snapshot_20260929.py @@ -0,0 +1,53 @@ +"""혼재 판정 그래프 메모리 사본이 Cypher 조회와 같은 답을 내는가 (2026-09-29). + +판정은 graph_queries 의 Cypher 9개 대신 graph_snapshot 사본을 쓴다. 로컬 Neo4j 의 모든 Chemical 을 +대상으로, 이웃 화물 묶음을 바꿔 가며 두 쪽의 답을 대조한다(행 순서는 Neo4j 도 정하지 않으므로 무시). +""" +import json +import random + +import pytest + +from app.agents.safety import graph_queries as gq +from app.agents.safety.graph_snapshot import load_snapshot + + +def _rows(rows): + return sorted(json.dumps(r, ensure_ascii=False, sort_keys=True) for r in rows) + + +@pytest.mark.asyncio +async def test_snapshot_matches_cypher_for_every_chemical(neo4j): + snap = await load_snapshot(neo4j) + ids = sorted(snap.names) + assert len(ids) > 50, "로컬 그래프에 화학물질이 거의 없다" + rng = random.Random(20260929) + assert snap.live_categories == await gq.find_live_categories(neo4j) + + for target in ids: + # 이웃 묶음: 무작위 8종 + 자기 자신 + 그래프에 없는 id(PostgreSQL 에만 있는 화물) + adjacent = list(dict.fromkeys(rng.sample(ids, 8) + [target, "NOT_IN_GRAPH"])) + all_ids = [target, *adjacent] + kw = dict(target_chem_id=target, adjacent_chem_ids=adjacent) + + assert _rows(snap.incompatible_conflicts(target, adjacent)) == _rows(await gq.find_incompatible_conflicts(neo4j, **kw)), target + assert _rows(snap.imdg_segregation_conflicts(target, adjacent)) == _rows(await gq.find_imdg_segregation_conflicts(neo4j, **kw)), target + assert snap.imdg_no_segregation_required(target, adjacent) == await gq.find_imdg_no_segregation_required(neo4j, **kw), target + assert snap.imdg_classes(all_ids) == await gq.find_imdg_classes(neo4j, chem_ids=all_ids), target + assert _rows(snap.bulk_group_conflicts(target, adjacent)) == _rows(await gq.find_bulk_group_conflicts(neo4j, **kw)), target + assert snap.bulk_groups(all_ids) == await gq.find_bulk_groups(neo4j, chem_ids=all_ids), target + assert snap.bulk_exceptions(target, adjacent) == await gq.find_bulk_exceptions(neo4j, **kw), target + assert snap.assessability_facts(all_ids) == await gq.find_assessability_facts(neo4j, chem_ids=all_ids), target + + +@pytest.mark.asyncio +async def test_snapshot_catches_known_conflicts(neo4j): + """무작위 묶음이 충돌을 거의 안 뽑으면 위 대조가 빈 목록끼리만 비교한다 — 충돌이 실제로 있는 쌍도 대조한다.""" + snap = await load_snapshot(neo4j) + pairs = [(a, b) for a in snap.groups for b in snap.groups if snap.bulk_group_conflicts(a, [b])][:40] + pairs += [(a, b) for a in snap.avoids for b in snap.classified if snap.incompatible_conflicts(a, [b])][:40] + assert len(pairs) >= 40, "충돌 쌍을 충분히 찾지 못했다" + for a, b in pairs: + kw = dict(target_chem_id=a, adjacent_chem_ids=[b]) + assert _rows(snap.bulk_group_conflicts(a, [b])) == _rows(await gq.find_bulk_group_conflicts(neo4j, **kw)) + assert _rows(snap.incompatible_conflicts(a, [b])) == _rows(await gq.find_incompatible_conflicts(neo4j, **kw))