Search

Agent & Langsmith

대분류
인공지능/데이터
프레임워크
소분류
LangChain/RAG
유형
LangChain
부유형
Application LangChain
주요 레퍼런스
https://wikidocs.net/233782
https://docs.smith.langchain.com/
최종 편집 일시
2024/11/03 11:55
생성 일시
2024/10/30 00:48
13 more properties

사전 Setup (Colab)

OpenAI Key 등록

import os os.environ['OPENAI_API_KEY'] = ''
Python
복사
# 구글 드라이브 연결(데이터 로드를 위해서) try: from google.colab import drive drive.mount('/content/data') DATA_PATH = "/content/data/MyDrive/Colab Notebooks/ai_study/3. Large Language Models/data/" except: DATA_PATH = "./data/"
Python
복사

Install

!pip install -U langchain langchain-community langchain-experimental langchain-core langchain-openai langsmith pypdf faiss-cpu
Python
복사
Langchain에서 만든 온라인 기반의 LLM 애플리케이션 모니터링, 테스트 지원, 배포 지원 도구
Agent에서 분기가 어떻게 이루어지고, Agent가 어떤 구조로 판단을 하는지를 상세하게 모니터링할 수 있다.
이 개념 가이드는 LangSmith에 추적 로그를 기록할 때 이해해야 할 중요한 주제를 다룬다.
Trace : 애플리케이션이 입력에서 출력으로 가는 일련의 단계를 의미
Run : 각각의 개별 단계
Project : 단순히 여러 개의 Trace 모음
문서에서 문서를 검색하고 답변을 생성하는 간단한 RAG(검색-생성) 앱의 컨텍스트에서의 개념

Run

LLM 애플리케이션 내에서 단일 작업 또는 운영 단위를 나타내는 범위
이는 LLM 또는 체인에 대한 단일 호출, 프롬프트 형식 지정 호출, 실행 가능한 람다 호출 등 무엇이든 될 수 있다.

Trace

단일 작업과 관련된 여러 Run의 모음
예를 들어, 사용자 요청이 체인을 트리거하고, 그 체인이 LLM에 대한 호출을 하고, 이후 출력 파서를 호출하는 등의 작업을 한다면, 이러한 모든 Run은 동일한 Trace의 일부가 된다.
Run은 고유한 Trace ID에 의해 Trace에 묶인다.

Projects

여러 Trace의 모음
프로젝트는 단일 애플리케이션이나 서비스와 관련된 모든 Trace를 담는 컨테이너로 생각할 수 있다.
여러 개의 프로젝트를 가질 수 있으며, 각 프로젝트는 여러 Trace를 포함할 수 있다.

Langsmith 설정

LANGCHAIN_TRACING_V2: "true"로 설정하면 추적을 시작
LANGCHAIN_ENDPOINThttps://api.smith.langchain.com 변경하지 않는다.
LANGCHAIN_API_KEY: 이전 단계에서 발급받은 키를 입력
LANGCHAIN_PROJECT: 프로젝트명을 기입하면 해당 프로젝트 그룹으로 모든 실행(Run) 이 추적됨
import os os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com" os.environ["LANGCHAIN_PROJECT"] = "AGENT TUTORIAL" os.environ["LANGCHAIN_API_KEY"] = ""
Python
복사
위 링크에서 회원가입/로그인을 한 후 진행
API 설명 및 Key Type 지정
Projects 접속
Project 선택
실행된 목록 선택
로그 확인
1개의 실행을 한 뒤 retrieve 된 문서의 검색 결과 뿐만 아니라, GPT 의 입출력 내용에 대해서 자세하게 기록한다.
따라서, 문서의 검색된 내용을 확인 후 검색 알고리즘을 변경해야할지 혹은 프롬프트를 변경해야할지 판단하는데 도움이 된다.
뿐만 아니라, 상단에는 1개의 실행(Run) 이 걸린 시간(약 30초)와 사용된 토큰(5,104) 등이 표기가 되고, 토큰에 마우스 호버를 하게 되면 청구 금액까지 표기해 준다.
사용자의 요청을 받은 후 어떤 기능을 어떤 순서로 실행할지 결정하는 역할
Agent의 주요 특징들
목표(정책) 기반 행동: 주어진 목표를 달성하기 위해 행동합니다.
자율성: 목표가 주어지면 자동으로 작동합니다.
감지: 주변 환경에서 정보를 수집합니다.
Tools : Agent가 할 수 있는 일(도구, 기술, 함수들)에 해당.

