2025년 6월 3일 화요일

Gemma3 기반 Ollama 활용 AI 에이전트 개발 핵심 Function Call 구현해보기

이 글은 AI 에이전트(Agent) 개발 시 필수적인 함수호출 방법을 오픈소스를 이용해 구현해 본다. 이를 위해, Gemma3(젬마) LLM(Large Language Model) 기반 Ollama 활용 Function Call(펑션콜) 실습 내용을 소개하고 실행 결과를 확인한다. 아울러, 이런 함수호출 방식의 한계점을 개선하기 위한 솔류션을 나눔한다. 이 실습의 결과는 다음과 같다. 


이 글은 다음 내용을 포함한다.
  • AI 에이전트 구현을 위한 함수 호출 방법
  • Ollama 를 통한 Gemma3 사용법
  • 채팅 형식 프롬프트 및 메모리 사용법
  • Gradio 기반 웹 앱 개발
  • Function call 의 한계와 솔류션
AI 에이전트 내부 Function call 메커니즘(Akriti, 2025)

이 글의 구현 코드는 다음 링크에서 확인할 수 있다.

Gemma 3는 구글이 개발해  2025년 3월 10일에 출시한 LLM으로, 차세대 경량 오픈 멀티모달 AI 모델로, 텍스트와 이미지를 동시에 처리할 수 있는 기능을 갖추고 있다. 이 모델은 다양한 크기와 사양으로 제공되어 단일 GPU 또는 TPU 환경에서도 실행 가능하다.

Gemma 3는 1B, 4B, 12B, 27B의 네 가지 모델 크기로 제공되며, 각각 10억, 40억, 120억, 270억 개의 파라미터를 갖추고 있다. 1B 모델은 텍스트 전용으로 32K 토큰의 입력 컨텍스트를 지원하고, 4B, 12B, 27B 모델은 멀티모달 기능을 지원하며 128K 토큰의 입력 컨텍스트를 처리할 수 있다. 이는 이전 Gemma 모델보다 16배 확장된 크기로, 훨씬 더 많은 양의 정보를 한 번에 처리할 수 있게 해준다.

이 모델은 텍스트와 이미지 데이터를 동시에 처리하고 이해하는 멀티모달 기능을 제공한다. 이미지 해석, 객체 인식, 시각적 질의응답 등 다양한 작업을 수행할 수 있으며, 텍스트 기반 작업에 시각적 정보를 효과적으로 활용할 수 있도록 지원한다. 


  
Welcome Gemma 3: Google's all new multimodal, multilingual, long context open LLM

Gemma 3는 140개 이상의 언어를 지원하여 전 세계 다양한 언어 사용자를 대상으로 하는 AI 애플리케이션 개발에 매우 유리하다. 사용자는 자신의 모국어로 Gemma 3와 상호작용할 수 있으며, 다국어 기반의 텍스트 분석 및 생성 작업도 효율적으로 수행할 수 있다.

이 모델은 다양한 작업 수행 능력을 갖추고 있다. 질문 답변, 텍스트 요약, 논리적 추론, 창의적인 텍스트 형식 생성(시, 스크립트, 코드, 마케팅 문구, 이메일 초안 등), 이미지 데이터 분석 및 추출 등 광범위한 자연어 처리 및 컴퓨터 비전 관련 작업을 수행할 수 있다. 또한, 함수 호출 및 구조화된 출력을 지원하여 개발자들이 특정 작업을 자동화하고 에이전트 기반의 경험을 구축하는 데 도움을 준다.

Gemma 3는 다양한 도구 및 프레임워크와 원활하게 통합된다. Hugging Face Transformers, Ollama, JAX, Keras, PyTorch, Google AI Edge, UnSloth, vLLM, Gemma.cpp 등 다양한 개발 도구 및 프레임워크와 호환되어 개발자들이 자신이 익숙한 환경에서 Gemma 3를 쉽게 활용하고 실험할 수 있다.

이 모델은 다양한 벤치마크 테스트에서 동급 모델 대비 최첨단 성능을 입증했다. 특히, Chatbot Arena Elo Score에서 1338점을 기록하며, 여러 오픈 소스 및 상용 모델보다 높은 성능을 보였다. 

Gemma 3는 오픈 모델로, 개방형 가중치를 제공하여 사용자가 자유롭게 조정하고 배포할 수 있다. Kaggle과 Hugging Face에서 다운로드 가능하며, Creative Commons 및 Apache 2.0 라이선스를 따름으로써, 개발자와 연구자에게 VLM 기술에 대한 접근성을 높여준다.

개발 환경
개발 환경은 다음과 같다. 미리 설치, 가입한다.
설치되어 있다면, 다음 명령을 터미널에서 실행한다.
ollama pull gemma3:4b
gemma3:4b GPU VRAM 소모량

이제 다음과 같이 모델을 실행해 볼 수 있다. 

참고로, GPU VRAM 등을 고려해 더 성능이 좋은 파라메터수 많은 대형 모델을 사용할 수도 있다.
gemma3 지원 모델들

처리 프로세스
이 실습 프로그램의 프로세스는 다음과 같다.

Gradio 앱이 시작되면, 사용자의 입력이 발생하고 이 입력은 process_message 함수에 전달된다. 이 함수는 사용자의 메시지를 chat_history에 추가하여 대화 기록을 저장한다. 이후 모델에게 전달할 대화 문맥을 구성하기 위해 messages 리스트가 생성된다.

그 다음 단계에서는 ollama.chat 함수를 통해 언어 모델에게 응답을 요청하게 되며, 이 응답 내에 함수 호출이 포함되어 있는지를 확인한다. 만약 응답에 함수 호출이 포함되어 있다면, 이를 parse_function_call 함수를 통해 파싱한다.

파싱된 함수가 google_search라면, 모델이 검색을 원한다고 판단하여 검색 쿼리를 추출하고 검색 수행 예정임을 사용자에게 안내하는 메시지를 추가한다. 이후 실제로 google_search 함수를 실행하여 외부 검색을 수행한다.

