코딩 컨벤션

언어 및 도구

항목	도구	설정
Python	3.14	requires-python = "==3.14.*"
포맷팅	Ruff	line-length = 88, target-version = "py314"
린팅	Ruff	E, F, UP, B, SIM, I, FAST
타입 체크	basedpyright	정적 타입 분석
패키지 관리	uv	uv sync, uv run

# 전체 코드 품질 검사 (커밋 전 필수)
uv run ruff format .
uv run ruff check --fix .
uv run basedpyright .

---

프로젝트 구조 컨벤션

모듈 레이아웃

src/truloop_assistant/
├── agent/          # pydantic-ai Agent 정의 (비즈니스 로직 없음)
├── integrations/   # 외부 서비스 클라이언트 (HTTP, SDK)
├── routes/         # FastAPI 라우트 핸들러 (HTTP 관심사만)
├── services/       # 비즈니스 로직 (Agent 실행, 알림, 이벤트 처리)
├── schemas/        # Pydantic 모델 (요청/응답/도메인)
└── utils/          # 순수 유틸리티 함수

원칙:

• routes/는 HTTP 파싱, Background task 시작, 즉시 응답만 담당 (25-88줄 이내)
• services/에 모든 비즈니스 로직 집중
• integrations/는 외부 API 호출만 캡슐화 (비즈니스 로직 금지)
• agent/는 Agent 설정과 Tool 정의만 담당

---

타입 힌트 컨벤션

모든 함수에 타입 힌트 필수

# Good
async def get_user_profile(user_eid: str) -> User:
    ...

# Bad - 타입 힌트 누락
async def get_user_profile(user_eid):
    ...

from __future__ import annotations 사용

순환 참조 방지 및 런타임 타입 평가 지연을 위해 파일 상단에 추가:

from __future__ import annotations

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from truloop_assistant.integrations.sendbird.client import SendbirdClient

Union 반환 타입 패턴 (도구)

Tool 함수는 성공 시 Pydantic 모델, 실패 시 str 에러 메시지를 반환:

@toolset.tool
async def search_places(
    ctx: RunContext[AgentDeps],
    query: str = Field(description="검색어"),
) -> SearchPlacesResponse | str:
    """장소를 검색합니다."""
    if not settings.google_maps_api_key:
        return "Google Places API key is not configured."
    ...

---

Agent 도구 작성 컨벤션

FunctionToolset 패턴

모듈 레벨에서 FunctionToolset 인스턴스를 생성하고 @toolset.tool 데코레이터로 등록:

from pydantic import BaseModel, Field
from pydantic_ai import FunctionToolset, RunContext

from truloop_assistant.agent.deps import AgentDeps

# 모듈 레벨에서 Toolset 인스턴스 생성
my_toolset = FunctionToolset[AgentDeps]()

# Response DTO 정의
class MyToolResponse(BaseModel):
    """도구 응답 DTO."""
    result: str

# Tool 등록
@my_toolset.tool
async def my_tool(
    ctx: RunContext[AgentDeps],
    param: str = Field(description="파라미터 설명 (Agent가 읽음)"),
) -> MyToolResponse | str:
    """도구 설명 (Agent가 호출 결정에 사용).

    상세 사용법, 주의사항 등을 docstring에 기술합니다.
    """
    ...

도구 등록 (__init__.py)

새 Toolset을 agent/tools/__init__.py의 all_toolsets 리스트에 추가:

all_toolsets: list[FunctionToolset[AgentDeps]] = [
    datetime_toolset,
    event_toolset,
    user_toolset,
    # ... 새 toolset 추가
]

도구 설계 원칙

• 단일 책임: 하나의 도구 = 하나의 명확한 기능
• Docstring 필수: Agent가 docstring을 읽고 호출 여부를 결정
• Field(description=...) 필수: 모든 파라미터에 설명 제공
• 에러 처리: API 오류는 ModelRetry 예외로 Agent에게 재시도 기회 제공, 내부 오류는 str 에러 메시지 반환
• ctx 접근: ctx.deps를 통해 런타임 의존성 접근 (절대 import로 직접 접근하지 않음)

