패스트캠퍼스 환급챌린지 41일차 : 테디노트의 RAG 비법노트 강의 후기

본 포스팅은 패스트캠퍼스 환급 챌린지 참여를 위해 작성하였습니다

https://fastcampus.info/4n8ztzq

(~6/20) 50일의 기적 AI 환급반💫 | 패스트캠퍼스

패스트캠퍼스 환급챌린지 41일차!

1) 공부 시작 시간 인증

 

2) 공부 종료 시간 인증

 

3) 강의 수강 클립 인증

 

4) 학습 인증샷

 

5) 학습통계

---

Today I Learned

도구 호출 에이전트(Tool Calling Agent)

• 도구 호출을 사용하면 모델이 하나 이상의 도구(tool) 가 호출되어야 하는 시기를 감지하고 해당 도구에 전달해야 하는 입력 으로 전달할 수 있습니다.
• API 호출에서 도구를 설명하고 모델이 이러한 도구를 호출하기 위한 인수가 포함된 JSON과 같은 구조화된 객체를 출력하도록 지능적으로 선택할 수 있습니다.
• 도구 API 의 목표는 일반 텍스트 완성이나 채팅 API를 사용하여 수행할 수 있는 것보다 더 안정적으로 유효하고 유용한 도구 호출(tool call) 을 반환하는 것입니다.
• 이러한 구조화된 출력을 도구 호출 채팅 모델에 여러 도구를 바인딩하고 모델이 호출할 도구를 선택할 수 있다는 사실과 결합하여 쿼리가 해결될 때까지 반복적으로 도구를 호출하고 결과를 수신하는 에이전트를 만들 수 있습니다.
• 이것은 OpenAI 의 특정 도구 호출 스타일에 맞게 설계된 OpenAI 도구 에이전트의 보다 일반화된 버전 입니다.
• 이 에이전트는 LangChain의 ToolCall 인터페이스를 사용하여 OpenAI 외에도 Anthropic, Google Gemini, Mistral과 같은 더 광범위한 공급자 구현을 지원합니다.

참고 : LangChain 공식 도큐먼트

# API 키를 환경변수로 관리하기 위한 설정 파일
from dotenv import load_dotenv
from langchain.tools import tool
from typing import List, Dict, Annotated
from langchain_teddynote.tools import GoogleNews
from langchain_experimental.utilities import PythonREPL

# API 키 정보 로드
load_dotenv()

# 도구 생성
@tool
def search_news(query: str) -> List[Dict[str, str]]:
    """Search Google News by input keyword"""
    news_tool = GoogleNews()
    return news_tool.search_by_keyword(query, k=5)

# 도구 생성
@tool
def python_repl_tool(
    code: Annotated[str, "The python code to execute to generate your chart."],
):
    """Use this to execute python code. If you want to see the output of a value,
    you should print it out with `print(...)`. This is visible to the user."""
    result = ""
    try:
        result = PythonREPL().run(code)
    except BaseException as e:
        print(f"Failed to execute. Error: {repr(e)}")
    finally:
        return result

print(f"도구 이름: {search_news.name}")
print(f"도구 설명: {search_news.description}")
print(f"도구 이름: {python_repl_tool.name}")
print(f"도구 설명: {python_repl_tool.description}")

# tools 정의
tools = [search_news, python_repl_tool]

Agent 프롬프트 생성

• chat_history : 이전 대화 내용을 저장하는 변수 (멀티턴을 지원하지 않는다면, 생략 가능합니다.)
• agent_scratchpad : 에이전트가 임시로 저장하는 변수
• input : 사용자의 입력

from langchain_core.prompts import ChatPromptTemplate

# 프롬프트 생성
# 프롬프트는 에이전트에게 모델이 수행할 작업을 설명하는 텍스트를 제공합니다. (도구의 이름과 역할을 입력)
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            "You are a helpful assistant. "
            "Make sure to use the `search_news` tool for searching keyword related news.",
        ),
        ("placeholder", "{chat_history}"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ]
)

Agent 생성

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent

# LLM 정의
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Agent 생성
agent = create_tool_calling_agent(llm, tools, prompt)

AgentExecutor

AgentExecutor는 도구를 사용하는 에이전트를 실행하는 클래스입니다.

주요 속성

• agent: 실행 루프의 각 단계에서 계획을 생성하고 행동을 결정하는 에이전트
• tools: 에이전트가 사용할 수 있는 유효한 도구 목록
• return_intermediate_steps: 최종 출력과 함께 에이전트의 중간 단계 경로를 반환할지 여부
• max_iterations: 실행 루프를 종료하기 전 최대 단계 수
• max_execution_time: 실행 루프에 소요될 수 있는 최대 시간
• early_stopping_method: 에이전트가 AgentFinish를 반환하지 않을 때 사용할 조기 종료 방법. ("force" or "generate")
  • "force" 는 시간 또는 반복 제한에 도달하여 중지되었다는 문자열을 반환합니다.
  • "generate" 는 에이전트의 LLM 체인을 마지막으로 한 번 호출하여 이전 단계에 따라 최종 답변을 생성합니다.
• handle_parsing_errors: 에이전트의 출력 파서에서 발생한 오류 처리 방법. (True, False, 또는 오류 처리 함수)
• trim_intermediate_steps: 중간 단계를 트리밍하는 방법. (-1 trim 하지 않음, 또는 트리밍 함수)

주요 메서드

1. invoke: 에이전트 실행
2. stream: 최종 출력에 도달하는 데 필요한 단계를 스트리밍

주요 기능

