현재 linkai 라는 LLM 멀티오케스트레이션 서비스를 구축하고 운영중이다.
항상 고민중인게 이 파이프라인과 문맥관리인데,
이번에 클로드 코드의 관련 내용을 백엔드 서버개발자 관점에서 탈탈 털어보았다.

우선 위의 사진에처럼, 질문별로 문서를 작성하고, 관련해서 꼬리에 꼬리를 무는 질문을하여
문서를 구체화 한 뒤에, 블로그에 작성할 내용을 정리하도록 하였다.

---

LLM 에이전트 맥락관리와 Tool Orchestration: FastAPI + Python 실전 가이드

최근 에이전트 시스템을 운영해보면, 성능 이슈의 대부분은 모델 자체보다 두 가지에서 터집니다.

• 맥락관리(Context Management)
• 도구 실행 제어(Tool Orchestration)

이 글은 실제 CLI 에이전트 구조를 바탕으로, 왜 이 두 축이 중요한지와 FastAPI + Python으로 어떻게 최소 구현할 수 있는지 정리합니다.

빠른 네비게이션

• 문제 정의: 왜 이 주제가 중요한가
• 설계 핵심: 설계 원칙 5가지, 아키텍처 한 장 요약
• 동작 예시: 구체 시나리오 1~4
• 구현 예시: FastAPI 미니 예제 1~3
• 운영 가이드: Prompt Caching, 운영 체크리스트, 전체 흐름 순서도

3줄 요약

• 맥락관리의 핵심은 "기록을 늘리는 것"이 아니라 "핵심 상태를 남기며 압축하는 것"입니다.
• Tool Orchestration의 핵심은 "많이 병렬화"가 아니라 "안전한 툴만 병렬화"입니다.
• Prompt Caching의 핵심은 "모든 블록 캐시"가 아니라 "안정된 경계 1개"입니다.

왜 이 주제가 중요한가

모델은 잘 대답하는데 서비스는 불안정한 이유는 보통 아래 패턴입니다.

• 메시지 히스토리가 계속 누적되어 컨텍스트 윈도우를 압박한다.
• 툴 호출이 병렬/직렬 기준 없이 섞여 레이스 컨디션이 난다.
• 권한 확인 없이 위험한 툴이 즉시 실행된다.
• 캐시 경계가 불안정해서 요청 비용이 매 턴 급증한다.

핵심은 간단합니다.

• 토큰은 예산이고,
• 툴은 상태를 바꾸는 트랜잭션이며,
• 캐시는 경계가 안정적일 때만 이득이 납니다.

설계 원칙 5가지

1. 최근 대화만 남기는 것이 아니라, 중요한 상태를 요약본으로 보존한다.
2. 툴은 무조건 병렬이 아니라, 안전한 툴만 병렬로 돌린다.
3. 위험한 툴은 권한 게이트를 통과해야 실행한다.
4. 캐시 마커는 많이 붙이는 것이 아니라, 안정된 경계 한 곳에 둔다.
5. 실패(PTL, timeout, permission deny)를 정상 경로로 취급하고 재시도 정책을 둔다.

아키텍처 한 장 요약

• API 레이어: 모델 호출, 스트리밍 처리, usage 수집
• Context 레이어: 히스토리 누적, 압축(snip/microcompact/autocompact), 첨부 재주입
• Orchestration 레이어: tool_use 파싱, 병렬/직렬 실행, 결과 병합
• Safety 레이어: 권한 정책, 사용자 승인, 위험도 분류
• Observability 레이어: cache hit/miss, 토큰 사용량, 실패 카운트

구체 시나리오 1: "src/auth.py 고쳐줘" 요청이 들어오면

아래는 실제 운영에서 가장 자주 보는 1턴 내부 흐름입니다.

[1] UserMessage("src/auth.py 고쳐줘") append
        -> 모델 호출

[2] AssistantMessage(text + tool_use: read_file)
        stop_reason = tool_use

[3] 런타임이 read_file 실행
        -> UserMessage(tool_result: 파일 내용)
        -> 같은 턴에서 모델 재호출

[4] AssistantMessage(text + tool_use: write_file)
        stop_reason = tool_use

[5] 권한 게이트 ask 발생 (write_file은 위험도 중간 이상)
        -> 사용자 승인 후 실행
        -> UserMessage(tool_result: 수정 완료)