---

Pydantic 모델 컨벤션

Response DTO 패턴

도구와 서비스의 반환값은 Pydantic BaseModel로 정의:

class SearchPlacesResponse(BaseModel):
    """search_places 도구 응답 DTO."""
    places: list[PlaceResult]
    query: str

Frozen 모델 (불변)

런타임 의존성 등 변경되면 안 되는 모델은 frozen=True:

class AgentDeps(BaseModel):
    model_config = {"frozen": True, "arbitrary_types_allowed": True}

---

비동기 패턴 컨벤션

Background Task 패턴

HTTP 라우트에서 Background task를 시작하고 즉시 202 Accepted 반환:

@router.post("/endpoint", status_code=202)
async def handle(payload: ..., processor: ProcessorDep) -> Response:
    create_tracked_task(
        processor.process(payload),
        name=f"process:{payload.id}",
    )
    return Response(status="accepted")

ConversationTaskRegistry 패턴

대화별로 하나의 활성 task만 유지. 새 메시지 수신 시 기존 task를 자동 취소:

await task_registry.start_or_replace(
    conversation_id=channel_url,
    user_message=message_text,
    coro=processor.process_user_message(payload, persona=persona),
    history_service=history_service,
    task_name=f"process:{user_id}:{eid}",
)

HTTP 클라이언트 수명

• 공유 클라이언트: Lifespan에서 생성하여 app.state에 저장 (Sendbird, truloop-core, Redis)
• 격리 클라이언트: 비회원 메시징 등 background task 내에서 async with httpx.AsyncClient()로 독립 생성

---

로깅 컨벤션

structlog + 표준 logging

logging.getLogger(__name__)를 사용하며, structlog가 JSON 형식으로 변환:

import logging

logger = logging.getLogger(__name__)

# extra dict로 구조화 메타데이터 전달
logger.info(
    "Processing started",
    extra={"user_eid": user_eid, "channel": channel_url},
)

로깅 규칙

• logger.info(): 주요 비즈니스 이벤트 (Agent 실행, 메시지 발송 등)
• logger.warning(): 복구 가능한 문제 (API 타임아웃 후 fallback 등)
• logger.error(): 복구 불가능한 오류
• logger.exception(): 예외와 함께 스택 트레이스 기록 (Sentry 이벤트 자동 생성)
• logger.debug(): 개발 디버깅용 (프로덕션에서 비활성)

---

테스트 컨벤션

AI 평가 우선 (pydantic-evals)

Agent 행동은 YAML 기반 평가 데이터셋으로 테스트:

# evals/datasets/datetime_tools.yaml
- inputs:
    prompt: "오늘 무슨 요일이야?"
  expected_output:
    tool_calls: ["get_day_of_week"]

유닛 테스트는 API 계약만

# tests/test_api_contracts.py
async def test_health_check(client):
    response = await client.get("/health")
    assert response.status_code == 200

테스트 도구

도구	용도
pytest + pytest-asyncio	비동기 테스트 실행
fakeredis	Redis 모킹
respx	httpx 모킹
freezegun	시간 고정
pydantic-evals	AI 평가

---

환경 설정 컨벤션

pydantic-settings 기반

모든 설정은 config.py의 Settings 클래스에서 환경변수로 관리:

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        case_sensitive=False,
        extra="ignore",
    )
    openrouter_api_key: str = ""

선택적 기능 패턴

API 키가 비어있으면 해당 기능을 비활성화하고, 관련 도구는 graceful하게 에러 메시지 반환:

@property
def rag_enabled(self) -> bool:
    return bool(self.openai_api_key)

프로덕션 검증

@model_validator로 프로덕션 환경에서 필수 설정 누락 시 시작 실패:

@model_validator(mode="after")
def validate_production_settings(self) -> Self:
    if self.env != "production":
        return self
    if not self.openrouter_api_key:
        errors.append("openrouter_api_key is required in production")
    ...

---

변경 이력

날짜	변경 내용
2026-03-10	초기 작성: FunctionToolset, 비동기, 로깅, 테스트, 설정 컨벤션