검색 결과는 다시 chat_history에 저장되며, 이 결과를 바탕으로 언어 모델에게 재질문을 하여 더 정확하고 완성된 응답을 유도한다. 모델이 생성한 최종 응답은 chat_history에 마지막으로 추가되고, 이 전체 대화 기록이 사용자에게 반환된다.

이 구조는 사용자의 질의에 따라 외부 정보까지 능동적으로 검색하고 반영할 수 있는 LLM 기반 AI 에이전트의 대표적인 흐름을 보여준다.

다음은 이 순서도를 보여준다.

구현하기
터미널에서 다음 라이브러리를 설치한다.
pip install langchain-core langchain-openai gradio ollama requests python-dotenv pydantic

새로운 파이썬 파일(코드 참고)을 생성한 후, 우선, 필요한 라이브러리를 임포트한다. 
import gradio as gr
import ollama
import requests, json, os
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from typing import Optional, Dict, Any, List

load_dotenv()
SERPER_API_KEY = os.getenv('SERPER_API_KEY')

그리고, 사용하는 API 키를 가져온다. 이를 위해, 미리 .env 파일을 다음과 같이 만들어 놓고, 해당 API를 입력해 놓야야 한다.
# .env
SERPER_API_KEY=<YOUR API KEY>


파라메터에서 검색 질의문, 함수호출명과 파라메터를 정의한다. 아울러, 질의 결과를 명확히 데이터항목으로 추출하기 위해서 검색 결과가 될 데이타항목(타이틀, 링크, 스닙펫) 형식을 pydantic의 basemodel을 이용해 명확히 정의한다. 그리고, LLM 호출 결과를 펑션콜이 가능한 형식으로 변환하기 위한 파싱 함수인 parse_function_call 을 정의한다.
class SearchParameters(BaseModel):
    query: str = Field(..., description="Search term to look up")

class FunctionCall(BaseModel):
    name: str
    parameters: Dict[str, Any]

class SearchResult(BaseModel):
    title: str
    link: str
    snippet: str

    def to_string(self) -> str:
        return f"Title: {self.title}\nLink: {self.link}\nSnippet: {self.snippet}"

def google_search(query: str) -> SearchResult:
    """Perform a Google search using Serper.dev API"""
    try:
        url = "https://google.serper.dev/search"
        payload = json.dumps({"q": query})
        headers = {
            'X-API-KEY': SERPER_API_KEY,
            'Content-Type': 'application/json'
        }
       
        response = requests.post(url, headers=headers, data=payload)
        response.raise_for_status()  # 잘못된 상태 코드에 대해 예외 발생
       
        results = response.json()
       
        if not results.get('organic'):
            raise ValueError("No search results found.")
           
        first_result = results['organic'][0]
        return SearchResult(
            title=first_result.get('title', 'No title'),
            link=first_result.get('link', 'No link'),
            snippet=first_result.get('snippet', 'No snippet available.')
        )
    except Exception as e:
        print(f"Search error: {str(e)}")
        raise

def parse_function_call(response: str) -> Optional[FunctionCall]:
    """Parse the model's response to extract function calls"""
    try:
        # Clean the response and find JSON structure
        response = response.strip()
        start_idx = response.find('{')
        end_idx = response.rfind('}') + 1
       
        if start_idx == -1 or end_idx == 0:
            return None
           
        json_str = response[start_idx:end_idx]
        data = json.loads(json_str)
        return FunctionCall(**data)
    except Exception as e:
        print(f"Error parsing function call: {str(e)}")
        return None


gemma에 지시할 시스템 프롬프트 명령을 정의한다. prompt_system_message는 이 챗봇이 어떻게 동작해야 하는지, 그리고 어떤 기준으로 답변을 해야 하는지에 대한 지침을 제공하는 역할을 한다. 이 메시지는 챗봇이 2024년까지의 정보를 학습한 AI 어시스턴트임을 명확히 하고, 사용자의 질문에 대해 가능한 경우에는 바로 답변을 하되, 최신 정보나 불확실한 내용, 시의성이 있는 질문에 대해서는 반드시 펑션콜을 통해 검색 기능을 활용해야 함을 명시한다. 이전 대화 내용이 함께 입력으로 주어지기 때문에, 챗봇은 이 대화 맥락을 참고하여 일관성 있고 상황에 맞는 답변을 해야 한다고 안내한다. 참고로, 준수해야 할 gemma3의 function call 형식은 다음과 같다. 
검색이 필요한 상황과 그렇지 않은 상황을 구체적으로 구분하여, 챗봇이 임의로 정보를 추정하거나 추가하지 않고, 검색 결과에 기반한 사실만을 간결하게 전달하도록 유도한다. 검색이 필요한 경우에는 정해진 JSON 형식으로만 응답하도록 하여, 시스템이 함수 호출 방식으로 검색을 처리할 수 있게 한다.
# 프롬프트 시스템 메세지 정의
prompt_system_message = """You are an AI assistant with training data up to 2024. Answer questions directly when possible, and use search when necessary.

You will receive previous conversation messages as part of the input. Use these prior messages to maintain context and provide coherent, context-aware answers.

DECISION PROCESS:
1. For historical events before 2024:
   - Answer directly from your training data.
2. For events in 2024:
   - If you are certain, answer directly.
   - If you are unsure, use search.
3. For events after 2024 or current/recent information:
   - Always use search.
4. For timeless information (scientific facts, concepts, etc.):
   - Answer directly from your training data.

ALWAYS USE SEARCH if the question:
- Contains words like "current", "latest", "now", "present", "today", "recent"
- Asks about someone in a changing position (champion, president, CEO, etc.)
- Requests information that might have changed since 2024
- Is time-sensitive and does not specify a time period

FUNCTION CALL FORMAT:
When you need to search, respond WITH ONLY THE JSON OBJECT, no other text, no backticks:
{
    "name": "google_search",
    "parameters": {
        "query": "your search query"
    }
}

SEARCH FUNCTION:
{
    "name": "google_search",
    "description": "Search for real-time information",
    "parameters": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "Search term"
            }
        },
        "required": ["query"]
    }
}

WHEN ANSWERING BASED ON SEARCH RESULTS:
- Use ONLY facts found in the search results below.
- Do NOT add any dates or information not present in the search results.
- Do NOT make assumptions about timing or events.
- Quote dates exactly as they appear in the results.
- Keep your answer concise and factual.
"""