1. 도구 검증: 에이전트와 호환되는 도구인지 확인
2. 실행 제어: 최대 반복 횟수 및 실행 시간 제한 설정 가능
3. 오류 처리: 출력 파싱 오류에 대한 다양한 처리 옵션 제공
4. 중간 단계 관리: 중간 단계 트리밍 및 반환 옵션
5. 비동기 지원: 비동기 실행 및 스트리밍 지원

최적화 팁

• max_iterations와 max_execution_time을 적절히 설정하여 실행 시간 관리
• trim_intermediate_steps를 활용하여 메모리 사용량 최적화
• 복잡한 작업의 경우 stream 메서드를 사용하여 단계별 결과 모니터링 from langchain.agents import AgentExecutor

from langchain.agents import AgentExecutor

# AgentExecutor 생성
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
    max_execution_time=10,
    handle_parsing_errors=True,
)

# AgentExecutor 실행
result = agent_executor.invoke({"input": "AI 투자와 관련된 뉴스를 검색해 주세요."})

print("Agent 실행 결과:")
print(result["output"])

Stream 출력으로 단계별 결과 확인

AgentExecutor의 stream() 메소드를 사용하여 에이전트의 중간 단계를 스트리밍할 것입니다.

stream()의 출력은 (Action, Observation) 쌍 사이에서 번갈아 나타나며, 최종적으로 에이전트가 목표를 달성했다면 답변으로 마무리됩니다.

다음과 같은 형태로 보일 것입니다.

1. Action 출력
2. Observation 출력
3. Action 출력
4. Observation 출력

... (목표 달성까지 계속) ...

그 다음, 최종 목표가 달성되면 에이전트는 최종 답변을 출력할 것입니다.

이러한 출력의 내용은 다음과 같이 요약됩니다.

출력 내용

Action	actions: AgentAction 또는 그 하위 클래스<br>messages: 액션 호출에 해당하는 채팅 메시지
Observation	steps: 현재 액션과 그 관찰을 포함한 에이전트가 지금까지 수행한 작업의 기록<br>messages: 함수 호출 결과(즉, 관찰)를 포함한 채팅 메시지
Final Answer	output: AgentFinish<br>messages: 최종 출력을 포함한 채팅 메시지

from langchain.agents import AgentExecutor

# AgentExecutor 생성
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=False,
    handle_parsing_errors=True,
)

# 스트리밍 모드로 실행합니다
result = agent_executor.stream({"input": "AI 투자와 관련된 뉴스를 검색해 주세요."})

for step in result:
    # 중간 단계 출력
    print(step)
    print("===" * 20)

중간 단계 출력을 사용자 정의 함수로 출력

• 다음의 3개 함수를 정의하고 이를 통해 중간 단계 출력을 사용자 정의합니다.
  • tool_callback: 도구 호출 출력을 처리하는 함수
  • observation_callback: 관찰(Observation) 출력을 처리하는 함수
  • result_callback: 최종 답변 출력을 처리하는 함수
• Agent 의 중간 단계 과정을 깔끔하게 출력하기 위하여 사용되는 콜백 함수입니다.
• 이 콜백 함수는 Streamlit 에서 중간 단계를 출력하여 사용자에게 제공할 때 유용할 수 있습니다.

from langchain_teddynote.messages import AgentStreamParser

agent_stream_parser = AgentStreamParser()

# 질의에 대한 답변을 스트리밍으로 출력 요청
result = agent_executor.stream(
    {"input": "matplotlib 을 사용하여 pie 차트를 그리는 코드를 작성하고 실행하세요."}
)

for step in result:
    # 중간 단계를 parser 를 사용하여 단계별로 출력
    # print(step)
    agent_stream_parser.process_agent_steps(step)
    
# AgentCallbacks와 AgentStreamParser를 langchain_teddynote.messages에서 가져옵니다.
from langchain_teddynote.messages import AgentCallbacks, AgentStreamParser

# 도구 호출 시 실행되는 콜백 함수입니다.
def tool_callback(tool) -> None:
    print("<<<<<<< 도구 호출 >>>>>>")
    print(f"Tool: {tool.get('tool')}")  # 사용된 도구의 이름을 출력합니다.
    print("<<<<<<< 도구 호출 >>>>>>")

# 관찰 결과를 출력하는 콜백 함수입니다.
def observation_callback(observation) -> None:
    print("<<<<<<< 관찰 내용 >>>>>>")
    print(
        f"Observation: {observation.get('observation')[0]}"
    )  # 관찰 내용을 출력합니다.
    print("<<<<<<< 관찰 내용 >>>>>>")

# 최종 결과를 출력하는 콜백 함수입니다.
def result_callback(result: str) -> None:
    print("<<<<<<< 최종 답변 >>>>>>")
    print(result)  # 최종 답변을 출력합니다.
    print("<<<<<<< 최종 답변 >>>>>>")

# AgentCallbacks 객체를 생성하여 각 단계별 콜백 함수를 설정합니다.
agent_callbacks = AgentCallbacks(
    tool_callback=tool_callback,
    observation_callback=observation_callback,
    result_callback=result_callback,
)

# AgentStreamParser 객체를 생성하여 에이전트의 실행 과정을 파싱합니다.
agent_stream_parser = AgentStreamParser(agent_callbacks)

# 질의에 대한 답변을 스트리밍으로 출력 요청
result = agent_executor.stream({"input": "AI 투자관련 뉴스를 검색해 주세요."})

for step in result:
    # 중간 단계를 parser 를 사용하여 단계별로 출력
    agent_stream_parser.process_agent_steps(step)