xsoft_comply Python SDK

AI 기본법 대응 거버넌스 플랫폼 xsoft_comply의 공식 Python SDK.

설치

0) 가상환경(venv) 먼저 — externally-managed-environment 에러 예방

Ubuntu 23.04+ · Debian 12+ · Fedora 38+ 등 최신 배포판의 시스템 파이썬은 PEP 668 로 잠겨 있어 pip install 이 다음 메시지로 멈춥니다.

error: externally-managed-environment
× This environment is externally managed

시스템 파이썬을 건드리지 말고 가상환경에 설치하세요(권장).

# 1) venv 만들기 (python3-venv 가 없다면: sudo apt install python3-venv)
python3 -m venv ~/venv

# 2) venv 의 pip 으로 설치 — activate 없이 경로로 직접 불러도 됩니다
~/venv/bin/pip install --upgrade pip

# 3) 실행도 같은 venv 로
~/venv/bin/python -c "import xsoft_comply; print(xsoft_comply.__version__)"

셸에 붙여 쓰려면 source ~/venv/bin/activate 후 pip·python 을 그대로 쓰면 됩니다 (빠져나올 때는 deactivate).

venv 를 만들 수 없는 환경(컨테이너 베이스 이미지 등)에서만 마지막 수단으로 pip install --break-system-packages ... 를 씁니다. 시스템 패키지와 충돌할 수 있어 권장하지 않습니다.

1) 설치 방법 — 지금은 로컬(소스) 설치

xsoft-comply 는 아직 PyPI 에 올라가 있지 않습니다. 지금은 저장소를 받아 로컬로 설치하세요.

# 저장소 루트에서
~/venv/bin/pip install -e sdk/python

PyPI 공개 후에는 아래가 표준 설치 경로가 됩니다.

~/venv/bin/pip install xsoft-comply

★ 안전하게 설치하기 (권장)

이 SDK 는 여러분의 서버 안에서 실행됩니다. 공급망 공격(타이포스쿼팅·계정 탈취)을 막으려면 이름을 정확히 쓰고 해시를 고정하세요.

# ① 정확한 배포 이름 = xsoft-comply  (import 이름은 xsoft_comply)
#    비슷한 이름(xsoftcomply · xsoft-compliance 등)은 우리 것이 아닙니다.
# ② 해시 고정 = requirements.txt 에 박아두고 --require-hashes 로 설치
~/venv/bin/pip install --require-hashes -r requirements.txt

requirements.txt 예시(해시는 릴리스 노트의 값을 그대로 복사):

xsoft-comply==0.1.0 \
    --hash=sha256:<릴리스 노트에 공지된 sha256>

설치 후 확인:

~/venv/bin/python -c "import xsoft_comply; print(xsoft_comply.__version__)"

우리가 지키는 것 — 외부 의존성 0(표준 라이브러리만), 설치 시 실행되는 코드 없음 (setup.py·.pth 없음), 배포는 Trusted Publishing(OIDC)+2FA 로만. import 만으로는 네트워크를 열지 않습니다(init() 을 불러야 전송 스레드가 뜹니다).

외부 의존성 없음(표준 라이브러리만). Python 3.9 이상.

빠른 시작

import xsoft_comply as xc

xc.init(
    api_key="xsk_live_...",            # API 키 (대시보드에서 발급)
    endpoint="https://your-domain.com/api/v1",
    log_llm_content=False,             # 기본값 · 원문 미전송
)

with xc.session(agent_id="support-bot") as s:
    s.llm_call(
        model="gpt-4o-mini",
        provider="openai",
        tokens_in=512,
        tokens_out=128,
        latency_ms=820,
    )
    s.tool_call(tool="crm.lookup", status="ok", latency_ms=12)
    s.data_access(resource="customer_profile", scope="read", record_count=1)
    s.decision(decision="escalate", options=3, confidence=0.72)
    s.human_review(human_review="approved", reviewer_role="manager")
    s.action(action="ticket.create", status="ok")

세션 컨텍스트를 벗어나면 session_end 이벤트가 자동으로 전송된다.

조항 증적 (제31~36조) · 2026-08-20

event_type 은 6종이다 — session_start · llm_call · tool_call · data_access · action · session_end.

decision() · human_review() 는 메서드로 그대로 남는다. 내부에서 action + 조항 증적(제34조 고영향 책무)으로 조립돼 나간다 → 기존 코드는 그대로 돌아간다. 바뀌는 것은 «어디에 적재되는가» 뿐이다(compliance_evidence + ev_high_impact 로 정규화 분배).