gemma에 전달할 메시지는 프롬프트 지시문, 사용자 질문을 포함한 이전 채팅 이력 메시지 등을 모두 포함한다. 이를 ollama LLM 에 전달할 수 있는 형식으로 변환하는 함수를 다음과 같이 준비한다. 
# 메시지 리스트를 생성하는 함수
def filter_memory(memory):
    """assistant의 검색 안내 메시지를 memory에서 제외"""
    return [
        msg for msg in memory
        if not (
            msg["role"] == "assistant" and (
                msg["content"].startswith("Searching for:") or
                msg["content"].startswith("Searched for:")
            )
        )
    ]

def build_messages(chat_history, user_input=None, prompt_system_message=prompt_system_message, N=6, search_result=None):
    """
    최근 N개 메시지와 system 메시지를 합쳐 messages 리스트를 만듭니다.
    search_result가 있으면, user_input 대신 검색 결과 기반 프롬프트를 추가합니다.
    """
    memory = chat_history[-N:] if len(chat_history) > N else chat_history[:-1]
    filtered_memory = filter_memory(memory)
    messages = [{"role": "system", "content": prompt_system_message}] + filtered_memory
    if search_result is not None:
        messages.append({
            "role": "user",
            "content": (
                "Refer to the following search result and provide a concise, factual answer based only on this information:\n"
                f"{search_result.to_string()}"
            )
        })
    elif user_input is not None:
        messages.append({"role": "user", "content": user_input})
    return messages


이제 process_message 함수를 구현한다. 이 함수는 사용자의 입력과 기존 채팅 기록을 받아 AI 모델과의 대화 흐름을 관리하는 역할을 한다.

먼저 사용자의 메시지를 채팅 기록에 추가하고, 이전 대화 내용(메모리)을 추출하여 시스템 메시지와 함께 모델에 전달할 메시지 목록을 구성한다. 이 메시지 목록을 Ollama 모델에 전달하여 응답을 받는다. 모델의 응답이 함수 호출(JSON) 형태라면, 그 내용을 파싱하여 검색이 필요한 경우 검색 쿼리를 추출한다.

검색이 필요하다고 판단되면, 검색 중임을 알리는 메시지를 채팅 기록에 추가하고, 실제로 검색을 수행한다. 검색 결과를 다시 채팅 기록에 반영한 뒤, 이 결과를 포함한 새로운 메시지 목록을 만들어 모델에 전달하여 최종 답변을 받는다. 최종적으로 받은 답변 역시 채팅 기록에 추가한다.

검색이 필요하지 않은 경우에는 모델의 응답을 바로 채팅 기록에 추가한다. 이 과정에서 각 단계별로 최신 채팅 기록을 반환하여, 사용자 인터페이스가 실시간으로 대화 상태를 갱신할 수 있도록 한다.
함수 실행 중 오류가 발생하면, 오류 메시지를 채팅 기록에 추가하여 사용자에게 알린다.
# Model name
MODEL_NAME = "gemma3"

def process_message(user_input, chat_history):
    """Process user message and update chat history"""
    try:
        # 사용자 메시지를 기록에 추가
        chat_history.append({"role": "user", "content": user_input})
        search_info = None

        # 최근 N개 메시지만 memory에 포함 (예: 최근 6개)
        N = 6
        messages = build_messages(chat_history, user_input=user_input, N=N)

        # 모델로부터 응답 받기
        response = ollama.chat(
            model=MODEL_NAME,
            messages=messages      
        )
       
        model_response = response['message']['content']
       
        # 함수 호출로 응답을 파싱 시도
        function_call = parse_function_call(model_response)
       
        if function_call and function_call.name == "google_search":
            # 검색 파라미터 검증
            search_params = SearchParameters(**function_call.parameters)
            search_query = search_params.query
           
            # 검색 정보 기록에 추가
            search_info = f"Searching for: {search_query}"
            chat_history.append({"role": "assistant", "content": search_info})
            yield chat_history
           
            # 검색 실행
            search_result = google_search(search_query)
           
            # 검색 결과로 정보 업데이트
            search_info = f"Searched for: {search_query}\n\nResult:\n{search_result.to_string()}"
            chat_history[-1] = {"role": "assistant", "content": search_info}
            yield chat_history

            # 검색 결과 기반 메시지 생성
            messages = build_messages(chat_history, N=N, search_result=search_result)
     
            # 검색 결과를 포함해 모델로부터 최종 응답 받기
            final_response = ollama.chat(
                model=MODEL_NAME,
                messages=messages
            )
           
            assistant_response = final_response['message']['content']
        else:
            # 함수 호출이 없으면 직접 응답 반환
            assistant_response = model_response
       
        # 최종 응답을 기록에 업데이트
        if search_info:
            chat_history.append({"role": "assistant", "content": f" Response:\n{assistant_response}"})
        else:
            chat_history.append({"role": "assistant", "content": assistant_response})
       
        yield chat_history
           
    except Exception as e:
        error_msg = f"An error occurred: {str(e)}"
        chat_history.append({"role": "assistant", "content": error_msg})
        yield chat_history


이제 Gradio UI 를 정의하고, 메인 엔트리에서 이 앱을 실행한다.