주요 특징

자율성 : 에이전트는 사전에 정의된 규칙이나 명시적인 프로그래밍 없이도 스스로 결정을 내리고 행동할 수 있다.
목표 지향성 : 특정 목표나 작업을 달성하기 위해 설계되어 있다.
환경 인식 : 주변 환경이나 상황을 인식하고 이에 따라 적응할 수 있다.
도구 사용 : 다양한 도구나 API를 활용하여 작업을 수행할 수 있다.
연속성 : 주어진 목표를 달성하기 위하여 1회 수행이 아닌 반복 수행을 통해 목표 달성을 추구한다.

LangChain 에이전트 구성요소

Tools : 에이전트가 사용할 수 있는 기능들의 집합
from langchain.tools import tool @tool def add(a:int, b:int) -> int: return a + b @tool def minus(a:int, b:int) -> int: return a - b
Python
복사
Toolkits : 관련된 도구들의 그룹
toolkit = [ add, minus ]
Python
복사
Agent : 의사 결정을 담당하는 핵심 컴포넌트
agent = create_openai_tools_agen( llm, toolkit, prompt )
Python
복사
AgentExecutor : 에이전트의 실행을 관리하는 컴포넌트
agent_executor = AgentExecutor( agent=agent, tools=toolkit, verbos=True ) result = agent_executor.invoke({ "input":"what is 1 + 1" })
Python
복사

활용 사례 (근데 이건 LLM이 하는거 아닌가?)

정보 검색 및 분석: 웹 검색, 데이터베이스 쿼리 등을 수행
작업 자동화: 복잡한 워크플로우를 자동으로 처리
고객 서비스: 질문에 답변하고 문제 해결
의사 결정 지원: 데이터를 분석하고 권장 사항을 제공
창의적 작업: 글쓰기, 코드 생성 등의 창의적 작업을 수행

장단점

장점
복잡한 작업의 자동화
유연성과 적응성 (이건 ㅇㅈ)
다양한 도구와의 통합 가능성
단점
제어와 예측 가능성의 어려움
계산 비용과 리소스 요구사항

Agent의 작업 처리 순서도

Agent의 작업 수행 과정
1.
Input: 사용자가 Agent에게 작업을 할당한다.
2.
Thought: Agent가 작업을 완수하기 위해 무엇을 할지 생각한다.
3.
Action/Action Input: 사용할 도구를 결정하고, 도구의 입력(함수의 입력값)을 결정한다.
4.
Observation: 도구의 출력 결과를 관찰한다.
5.
관찰 결과 작업을 완료(Finish)했다는 판단에 도달할 때까지 2~4번 과정을 반복한다.
Agent Executor : LangChain에서 해당 과정을 처리하는 객체

Agent 사용법

에이전트, 체인 또는 LLM이 외부 세계와 상호작용하기 위한 인터페이스
LangChain 에서 기본 제공하는 도구를 사용하여 쉽게 도구를 활용할 수 있으며, 사용자 정의 도구(Custom Tool) 를 쉽게 구축하는 것도 가능하다.
이 도구는 Python 코드를 REPL(Read-Eval-Print Loop) 환경에서 실행하기 위한 두 가지 주요 클래스를 제공한다.
from langchain_core.tools import Tool from langchain_experimental.utilities import PythonREPL from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableLambda
Python
복사

Tool 생성 & 사용법

