# Langfuse를 JWT를 통해 LLM과 통합하기: 동적 권한 부여 문제 해결
Langfuse는 LLM 요청 추적, 프롬프트 관리, 모델 평가를 위한 강력한 도구입니다. 그러나 엔터프라이즈 환경에서는 중요한 문제가 발생합니다: 시스템이 추론 게이트웨이 접근을 위해 동적 JWT 토큰을 요구하는 반면, Langfuse는 정적 API 키만 지원합니다. 이 불일치로 인해 직접 통합이 불가능합니다. 해결책은 정적 권한 부여를 동적으로 변환하는 얇은 프록시 서비스를 만드는 것입니다.
OpenAI와 유사한 API와 호환되는 아키텍처 솔루션을 살펴보겠습니다. 이는 Langfuse에 대한 투명성을 유지하면서 엔터프라이즈 인프라 보안 요구 사항을 충족합니다.
프록시 아키텍처: 정적과 동적 간의 간극 메우기
핵심 문제는 패러다임 충돌입니다:
- Langfuse는 고정 연결 설정을 API 키와 함께 사용합니다
- 엔터프라이즈 추론 게이트웨이는 짧은 수명 JWT 토큰을 TTL이 짧게 요구합니다
프록시는 어댑터 역할을 하며, 세 가지 중요한 작업을 수행합니다:
- Langfuse로부터 OpenAI 호환 요청 수신
- client_credentials 흐름을 통해 동적으로 JWT 획득
- Authorization 헤더 교체를 통해 상위 API로 요청 투명하게 전달
이렇게 하면 코드 변경 없이 표준 Langfuse 워크플로를 유지하면서 내부 인프라 보안 정책을 준수할 수 있습니다.
기술 구현: 단계별 구축
Step 1: FastAPI를 사용한 비동기 코어
프록시는 네트워크 호출에서 주로 부하가 발생하는 I/O 중심 서비스입니다. FastAPI와 httpx.AsyncClient를 사용한 비동기 아키텍처는 다음을 제공합니다:
- 병렬 요청의 비차단 처리
- 효율적인 연결 관리
- 최소 자원으로 최대 처리량
프로젝트 구조:
app/
├── __init__.py
├── app.py # Entry point and lifecycle management
├── routers.py # Request routing
├── proxy_service.py # Proxy logic
├── jwt_provider.py # Token retrieval and caching
└── settings.py # Pydantic-based configuration라이프사이클 관리 구현 예시:
@asynccontextmanager
async def lifespan(app: FastAPI):
configure_logging()
logger = get_logger()
logger.info("Microservice is starting...")
proxy_service = LLMProxyService()
app.state.proxy_service = proxy_service
try:
yield
finally:
await proxy_service.aclose()
logger.info("Microservice is shutting down...")Step 2: OpenAI 호환 엔드포인트 라우팅
최소 요구 라우트:
/v1/chat/completions— 생성을 위한 주요 엔드포인트/v1/models— 사용 가능한 모델 확인/healthz— 오케스트레이터를 위한 헬스 체크
클라이언트 라이브러리 호환성을 깨뜨리지 않기 위해 원본 경로를 유지하는 것이 중요합니다. 라우터 구현:
@root_router.post("/v1/chat/completions")
async def proxy_chat_completions(request: Request):
proxy_service = _get_proxy_service(request)
return await proxy_service.proxy_chat_completions(request)Step 3: Pydantic Settings를 통한 동적 구성
pydantic-settings를 사용하면 환경 간 유연한 매개변수 관리가 가능합니다. 주요 매개변수:
LLM_BASE_URL— 추론 게이트웨이 주소SYSTEM_ID_HEADER_NAME— 클라이언트 식별자 헤더JWT_ISSUER및JWT_SCOPE— OIDC 제공자 매개변수- 모든 외부 호출에 대한 타임아웃
설정 클래스 예시:
class Settings(BaseSettings):
llm_base_url: str = Field(default="", alias="LLM_BASE_URL")
system_id_header_name: str = Field(
default="systemId", alias="SYSTEM_ID_HEADER_NAME")
jwt_issuer: str = Field(default="", alias="JWT_ISSUER")Step 4: 캐싱을 지원하는 스마트 JWT 제공자
중요한 최적화는 만료까지 토큰 캐싱입니다. 워크플로 알고리즘:
- 수신 요청에서 client_id와 client_secret 추출
- OIDC 디스커버리를 통해 token_endpoint 자동 발견
- client_credentials 흐름을 사용해 토큰 요청
- expires_in을 고려해 access_token 캐싱
캐싱 없이 고부하 시 인증 서비스가 병목이 됩니다. LRU 캐시를 사용한 구현:
@lru_cache(maxsize=128)
async def get_token(
self,
client_id: str,
client_secret: str
) -> str:
# Token retrieval and caching logicStep 5: 투명한 요청 프록싱
주요 구현 측면:
- 원본 쿼리 매개변수와 요청 본문 보존
- 전송 헤더 필터링 (Host, Content-Length)
- Authorization을 새로운 JWT로 교체
- 모든 HTTP 상태 코드를 수정 없이 전달
페이로드를 절대 수정하지 마세요 — 이는 Langfuse 클라이언트 라이브러리 호환성을 보장합니다. 처리 예시:
async def _forward(self, request: Request, path: str) -> Response:
client_id = request.headers.get(self._settings.system_id_header_name)
authorization = request.headers.get("authorization")
_, _, client_secret = authorization.partition(" ")
token = await self._token_provider.get_token(
client_id=client_id,
client_secret=client_secret,
)
return await self._send(
request=request,
url=f"{self._settings.llm_base_url}{path}",
token=token,
)주요 요점: 중요한 사항
- 프록시를 필수 어댑터로: 짧은 수명 토큰으로 인해 Langfuse 직접 통합 불가능 — 중간 계층 필요
- 토큰 캐싱이 핵심: 없으면 부하 시 인증 서비스 병목
- 투명성 유지: 페이로드나 HTTP 상태 코드 수정 금지 — 호환성 보장
- 데이터 보안: Authorization 헤더 로깅 금지; 프록시 접근 제한
- 유연한 구성: .env 매개변수화로 환경 간 쉽게 적응
Langfuse 설정: 최종 단계
Langfuse 인터페이스 구성:
- Base URL을 프록시 주소로 설정 (예:
http://proxy-service/v1) - API Key에 client_secret 입력 (Bearer 자격 증명으로 전달)
- Custom Headers에 SYSTEM_ID_HEADER_NAME의 이름으로 헤더 추가
이렇게 하면 폐쇄 루프가 생성됩니다: Langfuse가 고정 매개변수로 요청 전송 → 프록시가 동적으로 JWT 가져옴 → 요청이 요구 형식으로 추론 게이트웨이에 도달.
중요한 테스트 시나리오:
- 토큰 만료
- 인증 서비스 오류
- 고부하 (캐싱 효율성 검증)
이 아키텍처는 LLM 요청 모니터링을 위한 모든 Langfuse 이점을 유지하면서 엄격한 엔터프라이즈 보안 요구 사항을 충족합니다. 기본 FastAPI와 OIDC 프로토콜 지식으로 구현에 1-2일 소요됩니다.
— Editorial Team
아직 댓글이 없습니다.