# Gradio 인터페이스 생성
with gr.Blocks(css="footer {visibility: hidden}") as demo:
    gr.Markdown("""
    # Agent based on Gemma3 using Function Call
   

    """)
   
    chatbot = gr.Chatbot(
        height=500,
        show_label=False,
        avatar_images=(None, "https://api.dicebear.com/9.x/identicon/svg?seed=Mason"),
        type="messages"
    )
   
    with gr.Row():
        msg = gr.Textbox(
            scale=5,
            show_label=False,
            placeholder="Ask me anything...",
            container=False
        )
        submit_btn = gr.Button("Send", scale=1)
   
    with gr.Row():
        clear_btn = gr.Button("Clear Chat")
   

    # 이벤트 핸들러 설정
    msg.submit(
        process_message,
        [msg, chatbot],
        [chatbot],
    )
   
    submit_btn.click(
        process_message,
        [msg, chatbot],
        [chatbot],
    )
   
    clear_btn.click(
        lambda: [],
        None,
        chatbot,
        queue=False
    )
   
    # 메시지 전송 후 텍스트박스 비우기
    msg.submit(lambda: "", None, msg)
    submit_btn.click(lambda: "", None, msg)

if __name__ == "__main__":
    demo.launch(inbrowser=True, share=True)

실행
앞에 구현된 앱을 실행한다. 그리고, 적절한 질문을 입력해 본다. 다음과 같이 실행되면 성공한 것이다.

펑션콜 문제 개선 방법
실제로 질의해보면 불명확한 프롬프트 입력 등에서 부적절한 함수 호출이 수행되는 것을 알 수 있다. 이를 개선하기 위해 다음 사항을 고려한다.
  • 프롬프트 설계의 명확성
함수 호출이 필요한 상황, 호출 방식(JSON 포맷 등), 호출 예시를 SYSTEM_MESSAGE에 명확하게 안내해야 한다. 함수 호출이 아닌 일반 답변을 하면 안 된다는 점을 반복적으로 강조한다.
예시 프롬프트:
"질문에 답변하기 위해 함수 호출이 필요하다고 판단되면 반드시 아래 JSON 형식으로만 응답하라. 다른 텍스트나 설명은 절대 포함하지 마라."
  • 함수 정의의 구체성
함수의 목적, 파라미터, 반환값, 사용 예시를 상세하게 기술한다. 각 파라미터의 타입, 필수 여부, 설명을 명확히 한다. 함수가 처리할 수 없는 입력(예: 빈 문자열, 잘못된 타입 등)에 대한 예외 상황도 명시한다.
  • 예시 기반 Few-shot Prompting
SYSTEM_MESSAGE 또는 user message에 함수 호출이 필요한 질문과 그에 대한 올바른 함수 호출 예시를 여러 개 포함시킨다. 예시가 많을수록 모델이 패턴을 더 잘 학습한다.
  • 함수 호출 실패 시 재시도 로직
모델이 함수 호출을 하지 않거나 잘못된 형식으로 응답하면, 내부적으로 "함수 호출이 필요합니다. 반드시 JSON 형식으로만 응답하세요."와 같은 추가 프롬프트로 재요청한다.
  • 출력 파싱의 견고성
모델이 JSON 외의 텍스트를 섞어서 반환할 수 있으므로, 파싱 로직에서 JSON 부분만 추출하거나, 불완전한 JSON도 최대한 보완해서 파싱하도록 한다.
  • 함수 호출 의도 강화 프롬프트
SYSTEM_MESSAGE에 "함수 호출이 필요한 상황에서는 반드시 함수 호출을 우선적으로 고려하라"는 문구를 추가한다. "만약 함수 호출이 필요하지 않다고 판단되면, 그 이유를 설명하지 말고 바로 답변만 하라." 등 불필요한 설명을 억제한다.
  • 모델 버전 및 파라미터 최적화
최신 GPT-4 Turbo 등 함수 호출에 최적화된 모델을 사용한다. temperature, top_p 등 파라미터를 낮춰 일관된 응답을 유도한다. 
  • 함수 호출 실패 케이스 수집 및 개선 
실제 사용자 입력 중 함수 호출이 누락된 사례를 수집하여, SYSTEM_MESSAGE나 예시 프롬프트를 지속적으로 개선한다.

이외에 잘 활용되는 함수에 대한 파인튜닝을 수행해 본다. 

마무리
본 글은 ollama 를 이용한 gemma3 모델을 로딩해 Agent 개발 시 핵심이 되는 function call을 구현해 보았다. 실행해 보면 알겠지만, 펑션콜은 프롬프트 입력에 따라 민감하게 동작한다는 것을 알 수 있다. 그러므로, 함수 호출 방식은 적절히 LLM 오케스트레이션 및 튜닝되어야 한다는 것을 알 수 있다. 

레퍼런스

2025년 5월 17일 토요일

ReAct 에이전트 프레임웍 내부 코드 구조 개발해보기

ReAct(Reasoning and Acting) 에이전트 구조는 대형 언어 모델(LLM)을 기반으로 도구 호출 기능을 통합한 지능형 에이전트 시스템을 구현하는 설계 방식이다. 사용자의 지시를 추론하고 필요한 경우 외부 도구를 호출하여 목표를 달성하는 능력을 지니며, 최근 AutoGPT, BabyAGI, Manus 등 다양한 오픈소스 프로젝트에서 채택되고 있다. 이 글에서는  ReAct 에이전트 프레임웍 내부 코드 구조를 직접 개발해본다. 아울러, ReAct 에이전트의 문제점들을 살펴보고 솔류션을 생각해본다. 

ReAct 구조 

ReAct 시퀀스 처리 구조는 LLM 기반 에이전트가 사고(Think), 행동(Act), 관찰(Observe), 결론(Final)의 단계로 사용자 질의에 응답하는 체계적 프로세스이다. 이 구조는 복잡한 문제 해결을 위해 LLM이 도구와 상호작용하면서 점진적으로 정답을 유도해내도록 설계되었다.