[6] AssistantMessage("수정 완료")
        stop_reason = end_turn

핵심:

• tool_use가 나오면 턴이 끝나는 게 아니라 follow-up 루프가 이어집니다.
• tool_result는 user role로 다시 들어가 다음 모델 호출 컨텍스트가 됩니다.

구체 시나리오 2: 스트리밍 이벤트가 메시지로 조립되는 방식

아래처럼 네트워크 이벤트(델타)를 받아 상위 메시지로 조립합니다.

message_start
-> content_block_start(text)
-> content_block_delta(text_delta: "파일을")
-> content_block_delta(text_delta: " 확인할게요")
-> content_block_stop
    => AssistantMessage(text) 방출

-> content_block_start(tool_use)
-> content_block_delta(input_json_delta: '{"file_path":"src/auth.py"')
-> content_block_delta(input_json_delta: '}')
-> content_block_stop
    => AssistantMessage(tool_use) 방출

-> message_delta(stop_reason: tool_use, usage: ...)
    => 직전 메시지 메타 확정(usage/stop_reason)

핵심:

• 사용자 체감 출력은 빠르게 먼저 내보내고,
• usage/stop_reason은 message_delta에서 최종 보강됩니다.

구체 시나리오 3: 병렬/직렬 오케스트레이션 결과 비교

요청된 tool plan이 아래와 같다고 가정합니다.

{
    "calls": [
        {"name": "read_file", "args": {"path": "src/auth.py"}},
        {"name": "search_code", "args": {"query": "login"}},
        {"name": "write_file", "args": {"path": "src/auth.py", "patch": "..."}}
    ]
}

실행 정책:

• read_file + search_code: 병렬 배치
• write_file: 직렬 단독 배치

결과적으로 읽기/검색은 빠르게 합쳐지고, 상태 변경은 안전하게 뒤에서 실행됩니다.

구체 시나리오 4: 캐시 마커는 "많이"가 아니라 "안정적으로 1개"

잘못된 예시(경계를 여러 군데 흔듦):

messages.map(m => ({
    ...m,
    content: m.content.map(c => ({
        ...c,
        cache_control: { type: 'ephemeral', ttl: '1h' },
    })),
}))

권장 예시(요청당 message-level marker 1개):

const markerIndex = skipCacheWrite ? messages.length - 2 : messages.length - 1
const payload = messages.map((m, i) => toMessageParam(m, i === markerIndex))

핵심:

• cache_control은 캐시 데이터 본문이 아니라 경계 힌트입니다.
• 경계가 자주 흔들리면 cache hit rate가 급락하고 비용/지연이 증가합니다.

FastAPI 예제 실행 방법 (로컬)

아래 예제들은 각각 독립 파일로 저장해서 실행할 수 있습니다.

1. 가상환경 및 패키지 설치

python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn pydantic

2. 예제 파일 저장

• example_context.py
• example_orchestrator.py
• example_guard.py

3. 서버 실행

uvicorn example_context:app --reload --port 8000
uvicorn example_orchestrator:app --reload --port 8001
uvicorn example_guard:app --reload --port 8002

4. 빠른 테스트

curl -X POST http://127.0.0.1:8000/chat \
    -H 'Content-Type: application/json' \
    -d '{"session_id":"s1","user_input":"문서 20개 요약해줘"}'

FastAPI 미니 예제 1: 맥락관리 파이프라인

아래 예제는 턴마다 히스토리를 누적하고, 임계값을 넘으면 자동 요약(compact)을 수행합니다.

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Dict

app = FastAPI()


class Turn(BaseModel):
    role: str
    content: str


class ChatRequest(BaseModel):
    session_id: str
    user_input: str


SESSIONS: Dict[str, List[Turn]] = {}

TOKEN_LIMIT = 12000
AUTO_COMPACT_THRESHOLD = 9000