with xc.session(agent_id="loan-bot") as s:
    # 한 이벤트에 여러 조항 증적을 담을 수 있다(배열)
    s.action("loan.approve", status="ok", evidence=[
        xc.evidence_block(34, supervisor_id="sup-1", decision_basis="심사 기준표 v3"),
        xc.evidence_block(31, ai_generated_label="labeled", label_method="watermark",
                          output_id="out-99"),
    ])

    # 런타임이 아닌 «선언» 증적도 같은 길로 보낸다(원천 우회 금지 = hash-chain 을 받는다)
    s.evidence(36, source="declared", evidence_key="agent-2026",
               company_country="US", has_korean_address="false",
               domestic_agent_name="…", domestic_agent_contact="…")

• evidence_key = 개발자 멱등키. 같은 선언을 여러 번 보내도 증적은 1행이다. 생략하면 이벤트마다 새 키가 되어 매번 새 증적이 된다.
• 표준필드는 조항별 화이트리스트만 받는다. 오타·미정의 키는 경고 후 드롭된다(이벤트는 거부되지 않는다).
• 값에 원문(prompt/response)이나 판정어("준수/위반")를 넣지 마라.

확인 3종 — 보내기 전에 눈으로 확인한다

xc.init(..., dry_run=True)     # ① 전송 0. payload·조항매핑·필수 표준필드·마스킹 결과를 콘솔에 출력
report = xc.validate(evt)      # ② 조항별 필수 표준필드 «사실» 집계 (판정 아님)
print(xc.preview())            # ③ 1콜 output 로컬 확인

python -m xsoft_comply.preview   # 설치 확인용 1콜 출력

validate() 는 「필수 N개 중 M개가 비어 있다」 는 사실만 돌려준다. 준수·위반 여부를 판정하지 않는다(변호사법 109조).

LangChain 통합

from langchain_openai import ChatOpenAI

handler = xc.langchain_handler()
llm = ChatOpenAI(callbacks=[handler])

with xc.session(agent_id="lc-agent") as s:
    response = llm.invoke("안녕하세요")

langchain 또는 langchain-core 가 설치돼 있으면 BaseCallbackHandler 를 상속한다. 없으면 순수 Python 클래스로 동작한다.

OpenAI Agents SDK 통합

import asyncio
import xsoft_comply as xc
from agents import Agent, Runner, function_tool

@function_tool
def lookup_order(order_id: str) -> str:
    return f"주문 {order_id} 상태 = 배송중"

agent = Agent(name="order-bot", model="gpt-4o-mini", tools=[lookup_order])
xc.instrument(agent)                     # AgentHooks 주입

async def main():
    with xc.session(agent_id="order-bot"):
        await Runner.run(
            agent, "주문번호 A-1024 상태 알려줘.",
            hooks=xc.agents_run_hooks(),  # RunHooks 주입(핸드오프까지 잡는다)
        )

asyncio.run(main())

기록되는 것 = 에이전트 시작·종료, 도구 호출(이름·소요시간), LLM 호출(모델·토큰 수), 핸드오프. 원문(입력·출력)은 보내지 않고 길이만 남긴다. 두 훅을 함께 써도 이중기록되지 않는다(실행훅이 에이전트훅이 이미 붙은 에이전트는 건너뛴다).

이벤트 종류별 로그 토글

xc.init(..., log_events={"tool_call": False, "data_access": False})

끈 종류는 seq 를 소비하지 않는다 → 체인이 끊기지 않는다. session_start · session_end 는 끌 수 없다(끄면 trace 자체가 성립하지 않는다).

정책 자동차단

rules = [
    {"id": "no-mass-delete", "when": {"tool": "crm.delete_all"},
     "effect": "block", "reason": "대량 삭제는 사람 승인 후에만"},
    {"id": "export-warn", "when_re": {"tool": r"^crm\.export"}, "effect": "warn"},
]
xc.init(..., policy=rules)              # policy_enforce=False 면 기록만 하고 막지 않는다

with xc.session(agent_id="support-bot") as s:
    outcome = s.run_tool("crm.delete_all", delete_everything)
    if outcome.blocked:
        ...                              # 도구는 **호출되지 않았다**

차단은 예외를 던지지 않는다. outcome.blocked 로 알려주고, 감사기록에는 policy_result="blocked" · metadata.policy_rule 이 남는다. s.check("tool_call", tool="…") 로 평가만 할 수도 있다.

전자서명 (ed25519 · 의존성 0)

xc.init(..., sign=True)                 # 키가 없으면 로컬에 만들어 0600 으로 보관
print(xc.signing_public_key())          # 감사인에게 넘길 공개키(개인키는 반환 경로 없음)

서명 대상 = trace_id·seq·event_id·type·occurred_at·라벨 컬럼·metadata 해시·직전 서명. DB 에 저장된 행만 보고 검증할 수 있고, 값이 한 글자만 달라져도 깨진다. 서명은 전송 스레드에서 한다(호출 스레드는 O(1) 유지).

원문 보관 (객체스토리지 · 옵션)