다음은 해당 코드에서 구현된 ReAct 시퀀스의 흐름이다.

  1. Think
    LLM은 질문에 대한 분석을 수행하며 필요한 경우 어떤 도구를 사용할지를 판단한다. 이 단계는 주로 텍스트 상의 추론 내용으로 표현되며, 도구 실행 없이 논리 전개만 이루어진다.

  2. Act
    필요하다고 판단되면 Act: ToolName[Input] 또는 Act: ToolName(Input) 형식으로 도구 호출을 지시한다. 이 지시문은 정규표현식으로 감지되어 실제 도구 실행이 트리거된다.
  3. Observe
    호출된 도구가 실행되고, 그 결과는 Observe: 접두사를 붙여 기록된다. 이 결과는 이후 컨텍스트에 포함되어 다음 LLM 호출의 입력으로 사용된다.

  4. Final
    충분한 정보가 수집되면 Final Answer: 형식으로 최종 응답을 제공한다. 이는 루프를 종료시키는 조건이기도 하며, 에이전트의 실행 결과로 사용자에게 출력된다.

이러한 시퀀스를 통해 LLM은 단순히 한 번에 답변하는 것이 아니라, 도구를 여러 번 사용하고 그 결과를 반영하며 점진적으로 정확한 응답에 도달한다.

ReAct 에이전트 내부 코드 구현해보기

앞서 언급된 개념을 고려해 ReAct 에이전트의 핵심적인 코드만 구현해 본다. 다음과 같이 관련 라이브러리를 설치한다.

pip install langchain 

다음과 같이 코드를 구현한다. 

import re, os
from langchain.tools import Tool
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain.chat_models import ChatOpenAI
from langchain.schema import HumanMessage
from dotenv import load_dotenv

load_dotenv()

# Calculator function
def calculator(input_str: str) -> str:
    try:
        result = eval(input_str)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

def FireCrawlResults(max_results=3, search_type="web"):
    from langchain_community.tools.fire_crawl import FireCrawlResults
    return FireCrawlResults(max_results=max_results, search_type=search_type) # TBD

# Initialize tools
tools = [
    Tool(
        name="Calculator",
        func=calculator,
        description="Evaluates mathematical expressions. Input should be a valid Python expression."
    ),
    TavilySearchResults(max_results=3, search_type="web"),
    # FireCrawlResults(max_results=3, search_type="web"),
]

# Function to extract tool prototype info
def get_tools_info(tools):
    info_list = []
    for tool in tools:
        # Try to get function signature if possible
        if hasattr(tool, "func"):
            proto = f"{tool.name}(input: str)"
        else:
            proto = f"{tool.name}(input: str)"
        desc = getattr(tool, "description", "No description.")
        info_list.append(f"- {proto}: {desc}")
    return "\n".join(info_list)

# Initialize the language model
llm = ChatOpenAI(temperature=0, model="gpt-4", max_tokens=4000)
answer_validation_llm = ChatOpenAI(temperature=0, model="gpt-4-turbo", max_tokens=4000)

# Prompt template for ReAct
prompt_template = """
Instruction: {instruct}
IMPORTANT: If you do not know the answer, do not use 'Final Answer', just say 'I don't know'.

Tools you can use:
{tools}

Context:
{context}

Query:
{query}

You should follow the ReAct pattern:
- Think: Reason about the question or next step.
- Act: If needed, use a tool in the format Act: <tool>[<input>].
- Observe: Note the result of the action.
- Final: Give the final answer in the format Final Answer: <answer>.
"""

answer_validation_prompt = """
Question: {question}
Answer: {answer}
Is the answer relevant and correct for the question? Reply only with "yes" or "no" and a short reason.
"""

class ReActAgent:
    def __init__(self, llm, tools, prompt_template):
        self.llm = llm
        self.tools = {tool.name: tool for tool in tools}
        self.prompt_template = prompt_template
        self.tools_info = get_tools_info(tools)

    def validate_answer(self, question, answer):
        prompt = answer_validation_prompt.format(question=question, answer=answer)
        messages = [HumanMessage(content=prompt)]
        output = answer_validation_llm(messages)
        response = output.content.strip().lower()
        return response.startswith("yes"), response

    def run(self, input_query: str):
        history = []
        while True:
            prompt = self.prompt_template.format(
                instruct="Answer the question in detail using ReAct reasoning.",
                tools=self.tools_info,
                context="\n".join(history),
                query=input_query
            )
            messages = [HumanMessage(content=prompt)]
            output = self.llm(messages)
            response = output.content.strip()
            history.append(response)

            # Check for Final Answer
            final_match = re.search(r"Final Answer:\s*(.*)", response, re.IGNORECASE)
            if final_match:
                answer = final_match.group(1)
                # is_valid, validation_msg = self.validate_answer(input_query, answer) # TBD. 실제 react 코드 확인.
                # if is_valid:
                return answer

            # Detect action
            action_match = re.search(r"Act:\s*(\w+)[\[\(](.*)[\]\)]", response)
            if action_match:
                action_name = action_match.group(1)
                action_input = action_match.group(2)
                print(f"Action detected: {action_name} with input: {action_input}")

                tool = self.tools.get(action_name)
                if not tool:
                    history.append(f"Observe: Unknown tool: {action_name}")
                    continue
                action_input = action_input.replace("'", "").replace('"', "")  
                result = tool.run(action_input)
                history.append(f"Observe: {result}")
            else:
                # If no action and no final answer, just continue (could be Think step)
                continue

agent = ReActAgent(llm=llm, tools=tools, prompt_template=prompt_template)

def test():
    query = "What is the result of 12 * (3 + 4)?"
    result = agent.run(query)
    print(result)

    query = "What is the capital of France?"
    result = agent.run(query)
    print(result)

    query = "In web, What is Taewook kang's paper about Geo BIM using BIM and GIS?"
    result = agent.run(query)
    print(result)