tool 생성
python_tool = PythonREPL() python_tool.run("print(1+1)")
Python
복사
tool 사용하는 함수 정의
# 파이썬 코드를 실행하고 중간 과정을 출력하고 도구 실행 결과를 반환하는 함수 def print_and_execute(code, debug=True): if debug: print("CODE:") print(code) return python_tool.run(code)
Python
복사
prompt 정의
# 파이썬 코드를 작성하도록 요청하는 프롬프트 prompt = ChatPromptTemplate.from_messages( [ ( "system", "You are Raymond Hetting, an expert python programmer, well versed in meta-programming and elegant, concise and short but well documented code. You follow the PEP8 style guide. " "Return only the code, no intro, no explanation, no chatty, no markdown, no code block, no nothing. Just the code.", ), ("human", "{input}"), ] )
Python
복사
LLM & chain 정의 후 실행
# LLM 모델 생성 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 프롬프트와 LLM 모델을 사용하여 체인 생성 chain = prompt | llm | StrOutputParser() | RunnableLambda(print_and_execute)
Python
복사
# 결과 출력 print(chain.invoke("로또 번호 생성기를 출력하는 코드를 작성하세요.")) CODE: import random def generate_lotto_numbers(): """Generate a set of 6 unique lotto numbers from 1 to 45.""" return sorted(random.sample(range(1, 46), 6)) if __name__ == "__main__": print(generate_lotto_numbers())
Python
복사
Langsmith 로그 확인
Tavily 검색 API를 활용하여 검색 기능을 구현하는 도구
LLMs의 한계점
대규모 언어 모델(Large Language Models, LLMs)은 자연어 처리 분야에서 혁명적인 발전을 이루었지만, 여전히 중요한 한계점을 가지고 있다. 그 중 하나는 최신 정보 처리의 어려움이다.
대부분의 LLM은 특정 시점까지의 데이터로 학습되어 있어, 그 이후의 정보나 실시간 변화하는 상황에 대해 정확한 답변을 제공하기 어렵습니다.
Tavily 주요 특징
AI 최적화: LLM과의 원활한 통합을 위해 설계되었습니다.
최신 정보: 실시간으로 업데이트되는 정보를 제공합니다.
다양한 검색 모드: 일반 검색, 뉴스 검색, 이미지 검색 등 다양한 모드를 제공합니다.
콘텐츠 필터링: 신뢰할 수 있는 소스의 정보만을 제공합니다.
무료 사용량: 월 1,000회의 무료 API 호출을 제공합니다.
# TAVILY API KEY를 기입 os.environ["TAVILY_API_KEY"] = ""
Python
복사
TavilySearchResults
Tavily 검색 API를 쿼리하고 JSON 형식의 결과를 반환
포괄적이고 정확하며 신뢰할 수 있는 결과에 최적화된 검색 엔진
현재 이벤트에 대한 질문에 답변할 때 유용
주요 매개변수
max_results (int): 반환할 최대 검색 결과 수 (기본값: 5)
search_depth (str): 검색 깊이 ("basic" 또는 "advanced")
include_domains (List[str]): 검색 결과에 포함할 도메인 목록
exclude_domains (List[str]): 검색 결과에서 제외할 도메인 목록
include_answer (bool): 원본 쿼리에 대한 짧은 답변 포함 여부
include_raw_content (bool): 각 사이트의 정제된 HTML 콘텐츠 포함 여부
include_images (bool): 쿼리 관련 이미지 목록 포함 여부
from langchain_community.tools.tavily_search import TavilySearchResults # 도구 생성 tool = TavilySearchResults( max_results=6, include_answer=True, include_raw_content=True, # include_images=True, # search_depth="advanced", # or "basic" include_domains=["github.io", "wikidocs.net"], # exclude_domains = [] ) # 도구 실행 result = tool.invoke({"query": "LangChain Tools 에 대해서 알려주세요"}) > result [{'url': 'https://wikidocs.net/253106', 'content': '도구(Tools) 02. 도구 바인딩(Binding Tools) 03. 에이전트(Agent) 04. Claude, Gemini, Ollama ... LangChain 에 대해서 알려주세요.\\n'}, {'url': 'https://wikidocs.net/262582', 'content': "LangChain 에서 기본 제공하는 도구를 사용하여 쉽게 도구를 활용할 수 있으며, 사용자 정의 도구(Custom Tool) 를 쉽게 구축하는 것도 가능합니다. LangChain 한국어 튜토리얼\\n바로가기 👀\\n[LangChain] 에이전트(Agent)와 도구(tools)를 활용한 지능형 검색 시스템 구축 가이드\\n2024년 0209일\\n41 분 소요\\n이 글에서는 LangChain 의 Agent 프레임워크를 활용하여 복잡한 검색과 📍 전체 템플릿 코드\\n다음의 추적 링크에서 자세한 단계별 수행 결과를 확인할 수 있습니다\\nLangSmith 추적\\n마무리입니다!\\n 문서 기반 QA 시스템 설계 방법 - 심화편\\n2024년 0206일\\n22 분 소요\\nLangChain의 RAG 시스템을 통... > len(result) # 리스트 형태로 넘어오기 때문에 결과 개수를 반환할 수 있습니다. 6
Python
복사
해당 데코레이터는 함수를 도구로 변환하는 기능을 제공
다양한 옵션을 통해 도구의 동작을 커스터마이즈할 수 있다.
from langchain_core.tools import tool @tool def multiply(a: int, b: int) -> int: """Multiply two numbers.""" return a * b
Python
복사
# Let's inspect some of the attributes associated with the tool. print(multiply.name) print(multiply.description) print(multiply.args) multiply Multiply two numbers. {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}
Python
복사