xc.init(
    ...,
    log_llm_content=True,
    r2={"endpoint": "https://<account>.r2.cloudflarestorage.com",
        "bucket": "comply-raw", "access_key_id": "...", "secret_access_key": "..."},
)
s.llm_call(model="gpt-4o-mini", content=prompt_and_response)

원문은 고객 소유 버킷으로 직접 올라간다(우리 수집 API 를 거치지 않는다). 이벤트에는 위치(content_uri)와 길이·해시만 남는다. 버킷 장애 시에도 이벤트는 정상 적재되고 원문은 로컬 스풀에도 남기지 않는다.

보관기간(retention)

xc.init(..., retention_days=3650)       # 기본 1825일(5년) · 초장기 애드온은 늘려 잡는다

이벤트 metadata 에 _retain_until(만료일)이 남는다. 만료분 정리는 운영측 도구가 한다 (sdk/tools/retention.py).

수동 flush / 종료

xc.flush(timeout=5.0)   # 버퍼가 빌 때까지 최대 5초 대기
xc.shutdown()           # 전송 스레드 정지

프로세스 종료 시 atexit 핸들러가 자동으로 최대 2초 flush 를 시도한다.

---

장애격리 설계: "절대 안 멈추는" 이유

xsoft_comply SDK 는 고객 애플리케이션의 가용성을 절대 침해하지 않는다. 다음 8개 계약이 이를 보장한다.

1. 모든 공개 함수는 예외를 밖으로 내보내지 않는다

init() · session() · s.llm_call() 등 모든 공개 함수는 최상위 try/except BaseException 으로 감싼다. KeyboardInterrupt · SystemExit 만 재raise 한다(프로세스 종료 신호를 막지 않기 위해).

# SDK 내부 구조 (단순화)
def llm_call(self, ...):
    try:
        evt = build_event(...)
        buffer.put(evt)   # O(1) 비블로킹
    except (KeyboardInterrupt, SystemExit):
        raise
    except BaseException:
        pass              # 모든 오류 삼킨다

2. 호출 스레드는 네트워크를 만지지 않는다

이벤트 기록은 queue.put_nowait() (O(1)) 만 실행한다. 큐가 가득 차면 가장 오래된 항목을 먼저 버리고 새 항목을 넣는다. 블로킹 없음.

고객 스레드  →  put_nowait(event)  →  [queue]  →  daemon 스레드  →  HTTP POST

3. 전송은 daemon 백그라운드 스레드 1개

threading.Thread(daemon=True) 로 생성된다. daemon 스레드는 Python 인터프리터가 종료될 때 강제 종료되므로 고객 프로세스 종료를 막지 않는다.

4. HTTP 타임아웃: connect 2s / read 3s

urllib.request 에 합산 5s 타임아웃을 설정한다. requests · httpx 등 외부 라이브러리 의존성 없음.

5. 전송 실패 시 디스크 스풀 → 지수 백오프

연결 거부 · 5xx · 타임아웃 등 모든 네트워크 장애 시:

1. 이벤트를 ~/.xsoft_comply/spool/events.jsonl 에 JSONL 로 append
2. 백오프 대기 (1s → 2s → 4s → 8s → 30s 상한)
3. 백오프 후 재전송 시도

6. 스풀 파일 상한 (기본 32 MB)

스풀 디렉터리 총 크기가 32 MB 를 초과하면 오래된 파일부터 삭제한다. 디스크 쓰기 실패도 삼킨다.

7. init() 전에 이벤트를 기록해도 죽지 않는다

큐가 None 인 상태에서 buffer.put() 은 즉시 반환(no-op) 된다.

8. atexit에 flush 등록 (타임아웃 2s)

import atexit
atexit.register(_atexit_handler)   # init() 최초 호출 시 1회 등록

def _atexit_handler():
    flush(timeout=2.0)             # 2초 후 반드시 반환

프로세스가 정상 종료될 때 버퍼에 남은 이벤트를 최대 2초 안에 전송 시도한다. 2초 안에 완료되지 않아도 프로세스 종료를 막지 않는다.

---

원문 미저장 보장

log_llm_content=True 로 설정해도 원문(prompt/response) 은 절대 전송되지 않는다. 대신 content_len (길이) 와 content_sha256_8 (sha256 앞 8자) 만 metadata 에 포함된다.

다음 metadata 키가 있으면 SDK 에서 자동 제거하고 _dropped_keys 에 이름만 기록한다: prompt, response, messages, content, input, output 등 25개.

PII 마스킹

전송 전 로컬에서 자동 마스킹된다:

• 한국 주민등록번호 \d{6}-?\d{7} → ******-*******
• 휴대전화 01X-XXXX-XXXX → 010-****-****
• 이메일 → a***@domain
• 카드번호 16자리 → 앞4-****-****-뒤4

마스킹 발생 시 metadata.pii_masked = true.

라이선스

MIT