def main():
    print("ReAct Agent is ready to use.")
    print("Available tools:")
    for tool in agent.tools_info.split("\n"):
        print(tool)

    # test()

    print("\nInteractive mode:")
    while True:
        query = input("Enter your query (or 'exit' to quit): ")
        if query.lower() == "exit":
            break
        result = agent.run(query)
        print(result)

if __name__ == "__main__":
    main()


이 코드는 ReAct 시퀀스를 중심으로 구성되어 있다. ReAct란 LLM이 단순한 질의응답을 넘어서서, 생각하고 도구를 사용하고 그 결과를 바탕으로 다시 추론한 뒤 최종적인 답을 내리는 일련의 추론 패턴이다. 이 흐름은 크게 네 단계로 나뉜다. 

첫째, Think 단계에서는 주어진 질문에 대해 어떻게 접근할지를 LLM이 서술적으로 사고한다. 둘째, Act 단계에서는 필요하다고 판단한 도구를 지정된 형식으로 호출한다. 이때 Act: ToolName[Input] 형식을 사용한다. 셋째, Observe 단계에서는 실행된 도구의 결과를 받아 LLM의 다음 추론에 포함될 수 있도록 기록한다. 넷째, Final 단계에서는 Final Answer: 답변 형식으로 최종 결론을 도출한다.

이러한 시퀀스를 실제로 구현하기 위해 코드에서는 특정 정규표현식을 사용하여 각 단계를 감지하고 분기 처리한다. 예를 들어, Act 단계는 Act:라는 키워드를 통해 감지되고, 대응되는 도구 이름과 입력값이 추출된다. 대응되는 도구가 존재하면 해당 함수나 객체의 run 메서드를 호출하여 결과를 얻고, 이를 Observe로 기록하여 다음 프롬프트 생성 시 문맥으로 전달한다. Final Answer가 감지되면 그 값을 최종 출력으로 반환하면서 루프를 종료한다.

코드는 먼저 환경변수를 불러오고, 수학 계산용 calculator 함수와 웹 검색용 Tavily 도구를 정의하고 tools 목록에 등록한다. 각 도구는 Tool 객체로 구성되며, name, func, description 속성을 갖는다. 이 도구들의 인터페이스 정보는 get_tools_info 함수를 통해 LLM 프롬프트에 전달할 수 있는 형식으로 변환된다.

ReActAgent 클래스는 핵심 로직을 담당한다. 생성자에서는 도구를 이름으로 접근 가능하도록 딕셔너리 형태로 구성하며, 프롬프트 템플릿에 포함할 도구 설명을 자동으로 생성한다. run 메서드는 LLM과의 대화를 이끄는 루프이다. 사용자의 입력 쿼리를 바탕으로 Think부터 Final 단계까지의 응답을 순차적으로 생성하고, 도구 호출이 필요한 경우 자동으로 감지하여 실행한다.

LLM은 ChatOpenAI 클래스를 통해 초기화되며, 주 모델과 검증용 모델이 따로 구성된다. 현재 코드는 검증 부분이 주석 처리되어 있어 정답 검증은 수행되지 않지만 validate_answer 메서드는 존재한다. 해당 메서드는 question과 answer를 받아 프롬프트에 삽입하고 gpt-4-turbo 모델로부터 응답을 받아 yes 또는 no 여부와 사유를 반환한다.

main 함수는 인터랙티브한 사용자 입력을 받아 ReActAgent를 호출하는 실행 루틴이다. 사용자가 exit이라고 입력할 때까지 질의를 받고 응답을 출력하는 루프를 구성한다. 이로써 ReActAgent는 반복적인 질의응답 상황에서도 도구를 활용하여 논리적 사고 과정을 수행하고, 외부 정보에 기반한 정답을 도출할 수 있도록 설계되어 있다.

주요 문제 유형

ReAct 기반 에이전트는 구현 및 실전 적용 과정에서 여러 구조적 문제와 비효율이 발생한다. 이러한 문제 사례를 분석하고, 실제 커뮤니티에서 공유된 해결 전략과 함께 개선 방향을 확인해 본다.

추론 오류
에이전트가 잘못된 전제나 문맥에 기반하여 부적절한 도구를 선택하거나 오답을 도출하는 문제가 빈번히 발생한다. 이는 LLM의 한계와 프롬프트 설계 미비, 또는 이전 상태 정보의 왜곡에 기인한다.

반복 및 루프 문제
에이전트가 목표에 도달하지 못하고 동일한 행동을 반복 수행하는 루프에 빠지는 문제가 자주 발생한다. 이는 max_steps 제한이 없거나, 실패 판단 기준이 모호한 경우에 두드러진다.

결과 검증 실패
에이전트가 도출한 결과가 부정확하거나 무의미함에도 불구하고, 이를 최종 결과로 판단하고 종료되는 문제가 있다. 이는 정답 여부를 평가할 수 있는 검증 메커니즘의 부재 혹은 Verifier LLM의 오판 때문이다.

지연 및 비용 문제
ReAct 구조는 추론 Think, 행동 Act, 검증 Verify 단계에서 각각 LLM 호출이 필요하므로 응답 지연이 누적되며, 고성능 모델 사용 시 비용 또한 급증하게 된다.

상태 관리 실패
에이전트가 이전 문맥을 적절히 유지하지 못하거나, 기억을 잘못 참조하여 추론에 실패하는 경우가 있다. 이는 memory overflow, context 길이 초과 등으로 인해 발생한다.

해외 커뮤니티 보고 사례

Reddit, GitHub, Hacker News 등 해외 커뮤니티에서는 위 문제들이 반복적으로 보고되고 있다. 주요 사례는 다음과 같다

r/React 에서는 Langchain의 늦은 React 에이전트 성능에 대한 문제가 제기된 적이 있다. 이런 사례는 수도 없이 많다.

