14. 에이전트 루프 구축하기

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain.tools import tool
 
# 도구 정의
@tool
def get_weather(city: str) -> str:
    """도시의 현재 날씨를 가져옵니다."""
    fake_data = {"Tokyo": "18°C, cloudy", "Cairo": "31°C, sunny"}
    return fake_data.get(city, f"No weather data for {city}.")
 
@tool
def calculate(expression: str) -> str:
    """간단한 산술 표현식을 계산합니다. 예: '3 * 21'."""
    return str(eval(expression))  # 주의: eval()은 보안에 취약합니다. 프로덕션에서는 사용하지 마세요.
 
# 도구 바인딩
tools = [get_weather, calculate]
llm = ChatOpenAI(model="gpt-5-mini")
llm_with_tools = llm.bind_tools(tools)
tool_map = {t.name: t for t in tools}

def run_agent(user_input: str) -> str:
    """LLM이 최종 답변을 반환할 때까지 Think-Act-Observe 루프를 실행합니다."""
    messages = [
        SystemMessage(content="You are a helpful assistant."),
        HumanMessage(content=user_input),
    ]
 
    while True:
        # THINK: LLM을 호출하여 판단을 요청합니다
        print("THINK: LLM에게 판단을 요청합니다")
        ai_message = llm_with_tools.invoke(messages)
        messages.append(ai_message)
 
        # 종료 조건: tool_calls가 없으면 최종 답변입니다
        if not ai_message.tool_calls:
            return ai_message.content
 
        # ACT + OBSERVE: 요청된 도구를 실행하고 결과를 대화에 추가합니다
        for tool_call in ai_message.tool_calls:
            selected_tool = tool_map[tool_call["name"]]
            print(f"ACT: '{tool_call['name']}' 도구 호출, 인자={tool_call['args']}")
            
            tool_message = selected_tool.invoke(tool_call)
            print(f"OBSERVE: {tool_message.content}")
            
            messages.append(tool_message)

answer = run_agent("What's the weather in Tokyo, and is it warm enough for a walk?")
print(f'최종 응답: {answer}')

THINK: LLM에게 판단을 요청합니다
ACT: 'get_weather' 도구 호출, 인자={'city': 'Tokyo'}
OBSERVE: 18°C, cloudy
THINK: LLM에게 판단을 요청합니다
최종 응답: Right now in Tokyo it’s 18°C (about 64°F) and cloudy.
That temperature is generally mild and comfortable for a walk for most people.

answer = run_agent("Get the temperature in Cairo, then multiply the number by 3.")
print(f'최종 응답: {answer}')

THINK: LLM에게 판단을 요청합니다
ACT: 'get_weather' 도구 호출, 인자={'city': 'Cairo'}
OBSERVE: 31°C, sunny
THINK: LLM에게 판단을 요청합니다
ACT: 'calculate' 도구 호출, 인자={'expression': '31 * 3'}
OBSERVE: 93
THINK: LLM에게 판단을 요청합니다
최종 응답: Current temperature in Cairo: 31°C. Multiplied by 3 = 93.

def run_agent(user_input: str, max_steps: int = 10) -> str:
    """max_steps 이내에서 Think-Act-Observe 루프를 실행합니다."""
    messages = [
        SystemMessage(content="You are a helpful assistant."),
        HumanMessage(content=user_input),
    ]
 
    for step in range(max_steps):
        # THINK
        ai_message = llm_with_tools.invoke(messages)
        messages.append(ai_message)
 
        # 종료 조건: tool_calls가 없으면 최종 답변입니다
        if not ai_message.tool_calls:
            return ai_message.content
 
        # ACT + OBSERVE
        for tool_call in ai_message.tool_calls:
            selected_tool = tool_map[tool_call["name"]]
            tool_message = selected_tool.invoke(tool_call)
            messages.append(tool_message)
 
    # max_steps에 도달: 최종 답변 없이 루프가 종료되었습니다
    return f"[최대 반복 횟수({max_steps})에 도달하여 중단되었습니다]"