def rough_token_count(turns: List[Turn]) -> int:
    return sum(len(t.content) // 3 for t in turns)


def compact_history(turns: List[Turn]) -> List[Turn]:
    if len(turns) < 6:
        return turns

    head = turns[:-4]
    tail = turns[-4:]

    summary_text = " ".join(t.content for t in head)[:1200]
    summary = Turn(role="system", content=f"Summary of earlier context: {summary_text}")
    return [summary] + tail


@app.post("/chat")
def chat(req: ChatRequest):
    turns = SESSIONS.setdefault(req.session_id, [])

    turns.append(Turn(role="user", content=req.user_input))

    token_est = rough_token_count(turns)
    if token_est > AUTO_COMPACT_THRESHOLD:
        turns[:] = compact_history(turns)

    # 여기서 실제 LLM 호출을 수행한다고 가정
    answer = f"Echo: {req.user_input}"
    turns.append(Turn(role="assistant", content=answer))

    if rough_token_count(turns) > TOKEN_LIMIT:
        return {
            "ok": False,
            "reason": "context_window_exceeded",
            "hint": "compact 강도를 높이거나 입력 배치를 분할하세요",
        }

    return {
        "ok": True,
        "answer": answer,
        "token_estimate": rough_token_count(turns),
        "turns": len(turns),
    }

포인트:

• 자동 압축을 늦게 걸면 이미 PTL 근처까지 가서 실패 확률이 커집니다.
• 압축 후에도 첨부/파일 컨텍스트를 선택적으로 재주입하는 단계가 필요합니다.

FastAPI 미니 예제 2: Tool Orchestration (병렬/직렬 분리)

도구는 모두 병렬 실행하면 빨라 보이지만, 상태 변경 툴이 섞이면 깨집니다.

import asyncio
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Any, Dict, List

app = FastAPI()


class ToolCall(BaseModel):
    name: str
    args: Dict[str, Any]


class ToolPlan(BaseModel):
    calls: List[ToolCall]


TOOL_META = {
    "read_file": {"concurrency_safe": True},
    "search_code": {"concurrency_safe": True},
    "run_terminal": {"concurrency_safe": False},
    "write_file": {"concurrency_safe": False},
}


async def execute_tool(call: ToolCall) -> Dict[str, Any]:
    # 실제 구현에서는 도구별 dispatcher 연결
    await asyncio.sleep(0.05)
    return {"tool": call.name, "ok": True, "result": call.args}


@app.post("/orchestrate")
async def orchestrate(plan: ToolPlan):
    safe_calls = [
        c for c in plan.calls if TOOL_META.get(c.name, {}).get("concurrency_safe", False)
    ]
    unsafe_calls = [c for c in plan.calls if c not in safe_calls]

    safe_results = await asyncio.gather(*(execute_tool(c) for c in safe_calls))

    unsafe_results = []
    for c in unsafe_calls:
        unsafe_results.append(await execute_tool(c))

    return {
        "parallel_count": len(safe_calls),
        "serial_count": len(unsafe_calls),
        "results": safe_results + unsafe_results,
    }

포인트:

• 읽기 전용/조회성 툴은 병렬, 상태 변경 툴은 직렬이 기본값입니다.
• 이 분리만 해도 툴 충돌과 재현 어려운 버그가 크게 줄어듭니다.

FastAPI 미니 예제 3: 권한 게이트(Approval Loop)

실무에서는 자동화보다 안전이 우선인 경우가 많습니다.

from enum import Enum
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()


class Decision(str, Enum):
    allow = "allow"
    deny = "deny"
    ask = "ask"


class GuardRequest(BaseModel):
    tool_name: str
    risk: str
    user_approved: bool = False


def policy(tool_name: str, risk: str) -> Decision:
    if risk == "low":
        return Decision.allow
    if tool_name in {"write_file", "run_terminal"}:
        return Decision.ask
    return Decision.deny


@app.post("/guard")
def guard(req: GuardRequest):
    d = policy(req.tool_name, req.risk)

    if d == Decision.allow:
        return {"decision": "allow"}

    if d == Decision.deny:
        return {"decision": "deny", "reason": "policy_blocked"}

    if req.user_approved:
        return {"decision": "allow", "reason": "user_approved"}

    raise HTTPException(
        status_code=409,
        detail={
            "decision": "ask",
            "reason": "need_user_approval",
        },
    )

포인트:

• allow/deny/ask의 3상태 모델이 운영 난이도를 크게 낮춥니다.
• ask 상태를 UI 승인 루프로 연결하면 안전성과 자동화를 함께 가져갈 수 있습니다.

Prompt Caching은 어떻게 봐야 하나

캐싱 관련 오해를 한 줄로 정리하면:

• cache_control은 캐시 경계 마커이지, 캐시 데이터 본문이 아닙니다.

즉 클라이언트는 평소처럼 메시지를 보내고, 서버가 캐시를 저장/재사용합니다.
그래서 중요한 건 다음입니다.

• 경계를 안정적으로 유지할 것
• 시스템 프롬프트/툴 스키마/베타 헤더를 자주 흔들지 말 것
• hit rate는 cache read 토큰과 응답 지연으로 관측할 것

운영 체크리스트

1. 압축 트리거 임계값이 PTL 전에 작동하는가
2. 병렬 툴과 직렬 툴의 기준이 명시되어 있는가
3. 위험 툴에 대해 사용자 승인 루프가 있는가
4. 캐시 경계를 턴마다 바꾸고 있지 않은가
5. 실패 시 재시도와 circuit breaker가 있는가
6. 비용/지연/실패 지표를 대시보드에서 보는가

마무리

에이전트 품질은 모델 선택보다 운영 구조에서 갈립니다.

• 맥락관리: 오래된 정보를 버리는 기술이 아니라, 중요한 상태를 보존하는 기술
• 툴 오케스트레이션: 많이 실행하는 기술이 아니라, 안전하게 실행 순서를 제어하는 기술

이 두 축만 제대로 잡아도, 같은 모델로도 체감 품질과 비용 효율이 크게 달라집니다.

실제 운영에서 가장 효과가 큰 개선 순서는 보통 아래와 같습니다.

1. PTL 전 자동 압축 임계값 조정
2. concurrency-safe 기준 정리(병렬/직렬 분리)
3. 위험 툴 approval loop 의무화
4. 캐시 경계 안정화(마커 정책 고정)

즉, 모델 교체보다 먼저 오케스트레이션과 맥락정책을 고정하면, 실패율과 비용이 함께 내려갑니다.

전체 흐름 순서도

[User Input]
    -> messages에 UserMessage append
    -> 토큰 추정 / 임계값 체크
            -> (초과) compact 실행: snip/microcompact/autocompact
            -> 모델 호출

모델 응답 처리:
    -> Assistant 응답에 tool_use 존재?
            -> No: stop_reason=end_turn -> 턴 종료 / 다음 턴 대기
            -> Yes:
                    -> Tool plan 생성
                    -> concurrency_safe 기준으로 병렬/직렬 분리
                    -> 권한 정책 분기
                            -> allow: tool 실행
                            -> deny: 에러 tool_result 생성
                -> ask: 사용자 승인 대기
                    -> (백그라운드) classifier/hook 판정 작업 진행
                    -> 승인 시 실행 / 거절 시 에러 tool_result
            -> tool 실행 결과를 UserMessage(tool_result)로 append
            -> 같은 턴에서 모델 재호출
            -> tool_use가 사라질 때까지 반복

캐시 처리:
    -> cache_control 경계 마커 적용
    -> 요청당 message-level marker 1개 유지
    -> TTL 정책 결정(사용자 eligibility + querySource allowlist)
    -> skipCacheWrite면 marker를 second-to-last로 이동
    -> 서버측 prompt cache read/write

백그라운드 실행 처리:
    -> 장시간 작업인가?
        -> Yes: run_in_background 경로로 전환, task id 반환, 턴은 계속 진행
        -> No: foreground 실행 후 즉시 tool_result 반환

캐시정책/백그라운드/동의 루프 운영 포인트

1. 캐시 정책

• cache_control은 캐시 데이터 본문이 아니라 경계 힌트입니다.
• 마커를 여러 개 두기보다, 요청당 1개를 안정적으로 유지해야 hit rate가 높습니다.
• TTL 1h는 항상 적용이 아니라 eligibility + allowlist 조건을 통과해야 적용됩니다.

2. 동의(ask) 루프

• ask 상태는 실행 전 대기 상태입니다.
• 대기 중에는 정책 판정 보조 작업(classifier/hook)을 백그라운드로 돌려 응답 시간을 줄일 수 있습니다.
• 승인되면 동일 tool call을 이어서 실행하고, 거절되면 에러 tool_result를 user role로 반환합니다.

3. 백그라운드 실행

• 백그라운드 전환은 권한 대기와 다릅니다.
• 권한 대기는 "아직 실행 전", 백그라운드는 "실행은 시작했고 분리 실행"입니다.
• 장시간 명령은 task id 기반으로 추적하고, 후속 턴에서 상태/로그를 조회하도록 설계하는 것이 안전합니다.