r/AutoGPT에서는 도구 호출 반복, final answer 오류 등 루프와 검증 문제를 다룬 글들이 다수 존재한다.

r/MachineLearning에서는 hallucination 문제, tool selection 오류에 대한 토론이 있다
GitHub Issues의 OpenDevin, Voyager 등의 리포지터리에서도 루프, 상태 관리 문제, 도구 실행 실패가 자주 보고되고 있다

이는 ReAct 구조의 본질적인 설계 한계와 관련되며, 이를 보완하기 위한 다양한 시도들이 커뮤니티에서 논의되고 있다

해결 방안

Verifier LLM의 도입
추론 결과에 대해 별도의 고성능 LLM을 사용하여 정답 여부를 판단하도록 구성하는 방식이다 예를 들어 GPT 35 기반 ReAct 에이전트를 GPT 4 기반 Verifier로 보완하는 사례가 있다

Confidence 기반 조건 실행
LLM이 응답에 대한 신뢰도 confidence score를 함께 출력하도록 하고, 일정 기준 미만일 경우에만 검증자 호출 또는 재시도를 수행하는 방식이다

행동 다양성 유도 및 페널티 적용
같은 도구를 반복 호출하지 않도록 행동 선택 시 다양성을 보장하거나 페널티 기반 scoring 방식을 적용한다

결과 품질 기반 Soft Fallback 적용
결과가 일정 기준 미달일 경우 완료 실패 메시지를 출력하거나, 대체 응답을 제공하는 방식으로 흐름을 마무리한다

도메인 기반 grounding 기법
도메인 특화 지식을 embedding하거나 retriever를 통해 grounding context를 제공함으로써 hallucination을 감소시키는 방식이다

LLM 처리 지연 보완을 위한 전략

ReAct 구조에서는 LLM을 여러 번 호출하는 구조로 인해 응답 시간이 증가하는 문제가 발생한다. 이를 해결하기 위한 보완 전략은 다음과 같다

첫째, 동일한 LLM을 사용하는 것이 아니라 추론 단계는 경량 LLM을 사용하고 검증 단계는 고성능 LLM을 사용하는 방식으로 역할 분리를 통해 효율을 높이는 방안이 있다

둘째, LLM의 confidence score를 이용하여 신뢰도가 충분히 높은 경우에는 검증 단계를 생략함으로써 전체 호출 횟수를 줄일 수 있다

셋째, 여러 행동 중 하나를 선택할 때 prefetch 또는 batch decoding 방식을 사용하여 예측을 병렬적으로 수행하면 전체 응답 지연을 줄일 수 있다

넷째, 반복되는 도구 호출에 대해서는 캐싱 메커니즘을 도입하거나 결과를 저장하여 재활용함으로써 불필요한 호출을 방지할 수 있다

다섯째, 검증자 역할을 하는 LLM을 완전한 LLM이 아니라 룰 기반 평가기 또는 소형 모델로 대체하여 빠르게 판단하게 하는 방식도 고려할 수 있다

코드 기반 구현 예시 요약

ReAct 기반 구조는 보통 다음 흐름으로 구성된다

  1. 사용자 입력 수신
  2. LLM을 통해 행동 계획 수립 Think
  3. 해당 도구 실행 Act
  4. 결과 수신 후 Verifier를 통해 정답 여부 판단 Verify
  5. 정답 시 종료, 실패 시 재시도 또는 fallback 처리

이러한 구조는 Python 코드로 구현할 수 있으며, 각 함수는 다음과 같은 역할을 수행한다

  1. think goal 행동 계획 생성
  2. act action 도구 실행 및 결과 수신
  3. verify result, goal 결과 검증
  4. run goal 전체 흐름 통제 및 반복 로직 포함

결론

ReAct 기반 에이전트는 고도의 자율성을 지닌 시스템이나, 구조적으로 여러 한계가 존재한다. 특히 결과 검증 실패, 반복 루프, hallucination, 비용 증가 문제는 실제 운영 및 서비스화에 있어 큰 장애 요인이 된다. 이를 해결하기 위해서는 검증자 LLM 도입, confidence 기반 흐름 제어, 도구 호출의 다양성 보장, grounding 및 fallback 전략이 함께 설계될 수 있다.


부록: 보고서 형식 AI 에이전트 패턴

ReAct 에이전트는 Reasoning and Acting의 약어로, LLM이 툴과 상호작용하며 문제를 해결하는 패턴이다. 이 구조는 인간처럼 사고하고, 필요한 정보를 외부 도구에서 검색하거나 계산을 수행한 뒤, 최종적으로 응답을 생성한다. 이로 인해 복잡한 질의나 멀티스텝 추론이 필요한 경우에 적합한 패턴으로 여겨진다. 그러나 이러한 장점에도 불구하고, LLM이 여러 차례 호출되고 툴도 반복적으로 작동하므로 응답 시간이 느려지고, API 사용량 증가로 인해 비용이 상승하는 단점이 존재한다.

한편 단순히 보고서를 생성하는 용도로 LLM을 사용하는 경우라면, 복잡한 Reasoning이나 여러 번의 외부 툴 호출은 필요하지 않다. 이때는 ReAct 에이전트를 사용하는 것이 오히려 비효율적이며, 불필요한 구조적 복잡성과 지연을 유발한다. 따라서 보고서 자동 생성이라는 목적에 최적화된 에이전트 구조는 단일 호출 기반 구조가 적합하다.

이 구조는 Single-Shot LLM Invocation 패턴으로 명명된다. 이 방식은 LLM을 한 번만 호출하여 입력된 요구 사항에 맞는 보고서를 즉시 생성한다. 별도의 도구 호출이나 intermediate step 없이 하나의 잘 설계된 프롬프트를 중심으로 전체 작업이 이루어진다. 이 패턴은 속도가 빠르고, 처리 흐름이 단순하며, LLM API 호출 횟수를 최소화하므로 비용이 낮다. 특히 단일 보고서, 요약, 이메일 초안, 블로그 콘텐츠와 같은 목적에 적합하며, 사용자의 입력 내용에 따라 문서 포맷이나 톤을 조정하는 것만으로 충분한 성능을 발휘한다.

