Agentic Loop Hook

Agentic Loop Hook

모델 응답을 가로채고, 서버 측에서 도구 호출을 처리한 뒤 모델을 다시 실행하는 CustomLogger 콜백을 만드는 방법을 알아봐요. 호출자에게는 모두 투명하게 처리돼요.

출처: 문서

본문

지원 호출 유형:

  • async 전용 (동기 호출은 훅을 트리거하지 않음)
  • 비스트리밍 전용 (스트리밍 응답은 도구 호출을 검사할 수 없음)
  • /v1/messages/v1/chat/completions 모두에서 동작

콜백 구현

CustomLogger의 두 메서드를 오버라이드해요:

from litellm.integrations.custom_logger import CustomLogger
from litellm.types.integrations.custom_logger import AgenticLoopPlan, AgenticLoopRequestPatch

MY_TOOL = "my_tool"

class MyToolCallback(CustomLogger):

    async def async_should_run_agentic_loop(
        self, response, model, messages, tools, stream, custom_llm_provider, kwargs
    ):
        # Return (True, context_dict) if there are tool calls to handle
        content = getattr(response, "content", None) or []
        calls = [b for b in content if isinstance(b, dict)
                 and b.get("type") == "tool_use" and b.get("name") == MY_TOOL]
        if not calls:
            return False, {}
        return True, {"tool_calls": calls}

    async def async_build_agentic_loop_plan(
        self, tools, model, messages, response,
        anthropic_messages_provider_config,
        anthropic_messages_optional_request_params,
        logging_obj, stream, kwargs,
    ):
        calls = tools["tool_calls"]
        results = [f"result for {c['input']}" for c in calls]  # your logic here

        follow_up = messages + [
            {"role": "assistant", "content": [
                {"type": "tool_use", "id": c["id"], "name": c["name"], "input": c["input"]}
                for c in calls
            ]},
            {"role": "user", "content": [
                {"type": "tool_result", "tool_use_id": c["id"], "content": results[i]}
                for i, c in enumerate(calls)
            ]},
        ]
        return AgenticLoopPlan(
            run_agentic_loop=True,
            request_patch=AgenticLoopRequestPatch(messages=follow_up),
        )

/v1/chat/completions의 경우 대신 async_build_chat_completion_agentic_loop_plan을 오버라이드해요. 같은 아이디어이며, anthropic_messages_optional_request_params 대신 optional_params를 사용해요.

등록

import litellm
litellm.callbacks = [MyToolCallback()]

또는 config.yaml에서, 모듈이 만든 인스턴스를 가리켜요:

# my_module.py, after the class definition above
my_tool_callback = MyToolCallback()
litellm_settings:
  callbacks: ["my_module.my_tool_callback"]

warning: 점 경로는 인스턴스를 지칭해야 하며 클래스가 아니에요. callbacks: ["my_module.MyToolCallback"]는 config 로드를 실패시키고, 그 검사 이전 버전에서는 깨끗하게 시작하고 훅을 절대 실행하지 않았어요.

AgenticLoopPlan 필드

필드 효과
run_agentic_loop=True + request_patch 패치된 요청으로 모델을 다시 실행
response_override 이 값을 호출자에게 직접 반환 (재실행 없음)
terminate=True 루프를 멈추고 현재 응답 반환
run_agentic_loop=False (기본값) 건너뛰기. 다음 콜백이 검사됨

AgenticLoopRequestPatchmodel, messages, tools, max_tokens, optional_params, kwargs를 받아요.

루프 안전

  • 기본 최대 재실행: 3, 요청별로 kwargs["max_agentic_loops"]로 오버라이드
  • 동일한 도구 호출 지문은 루프를 자동으로 중단
  • 현재 깊이는 kwargs["_agentic_loop_depth"]에 있음

이 저장소의 예시

  • litellm/integrations/compression_interception/handler.py
  • litellm/integrations/websearch_interception/handler.py

더 알아보기 (Learn more)

  • LiteLLM 커스텀 로거
  • 사용자 정의 플러그인 만들기