Tool

실험을 위한 도구(tool) 를 정의
get_word_length : 단어의 길이를 반환하는 함수
add_function : 두 숫자를 더하는 함수
naver_news_crawl : 네이버 뉴스 기사를 크롤링하여 본문 내용을 반환하는 함수
import re import requests from bs4 import BeautifulSoup from langchain.agents import tool # 도구를 정의합니다. @tool def get_word_length(word: str) -> int: """Returns the length of a word.""" return len(word) @tool def add_function(a: float, b: float) -> float: """Adds two numbers together.""" return a + b @tool def naver_news_crawl(news_url: str) -> str: """Crawls a 네이버 (naver.com) news article and returns the body content.""" # HTTP GET 요청 보내기 response = requests.get(news_url) # 요청이 성공했는지 확인 if response.status_code == 200: # BeautifulSoup을 사용하여 HTML 파싱 soup = BeautifulSoup(response.text, "html.parser") # 원하는 정보 추출 title = soup.find("h2", id="title_area").get_text() content = soup.find("div", id="contents").get_text() cleaned_title = re.sub(r"\n{2,}", "\n", title) cleaned_content = re.sub(r"\n{2,}", "\n", content) else: print(f"HTTP 요청 실패. 응답 코드: {response.status_code}") return f"{cleaned_title}\n{cleaned_content}" tools = [get_word_length, add_function, naver_news_crawl]
Python
복사

bind_tools()

llm 모델에 bind_tools() 를 사용하여 도구를 바인딩
from langchain_openai import ChatOpenAI # 모델 생성 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 도구 바인딩 llm_with_tools = llm.bind_tools(tools) # 실행 결과 llm_with_tools.invoke("What is the length of the word 'teddynote'?").tool_calls
Python
복사

Chain