단일 호출 구조는 별도의 외부 툴이나 체인을 필요로 하지 않기 때문에 LangChain이나 ReAct 구조 없이도 Python 코드 수준에서 간단히 구현이 가능하다. 예를 들어, 사용자가 요구사항을 입력하면 그에 맞는 템플릿 기반 프롬프트를 작성하고, 이를 LLM에 전달해 한 번에 결과를 반환받는 방식이다. 이러한 프롬프트에 보고서의 형식, 어조, 길이 등을 명시하면, 결과물의 일관성과 품질도 충분히 확보할 수 있다.

이 외에도 필요에 따라 Plan-and-Write 패턴을 고려할 수 있다. 이 패턴은 먼저 전체 문서의 개요를 작성한 뒤, 각 항목에 따라 내용을 생성하는 방식이다. 좀 더 명확한 구성과 섹션 구분이 필요할 때 유용하다. 통계 수치나 외부 데이터를 포함해야 할 경우에는 Toolformer와 같이 선택적으로 툴을 호출하는 패턴을 도입할 수 있다. 마지막으로 문서가 일정한 형식을 유지해야 하거나 후처리를 자동화해야 할 때는 Structured Prompting 패턴을 사용할 수 있으며, JSON 또는 Markdown 형식으로 문서 구조를 고정하여 일관성을 높인다.

결론적으로, 보고서 자동 생성처럼 복잡한 추론이나 외부 툴 연동이 불필요한 작업에는 ReAct와 같은 다단계 에이전트 구조는 적합하지 않다. 대신 단일 LLM 호출로 작업을 완료하는 Single-Shot 방식이 가장 효율적이며, 프롬프트 최적화만으로 높은 품질의 결과를 얻을 수 있다. 

부록: AI 에이전트 오케스트레이션

사용자 질문에 따라 적절한 에이전트 패턴을 선택하기 위해서는 입력의 목적과 복잡성을 빠르게 분류할 수 있는 에이전트 오케스트레이션 메커니즘이 필요하다. 다음 의사코드는 솔류션을 보여준다. 

def classify_input(user_input):
    if "보고서" in user_input or "작성해" in user_input:
        return "simple_generation"
    if "json" in user_input or "표로" in user_input:
        return "structured_output"
    if "비교해" in user_input or "어떤게 더 나아" in user_input:
        return "tool_assisted"
    if "계산" in user_input or any(op in user_input for op in "+-*/"):
        return "tool_assisted"
    return "multi_step_reasoning"

def classify_input(user_input):
    result = rule_based_classifier(user_input)
    if result == "unknown":
        return llm_classifier(user_input)
    return result

여기서는 classify_input 함수가 사용되며, 이 함수는 사용자의 입력 문장을 분석하여 어떤 처리 패턴이 적합한지를 판단하는 역할을 수행한다. 하지만 이 함수 자체가 느리거나 LLM 호출을 과도하게 유발하면 전체 시스템의 응답성이 떨어지므로, 이를 빠르게 실행할 수 있는 설계가 중요하다.

classify_input을 빠르게 실행하기 위한 방법은 크게 세 가지로 구분된다. 첫 번째는 경량 LLM을 사용하는 방식이다. 이 방법은 GPT-3.5 Turbo, Claude Instant, DistilBERT 등의 속도가 빠른 경량 언어 모델을 활용하여 입력을 분류하는 방식이다. 프롬프트는 미리 정의된 범주 중에서 입력이 어떤 태스크 유형에 해당하는지를 선택하게 하며, 이에 따라 적절한 패턴으로 연결된다. 이 방식은 유연하고 적응력이 높지만, 여전히 LLM 호출이기 때문에 응답 속도에 영향을 줄 수 있다. 그럼에도 불구하고 ReAct 등 복잡한 구조보다는 훨씬 빠르고 저비용이다.

두 번째 방식은 규칙 기반 분류 방식이다. 이는 키워드 기반으로 입력 문장을 빠르게 분류하는 방법으로, 예를 들어 ‘보고서’, ‘작성해’ 등의 단어가 포함되어 있으면 단순 생성 태스크로 간주하고, ‘비교’, ‘계산’, ‘표로’ 등의 단어가 있으면 툴 기반 또는 구조화된 출력을 요구하는 태스크로 분류하는 것이다. 이 방식은 매우 빠르며, 수 마이크로초 이내에 실행이 가능하고, 비용도 발생하지 않는다. 그러나 복잡한 문장이나 여러 의미가 섞인 문장에 대해서는 정확하게 분류하지 못할 위험이 있다.

세 번째는 하이브리드 방식이다. 이 방식은 앞서 언급한 규칙 기반 분류기를 우선 사용하고, 그 결과가 불확실하거나 unknown일 경우에만 경량 LLM을 호출하여 보완 분류를 수행한다. 이 구조는 속도와 정확도 사이의 균형을 잡기 위해 매우 현실적인 대안이며, 실제로 OpenAI API 기반 에이전트나 LangChain의 라우팅 모듈에서도 유사한 방식이 채택된다. 구현은 상대적으로 복잡할 수 있으나, 전체 시스템의 응답성과 품질을 함께 유지할 수 있는 방법이다.

결론적으로 classify_input 함수를 빠르게 실행하기 위해서는 입력의 유형과 빈도에 따라 최적화 전략을 선택해야 한다. 단순한 텍스트 입력이 반복되는 시스템에서는 규칙 기반 분류기로 충분하며, 복합 입력이 자주 등장하는 환경에서는 하이브리드 분류기를 설계하는 것이 이상적이다. 

레퍼런스