@tool
def unhelpful_search(query: str) -> str:
    """정보를 검색합니다."""
    return "No results found. Try rephrasing your query."
 
llm_with_bad_tool = llm.bind_tools([unhelpful_search])
tool_map_bad = {unhelpful_search.name: unhelpful_search}
 
def run_agent_bad(user_input: str, max_steps: int = 5) -> str:
    messages = [
        SystemMessage(content=(
            "You must ALWAYS use the unhelpful_search tool to find information. "
            "You are NOT allowed to answer from your own knowledge. "
            "If the tool returns no results, you MUST rephrase and search again. "
            "Keep searching until you find the answer."
        )),
        HumanMessage(content=user_input),
    ]
 
    for step in range(max_steps):
        print(f"--- Step {step + 1} ---")
        ai_message = llm_with_bad_tool.invoke(messages)
        messages.append(ai_message)
 
        if not ai_message.tool_calls:
            return ai_message.content
 
        for tool_call in ai_message.tool_calls:
            selected_tool = tool_map_bad[tool_call["name"]]
            tool_message = selected_tool.invoke(tool_call)
            print(f"ACT: '{tool_call['name']}' → {tool_message.content}")
            messages.append(tool_message)
 
    return f"[최대 반복 횟수({max_steps})에 도달하여 중단되었습니다]"
 
answer = run_agent_bad("What is the population of France?")
print(f"\n최종 응답: {answer}")

--- Step 1 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
--- Step 2 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
--- Step 3 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
--- Step 4 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
--- Step 5 ---
ACT: 'unhelpful_search' → No results found. Try rephrasing your query.
 
최종 응답: [최대 반복 횟수(5)에 도달하여 중단되었습니다]

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from langchain.tools import tool
 
# 도구 정의
@tool
def calculate(expression: str) -> str:
    """간단한 산술 표현식을 계산합니다. 예: '3 * 21'."""
    return str(eval(expression))  # 주의: eval()은 보안에 취약합니다. 프로덕션에서는 사용하지 마세요.
 
# 도구 바인딩
tools = [calculate]
llm = ChatOpenAI(model="gpt-5-mini")
llm_with_tools = llm.bind_tools(tools)
tool_map = {t.name: t for t in tools}
 
def run_agent(user_input: str, max_steps: int = 10) -> str:
    """도구 실패 시 오류를 LLM에게 전달하여 자기 교정을 가능하게 하는 에이전트 루프."""
    messages = [
        SystemMessage(content="You are a helpful assistant."),
        HumanMessage(content=user_input),
    ]
 
    for step in range(max_steps):
        # THINK
        print("THINK: LLM에게 판단을 요청합니다")
        ai_message = llm_with_tools.invoke(messages)
        messages.append(ai_message)
 
        if not ai_message.tool_calls:
            return ai_message.content
 
        # ACT + OBSERVE
        for tool_call in ai_message.tool_calls:
            selected_tool = tool_map[tool_call["name"]]
            try:
                print(f"ACT: '{tool_call['name']}' 도구 호출, 인자={tool_call['args']}")
                tool_message = selected_tool.invoke(tool_call)
                print(f"OBSERVE: {tool_message.content}")
            except Exception as e:
                print(f"OBSERVE: Error - {e}")
                tool_message = ToolMessage(
                    content=f"Error: {e}",
                    tool_call_id=tool_call["id"],
                )
            messages.append(tool_message)
 
    return f"[최대 반복 횟수({max_steps})에 도달하여 중단되었습니다]"

answer = run_agent("Use the calculator tool to compute 10 / 0")
print(f"최종 응답: {answer}")

THINK: LLM에게 판단을 요청합니다
ACT: 'calculate' 도구 호출, 인자={'expression': '10 / 0'}
OBSERVE: Error - division by zero
THINK: LLM에게 판단을 요청합니다
최종 응답: I used the calculator tool and it returned an error: "division by zero."
 
Explanation: 10 / 0 is undefined in ordinary arithmetic, so it cannot produce a finite number.
 
Would you like me to:
- compute the one-sided limits,
- show the IEEE-754 floating-point result,
- or evaluate a different expression?