from langchain_core.output_parsers.openai_tools import JsonOutputToolsParser # 도구 바인딩 + 도구 파서 chain = llm_with_tools | JsonOutputToolsParser(tools=tools) # 실행 결과 tool_call_results = chain.invoke("What is the length of the word 'teddynote'?")
Python
복사
type: 도구의 이름
args: 도구에 전달되는 인자
print(tool_call_results, end="\n\n==========\n\n") # 첫 번째 도구 호출 결과 single_result = tool_call_results[0] # 도구 이름 print(single_result["type"]) # 도구 인자 print(single_result["args"]) tool_call_results[0]["type"], tools[0].name
Python
복사
def execute_tool_calls(tool_call_results): """ 도구 호출 결과를 실행하는 함수 :param tool_call_results: 도구 호출 결과 리스트 :param tools: 사용 가능한 도구 리스트 """ # 도구 호출 결과 리스트를 순회합니다. for tool_call_result in tool_call_results: # 도구의 이름과 인자를 추출합니다. tool_name = tool_call_result["type"] tool_args = tool_call_result["args"] # 도구 이름과 일치하는 도구를 찾아 실행합니다. # next() 함수를 사용하여 일치하는 첫 번째 도구를 찾습니다. matching_tool = next((tool for tool in tools if tool.name == tool_name), None) if matching_tool: # 일치하는 도구를 찾았다면 해당 도구를 실행합니다. result = matching_tool.invoke(tool_args) # 실행 결과를 출력합니다. print(f"[실행도구] {tool_name}\n[실행결과] {result}") else: # 일치하는 도구를 찾지 못했다면 경고 메시지를 출력합니다. print(f"경고: {tool_name}에 해당하는 도구를 찾을 수 없습니다.")
Python
복사

bind_tools + Parser + Execution

llm_with_tools : 도구를 바인딩한 모델
JsonOutputToolsParser : 도구 호출 결과를 파싱하는 파서
execute_tool_calls : 도구 호출 결과를 실행하는 함수
from langchain_core.output_parsers.openai_tools import JsonOutputToolsParser # bind_tools + Parser + Execution chain = llm_with_tools | JsonOutputToolsParser(tools=tools) | execute_tool_calls # 실행 결과 chain.invoke("What is the length of the word 'teddynote'?") # 실행 결과 chain.invoke("114.5 + 121.2") print(114.5 + 121.2) # 실행 결과 chain.invoke( "뉴스 기사 내용을 크롤링해줘: https://n.news.naver.com/mnews/hotissue/article/092/0002347672?type=series&cid=2000065" )
Python
복사
도구 호출을 사용하면 모델이 하나 이상의 도구(tool) 가 호출되어야 하는 시기를 감지하고 해당 도구에 전달해야 하는 입력 으로 전달할 수 있다.
API 호출에서 도구를 설명하고 모델이 이러한 도구를 호출하기 위한 인수가 포함된 JSON과 같은 구조화된 객체를 출력하도록 지능적으로 선택할 수 있다.
도구 API의 목표는 일반 텍스트 완성이나 채팅 API를 사용하여 수행할 수 있는 것보다 더 안정적으로 유효하고 유용한 도구 호출(tool call) 을 반환하는 것

Tool 생성

from langchain.tools import tool from typing import List, Dict, Annotated from langchain_community.tools.tavily_search import TavilySearchResults from langchain_experimental.utilities import PythonREPL # 도구 생성 @tool def search_news(query: str) -> List[Dict[str, str]]: """Search News by input keyword""" news_tool = TavilySearchResults( max_results=6, include_answer=True, include_raw_content=True, # include_images=True, # search_depth="advanced", # or "basic" include_domains=["google.com", "naver.com"], # exclude_domains = [] ) return news_tool.invoke({"query": query}) # 도구 생성 @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}") 도구 이름: search_news 도구 설명: Search News by input keyword 도구 이름: python_repl_tool 도구 설명: 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. # tools 정의 tools = [search_news, python_repl_tool]
Python
복사

Prompt 생성

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}"), ] )
Python
복사

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)
Python
복사

AgentExecutor

도구를 사용하는 에이전트를 실행하는 클래스
주요 메서드
invoke: 에이전트 실행
stream: 최종 출력에 도달하는 데 필요한 단계를 스트리밍
주요 속성
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 하지 않음, 또는 트리밍 함수)
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"])
Python
복사