이 글은 Functions, Tools and Agents with LangChain 코스를 보고 정리한 글입니다.
이 코스의 목적은 LLM 을 외부 세상에 있는 소프트웨어와 상화작용해서 더 효과적으로 작동할 수 있도록 하는 방법을 배움.
Outline:
- OpenAI function calling
- LangChain Expression Language (LCEL)
- OpenAI funciton calling in LangChain
- Tagging and extraction using OpenAI function calling
- Tools and Routing
- Conversation Agent
1. OpenAI Function Calling
OpenAI function calling 은 LLM 이 함수를 호출할 수 있게 해주는거임. 그리고 함수 호출 결과를 보고 답변을 작성할 수 있도록 해준다.
간단하니까 바로 OpenAI function calling 의 사용 방법부터 보자.
import os
import openai
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv()) # read local .env file
openai.api_key = os.environ['OPENAI_API_KEY']
import json
# Example dummy function hard coded to return the same weather
# In production, this could be your backend API or an external API
def get_current_weather(location, unit="fahrenheit"):
"""Get the current weather in a given location"""
weather_info = {
"location": location,
"temperature": "72",
"unit": unit,
"forecast": ["sunny", "windy"],
}
return json.dumps(weather_info)
# define a function
functions = [
{
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
},
"required": ["location"],
},
}
]
messages = [
{
"role": "user",
"content": "What's the weather like in Boston?"
}
]
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo-0613",
messages=messages,
functions=functions
)
messages = [
{
"role": "user",
"content": "hi!",
}
]
function calling 변수에 들어갈 수 있는 값은 여러가지 있음.
- "auto": 모델이 적절하다고 판단될 때 자동으로 함수를 호출함.
- "none": 모델이 함수를 호출하지 않도록 강제함
- { "name": "", "arguments": <arguments> }: 특정 함수와 그 함수에 전달될 인수를 명시적으로 지정하여 함수를 호출함
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo-0613",
messages=messages,
functions=functions,
function_call={"name": "get_current_weather"},
)
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo-0613",
messages=messages,
functions=functions,
function_call="none",
)
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo-0613",
messages=messages,
functions=functions,
function_call="auto",
)
2. LangChain Expression Language (LCEL)
LangChain Expression 에 대해서 알아보자. 이건 함수형 프로그래밍 처럼 Chain (= Prompt + Model) 을 결합해서 복잡한 Chain 을 쉽게 만들고, 이를 통해 LLM 의 최종 답변을 얻어올 수 있는 방법임.
LLM 은 하나의 프롬포트와 하나의 모델 호출로 최종 답변을 생성하지 않고 복잡한 Workflow 를 가지고 있기 때문에 이걸 사용하면 훨씬 편하게 작성할 수 있음.
LCEL 은 다음과 같은 방식으로 선언적으로 복잡한 Chain 을 만들 수 있음.
chain = RunnableMap({
"context": lambda x: retriever.get_relevant_documents(x["question"]),
"question": lambda x: x["question"]
}) | prompt | model | output_parser
LCEL 의 장점은 다음과 같은 기능을 지원받아서 사용할 수 있다는거임:
- Full Sync and Async Support
- Streaming Support
- Optimized Parallel Execution
- Retries and Fallbacks
- Intermediate Results Access
- Input and Output Schemas
- Seamless LangSmith tracing
- Seamless LangServe deployment
- 자세한 내용은 https://python.langchain.com/v0.1/docs/expression_language/
LCEL 을 이용해서 간단한 Chain 을 만드는 예제:
import os
import openai
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv()) # read local .env file
openai.api_key = os.environ['OPENAI_API_KEY']
from langchain.prompts import ChatPromptTemplate
from langchain.chat_models import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser
prompt = ChatPromptTemplate.from_template(
"tell me a short joke about {topic}"
)
model = ChatOpenAI()
output_parser = StrOutputParser()
chain = prompt | model | output_parser
chain.invoke({"topic": "bears"})
이번엔 좀 더 복잡한 Chain 을 만드는 예제를 보자.
- LangChain 의 Runnable 에 대한 설명은 이 링크를 참고하자.
- https://v02.api.js.langchain.com/modules/langchain_core_runnables.html
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import DocArrayInMemorySearch
from langchain.schema.runnable import RunnableMap
vectorstore = DocArrayInMemorySearch.from_texts(
["harrison worked at kensho", "bears like to eat honey"],
embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()
retriever.get_relevant_documents("where did harrison work?")
template = """Answer the question based only on the following context:
{context}
Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
chain = RunnableMap({
"context": lambda x: retriever.get_relevant_documents(x["question"]),
"question": lambda x: x["question"]
}) | prompt | model | output_parser
chain.invoke({"question": "where did harrison work?"})
LCEL 을 OpenAI 와 Binding 하는 방법:
functions = [
{
"name": "weather_search",
"description": "Search for weather given an airport code",
"parameters": {
"type": "object",
"properties": {
"airport_code": {
"type": "string",
"description": "The airport code to get the weather for"
},
},
"required": ["airport_code"]
}
}
]
prompt = ChatPromptTemplate.from_messages(
[
("human", "{input}")
]
)
model = ChatOpenAI(temperature=0).bind(functions=functions)
runnable = prompt | model
runnable.invoke({"input": "what is the weather in sf"})
이런식으로 출력이 될거임.
AIMessage(content='', additional_kwargs={'function_call': {'name': 'weather_search', 'arguments': '{"airport_code":"SFO"}'}})
여러가지 Functions 을 바인딩 하는 것도 가능함.
functions = [
{
"name": "weather_search",
"description": "Search for weather given an airport code",
"parameters": {
"type": "object",
"properties": {
"airport_code": {
"type": "string",
"description": "The airport code to get the weather for"
},
},
"required": ["airport_code"]
}
},
{
"name": "sports_search",
"description": "Search for news of recent sport events",
"parameters": {
"type": "object",
"properties": {
"team_name": {
"type": "string",
"description": "The sports team to search for"
},
},
"required": ["team_name"]
}
}
]
model = model.bind(functions=functions)
runnable = prompt | model
runnable.invoke({"input": "how did the patriots do yesterday?"})
출력 결과:
AIMessage(content='', additional_kwargs={'function_call': {'name': 'sports_search', 'arguments': '{"team_name":"New England Patriots"}'}})
LCEL 의 기능으로 Fallback 을 넣는 것도 가능함. LLM 이 RateLimiter 에 걸렸거나, 네트워크 문제로 답변을 하지 못할 때 대체 LLM 을 넣어서 답변을 하도록 만드는 방법임.
from langchain.llms import OpenAI
import json
simple_model = OpenAI(
temperature=0,
max_tokens=1000,
model="gpt-3.5-turbo-instruct"
)
simple_chain = simple_model | json.loads
challenge = "write three poems in a json blob, where each poem is a json blob of a title, author, and first line"
simple_model.invoke(challenge)
model = ChatOpenAI(temperature=0)
chain = model | StrOutputParser() | json.loads
final_chain = simple_chain.with_fallbacks([chain])
final_chain.invoke(challenge)
다음은 LCEL 의 기능인 Batch, Asnyc, Strema 처리에 대해 간략하게 어떤 식으로 코드를 짜면 되는지 살펴보자.
prompt = ChatPromptTemplate.from_template(
"Tell me a short joke about {topic}"
)
model = ChatOpenAI()
output_parser = StrOutputParser()
chain = prompt | model | output_parser
# Batch
chain.batch([{"topic": "bears"}, {"topic": "frogs"}])
# Stream
for t in chain.stream({"topic": "bears"}):
print(t)
# Async
response = await chain.ainvoke({"topic": "bears"})
response
3. OpenAI Function Calling in LangChain
OpenAI function calling 을 더 쉽게 사용하는 법은 Pandatic 이라는 Validation 툴과 같이 쓰는거임.
이걸 통해서 함수 호출에 대한 설명을 타입이 있는 코드로 쉽게 만들 수 있음.
예시를 보면 다믕과 같다:
- WeatherSearch 함수에 대한 정의를 LLM 에게 보내기 위해서 WeatherSearch 라는 클래스를 만든 것.
import os
import openai
from langchain.utils.openai_functions import convert_pydantic_to_openai_function
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv()) # read local .env file
openai.api_key = os.environ['OPENAI_API_KEY']
from typing import List
from pydantic import BaseModel, Field
class WeatherSearch(BaseModel):
"""Call this with an airport code to get the weather at that airport"""
airport_code: str = Field(description="airport code to get weather for")
weather_function = convert_pydantic_to_openai_function(WeatherSearch)
weather_function
함수 정의에 대한 결과를 보자.
{'name': 'WeatherSearch',
'description': 'Call this with an airport code to get the weather at that airport',
'parameters': {'title': 'WeatherSearch',
'description': 'Call this with an airport code to get the weather at that airport',
'type': 'object',
'properties': {'airport_code': {'title': 'Airport Code',
'description': 'airport code to get weather for',
'type': 'string'}},
'required': ['airport_code']}}
class WeatherSearch1(BaseModel):
airport_code: str = Field(description="airport code to get weather for")
다음은 이제 이 함수 정의를 이용해서 OpenAI function calling 을 이용해보자.
from langchain.chat_models import ChatOpenAI
model = ChatOpenAI()
model.invoke("what is the weather in SF today?", functions=[weather_function])
출력은 다음과 같다:
AIMessage(content='', additional_kwargs={'function_call': {'name': 'WeatherSearch', 'arguments': '{"airport_code":"SFO"}'}})
다음은 이 function calling 을 LCEL 에서 사용하는 예제를 보자.
from langchain.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant"),
("user", "{input}")
])
chain = prompt | model_with_function
chain.invoke({"input": "what is the weather in sf?"})
다음은 여러개의 함수를 function calling 에 사용하는 예제:
class ArtistSearch(BaseModel):
"""Call this to get the names of songs by a particular artist"""
artist_name: str = Field(description="name of artist to look up")
n: int = Field(description="number of results")
functions = [
convert_pydantic_to_openai_function(WeatherSearch),
convert_pydantic_to_openai_function(ArtistSearch),
]
model_with_functions = model.bind(functions=functions)
model_with_functions.invoke("what is the weather in sf?")
model_with_functions.invoke("what are three songs by taylor swift?")
4. Tagging and Extraction
Tagging 과 Extraction 은 Function calling 을 위해서 LLM 이 Unstructured text 에서 함수 호출을 위한 파라미터 데이터를 구조화된 형식으로 만들어 주는 거임.
다만 이 방식에 따라서 Tagging 과 Extraction 은 약간의 차이가 있는데 Tagging 은 입력으로 들어온 비구조화 데이터에서 추론을 통해 데이터를 generate 하는 반면에, Extraction 은 추출을 한다는 점에서 동작 방식이 약간 다름.
사용 방식의 예시를 보자. 먼저 Tagging 부터.
이전에 살펴본 Pandatic 을 이용해서 만들어진 함수 정의는 다음과 같음.
- Pandatic 을 만들 때 들어올 수 있는 값에 예시를 같이 넣어주면 더 좋음.
import os
import openai
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv()) # read local .env file
openai.api_key = os.environ['OPENAI_API_KEY']
from typing import List
from pydantic import BaseModel, Field
from langchain.utils.openai_functions import convert_pydantic_to_openai_function
class Tagging(BaseModel):
"""Tag the piece of text with particular info."""
sentiment: str = Field(description="sentiment of text, should be `pos`, `neg`, or `neutral`")
language: str = Field(description="language of text (should be ISO 639-1 code)")
convert_pydantic_to_openai_function(Tagging)
출력 예시
{'name': 'Tagging',
'description': 'Tag the piece of text with particular info.',
'parameters': {'title': 'Tagging',
'description': 'Tag the piece of text with particular info.',
'type': 'object',
'properties': {'sentiment': {'title': 'Sentiment',
'description': 'sentiment of text, should be `pos`, `neg`, or `neutral`',
'type': 'string'},
'language': {'title': 'Language',
'description': 'language of text (should be ISO 639-1 code)',
'type': 'string'}},
'required': ['sentiment', 'language']}}
다음은 이제 Tagging 을 사용하는 코드:
from langchain.prompts import ChatPromptTemplate
from langchain.chat_models import ChatOpenAI
model = ChatOpenAI(temperature=0)
tagging_functions = [convert_pydantic_to_openai_function(Tagging)]
prompt = ChatPromptTemplate.from_messages([
("system", "Think carefully, and then tag the text as instructed"),
("user", "{input}")
])
model_with_functions = model.bind(
functions=tagging_functions,
function_call={"name": "Tagging"}
)
tagging_chain = prompt | model_with_functions
tagging_chain.invoke({"input": "I love langchain"})
출력 결과는 다음과 같다:
AIMessage(content='', additional_kwargs={'function_call': {'name': 'Tagging', 'arguments': '{"sentiment":"pos","language":"en"}'}})
다음은 이걸 JsonOutputFunctionsParser 를 이용해서 JSON 출력으로도 만들어보자.
from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser
tagging_chain = prompt | model_with_functions | JsonOutputFunctionsParser()
tagging_chain.invoke({"input": "non mi piace questo cibo"})
출력 결과는 다음과 같다:
{'sentiment': 'neg', 'language': 'it'}
이제는 Extraction 을 하는 코드를 보자.
from typing import Optional
class Person(BaseModel):
"""Information about a person."""
name: str = Field(description="person's name")
age: Optional[int] = Field(description="person's age")
class Information(BaseModel):
"""Information to extract."""
people: List[Person] = Field(description="List of info about people")
convert_pydantic_to_openai_function(Information)
extraction_functions = [convert_pydantic_to_openai_function(Information)]
extraction_model = model.bind(functions=extraction_functions, function_call={"name": "Information"})
extraction_chain = prompt | extraction_model
extraction_chain.invoke({"input": "Joe is 30, his mom is Martha"})
출력 결과는 다음과 같음.
AIMessage(content='', additional_kwargs={'function_call': {'name': 'Information', 'arguments': '{"people":[{"name":"Joe","age":30},{"name":"Martha"}]}'}})
이번에도 JsonOutputFunctionsParser 를 이용해서 데이터를 추출해보자.
extraction_chain = prompt | extraction_model | JsonOutputFunctionsParser()
extraction_chain.invoke({"input": "Joe is 30, his mom is Martha"})
JSON 출력은 다음과 같음:
{'people': [{'name': 'Joe', 'age': 30}, {'name': 'Martha'}]}
다음은 JsonKeyOutputFunctionsParser 를 이용해서 people 키를 추출해서 데이터를 확인해보자.
from langchain.output_parsers.openai_functions import JsonKeyOutputFunctionsParser
extraction_chain = prompt | extraction_model | JsonKeyOutputFunctionsParser(key_name="people")
extraction_chain.invoke({"input": "Joe is 30, his mom is Martha"})
출력 결과는 다음과 같음:
[{'name': 'Joe', 'age': 30}, {'name': 'Martha'}]
다음은 실전 예제로 해보자. 이건 Blog Post 에서 Extract 하는 코드임.
from langchain.document_loaders import WebBaseLoader
loader = WebBaseLoader("https://lilianweng.github.io/posts/2023-06-23-agent/")
documents = loader.load()
doc = documents[0]
page_content = doc.page_content[:10000]
class Overview(BaseModel):
"""Overview of a section of text."""
summary: str = Field(description="Provide a concise summary of the content.")
language: str = Field(description="Provide the language that the content is written in.")
keywords: str = Field(description="Provide keywords related to the content.")
overview_tagging_function = [
convert_pydantic_to_openai_function(Overview)
]
tagging_model = model.bind(
functions=overview_tagging_function,
function_call={"name":"Overview"}
)
tagging_chain = prompt | tagging_model | JsonOutputFunctionsParser()
tagging_chain.invoke({"input": page_content})
출력 결과는 다음과 같음:
{'summary': 'This text discusses the concept of building autonomous agents powered by LLM (large language model) as the core controller. It covers components such as planning, memory, and tool use, along with challenges and references to related studies.',
'language': 'English',
'keywords': 'LLM, autonomous agents, planning, memory, tool use, self-reflection, task decomposition, challenges, references'}
다음으로 저자의 이름과 제목을 extraction 하는 예제도 보자.
class Paper(BaseModel):
"""Information about papers mentioned."""
title: str
author: Optional[str]
class Info(BaseModel):
"""Information to extract"""
papers: List[Paper]
paper_extraction_function = [
convert_pydantic_to_openai_function(Info)
]
extraction_model = model.bind(
functions=paper_extraction_function,
function_call={"name":"Info"}
)
template = """A article will be passed to you. Extract from it all papers that are mentioned by this article.
Do not extract the name of the article itself. If no papers are mentioned that's fine - you don't need to extract any! Just return an empty list.
Do not make up or guess ANY extra information. Only extract what exactly is in the text."""
prompt = ChatPromptTemplate.from_messages([
("system", template),
("human", "{input}")
])
extraction_chain = prompt | extraction_model | JsonKeyOutputFunctionsParser(key_name="papers")
extraction_chain.invoke({"input": page_content})
출력 결과는 다음과 같다:
[{'title': 'Chain of thought (CoT; Wei et al. 2022)'},
{'title': 'Tree of Thoughts (Yao et al. 2023)'},
{'title': 'LLM+P (Liu et al. 2023)'},
{'title': 'ReAct (Yao et al. 2023)'},
{'title': 'Reflexion (Shinn & Labash 2023)'},
{'title': 'Chain of Hindsight (CoH; Liu et al. 2023)'},
{'title': 'Algorithm Distillation (AD; Laskin et al. 2023)'}]
다음은 Web Document 에 있는 내용들을 split 해서 extract 하는 코드
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.schema.runnable import RunnableLambda
text_splitter = RecursiveCharacterTextSplitter(chunk_overlap=0)
def flatten(matrix):
flat_list = []
for row in matrix:
flat_list += row
return flat_list
prep = RunnableLambda(
lambda x: [{"input": doc} for doc in text_splitter.split_text(x)]
)
chain = prep | extraction_chain.map() | flatten
chain.invoke(doc.page_content)
출력 결과는 다음과 같다.
[{'title': 'AutoGPT'},
{'title': 'GPT-Engineer'},
{'title': 'BabyAGI'},
{'title': 'Chain of thought'},
{'title': 'Tree of Thoughts'},
{'title': 'LLM+P'},
{'title': 'ReAct'},
{'title': 'Reflexion'},
{'title': 'Chain of Hindsight (CoH; Liu et al. 2023)'},
{'title': 'Algorithm Distillation (AD; Laskin et al. 2023)'},
{'title': 'Laskin et al. 2023'},
{'title': 'Miller 1956'},
{'title': 'Duan et al. 2017'},
{'title': 'Google Blog'},
{'title': 'ann-benchmarks.com'},
{'title': 'MRKL (Karpas et al. 2022)'},
{'title': 'TALM (Tool Augmented Language Models; Parisi et al. 2022)'},
{'title': 'Toolformer (Schick et al. 2023)'},
{'title': 'HuggingGPT (Shen et al. 2023)'},
{'title': 'API-Bank', 'author': 'Li et al. 2023'},
{'title': 'ChemCrow', 'author': 'Bran et al. 2023'},
{'title': 'Boiko et al. (2023)'},
{'title': 'Generative Agents Simulation (Park, et al. 2023)'},
{'title': 'Park et al. 2023'},
{'title': 'Sample Paper 1', 'author': 'Author A'},
{'title': 'Sample Paper 2', 'author': 'Author B'},
{'title': 'Paper A', 'author': 'Author A'},
{'title': 'Paper B', 'author': 'Author B'},
{'title': 'Chain of thought prompting elicits reasoning in large language models.'},
{'title': 'Tree of Thoughts: Deliberate Problem Solving with Large Language Models'},
{'title': 'Chain of Hindsight Aligns Language Models with Feedback'},
{'title': 'LLM+P: Empowering Large Language Models with Optimal Planning Proficiency'},
{'title': 'ReAct: Synergizing reasoning and acting in language models'},
{'title': 'Reflexion: an autonomous agent with dynamic memory and self-reflection'},
{'title': 'In-context Reinforcement Learning with Algorithm Distillation'},
{'title': 'MRKL Systems A modular, neuro-symbolic architecture that combines large language models, external knowledge sources and discrete reasoning'},
{'title': 'Webgpt: Browser-assisted question-answering with human feedback'},
{'title': 'TALM: Tool Augmented Language Models'},
{'title': 'Toolformer: Language Models Can Teach Themselves to Use Tools'},
{'title': 'API-Bank: A Benchmark for Tool-Augmented LLMs'},
{'title': 'HuggingGPT: Solving AI Tasks with ChatGPT and its Friends in HuggingFace'},
{'title': 'ChemCrow: Augmenting large-language models with chemistry tools'},
{'title': 'Emergent autonomous scientific research capabilities of large language models'},
{'title': 'Generative Agents: Interactive Simulacra of Human Behavior'}]
5. Tools and Routing
여기서는 두 가지 작업을 해볼거임.
- Agent 가 사용하는 Tool 을 만들고 이것을 OpenAI Function calling 으로 적용해서 사용할 수 있도록 하기
- LLM 이 호출할 수 있는 여러가지 함수들을 제공했을 때 실제로 함수를 호출해보도록 하는 것. (+ 함수 호출이 필요하지 않다고 LLM 이 판단했고 LLM 이 답변을 하는 거라면 답변을 출력하도록)
하나씩 해보자. 먼저 Custom Tool 을 만드는 코드는 다음과 같다:
import os
import openai
from langchain.agents import tool
from pydantic import BaseModel, Field
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv()) # read local .env file
openai.api_key = os.environ['OPENAI_API_KEY']
class SearchInput(BaseModel):
query: str = Field(description="Thing to search for")
@tool(args_schema=SearchInput)
def search(query: str) -> str:
"""Search for the weather online."""
return "42f"
search.name
search.description
search.args
search.run("sf")
이번에는 이전에 본 예제인 날씨 API 를 호출하는 것도 도구로 만들어보자.
import requests
from pydantic import BaseModel, Field
import datetime
# Define the input schema
class OpenMeteoInput(BaseModel):
latitude: float = Field(..., description="Latitude of the location to fetch weather data for")
longitude: float = Field(..., description="Longitude of the location to fetch weather data for")
@tool(args_schema=OpenMeteoInput)
def get_current_temperature(latitude: float, longitude: float) -> dict:
"""Fetch current temperature for given coordinates."""
BASE_URL = "https://api.open-meteo.com/v1/forecast"
# Parameters for the request
params = {
'latitude': latitude,
'longitude': longitude,
'hourly': 'temperature_2m',
'forecast_days': 1,
}
# Make the request
response = requests.get(BASE_URL, params=params)
if response.status_code == 200:
results = response.json()
else:
raise Exception(f"API Request failed with status code: {response.status_code}")
current_utc_time = datetime.datetime.utcnow()
time_list = [datetime.datetime.fromisoformat(time_str.replace('Z', '+00:00')) for time_str in results['hourly']['time']]
temperature_list = results['hourly']['temperature_2m']
closest_time_index = min(range(len(time_list)), key=lambda i: abs(time_list[i] - current_utc_time))
current_temperature = temperature_list[closest_time_index]
return f'The current temperature is {current_temperature}°C'
get_current_temperature.name
get_current_temperature.description
get_current_temperature.args
get_current_temperature({"latitude": 13, "longitude": 14})
이번에는 만든 Tool 을 OpenAI Function calling 을 위해 Function definition 을 만드는 코드도 보자.
from langchain.tools.render import format_tool_to_openai_function
format_tool_to_openai_function(get_current_temperature)
출력은 다음과 같을 것.
{'name': 'get_current_temperature',
'description': 'get_current_temperature(latitude: float, longitude: float) -> dict - Fetch current temperature for given coordinates.',
'parameters': {'title': 'OpenMeteoInput',
'type': 'object',
'properties': {'latitude': {'title': 'Latitude',
'description': 'Latitude of the location to fetch weather data for',
'type': 'number'},
'longitude': {'title': 'Longitude',
'description': 'Longitude of the location to fetch weather data for',
'type': 'number'}},
'required': ['latitude', 'longitude']}}
다음은 Wikipedia 검색도 Tool 로 만들어보자.
import wikipedia
@tool
def search_wikipedia(query: str) -> str:
"""Run Wikipedia search and get page summaries."""
page_titles = wikipedia.search(query)
summaries = []
for page_title in page_titles[: 3]:
try:
wiki_page = wikipedia.page(title=page_title, auto_suggest=False)
summaries.append(f"Page: {page_title}\nSummary: {wiki_page.summary}")
except (
self.wiki_client.exceptions.PageError,
self.wiki_client.exceptions.DisambiguationError,
):
pass
if not summaries:
return "No good Wikipedia Search Result was found"
return "\n\n".join(summaries)
search_wikipedia.name
search_wikipedia.description
format_tool_to_openai_function(search_wikipedia)
search_wikipedia({"query": "langchain"})
다음은 API 명세서를 바탕으로 이 API 를 OpenAI Function calling 으로 호출할 수 있도록 만드는 코드도 보자:
from langchain.chains.openai_functions.openapi import openapi_spec_to_openai_fn
from langchain.utilities.openapi import OpenAPISpec
from langchain.chat_models import ChatOpenAI
text = """
{
"openapi": "3.0.0",
"info": {
"version": "1.0.0",
"title": "Swagger Petstore",
"license": {
"name": "MIT"
}
},
"servers": [
{
"url": "http://petstore.swagger.io/v1"
}
],
"paths": {
"/pets": {
"get": {
"summary": "List all pets",
"operationId": "listPets",
"tags": [
"pets"
],
"parameters": [
{
"name": "limit",
"in": "query",
"description": "How many items to return at one time (max 100)",
"required": false,
"schema": {
"type": "integer",
"maximum": 100,
"format": "int32"
}
}
],
"responses": {
"200": {
"description": "A paged array of pets",
"headers": {
"x-next": {
"description": "A link to the next page of responses",
"schema": {
"type": "string"
}
}
},
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Pets"
}
}
}
},
"default": {
"description": "unexpected error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
}
},
"post": {
"summary": "Create a pet",
"operationId": "createPets",
"tags": [
"pets"
],
"responses": {
"201": {
"description": "Null response"
},
"default": {
"description": "unexpected error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
}
}
},
"/pets/{petId}": {
"get": {
"summary": "Info for a specific pet",
"operationId": "showPetById",
"tags": [
"pets"
],
"parameters": [
{
"name": "petId",
"in": "path",
"required": true,
"description": "The id of the pet to retrieve",
"schema": {
"type": "string"
}
}
],
"responses": {
"200": {
"description": "Expected response to a valid request",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Pet"
}
}
}
},
"default": {
"description": "unexpected error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Error"
}
}
}
}
}
}
}
},
"components": {
"schemas": {
"Pet": {
"type": "object",
"required": [
"id",
"name"
],
"properties": {
"id": {
"type": "integer",
"format": "int64"
},
"name": {
"type": "string"
},
"tag": {
"type": "string"
}
}
},
"Pets": {
"type": "array",
"maxItems": 100,
"items": {
"$ref": "#/components/schemas/Pet"
}
},
"Error": {
"type": "object",
"required": [
"code",
"message"
],
"properties": {
"code": {
"type": "integer",
"format": "int32"
},
"message": {
"type": "string"
}
}
}
}
}
}
"""
spec = OpenAPISpec.from_text(text)
pet_openai_functions, pet_callables = openapi_spec_to_openai_fn(spec)
pet_openai_functions
OpenAPI 명세서 (pet_openai_functions) 에 Function Deifinition 을 보면 다음과 같을 것:
[{'name': 'listPets',
'description': 'List all pets',
'parameters': {'type': 'object',
'properties': {'params': {'type': 'object',
'properties': {'limit': {'type': 'integer',
'maximum': 100.0,
'schema_format': 'int32',
'description': 'How many items to return at one time (max 100)'}},
'required': []}}}},
{'name': 'createPets',
'description': 'Create a pet',
'parameters': {'type': 'object', 'properties': {}}},
{'name': 'showPetById',
'description': 'Info for a specific pet',
'parameters': {'type': 'object',
'properties': {'path_params': {'type': 'object',
'properties': {'petId': {'type': 'string',
'description': 'The id of the pet to retrieve'}},
'required': ['petId']}}}}]
다음은 이 API 를 Function Calling 으로 잘 사용할 수 있는지 보자.
model = ChatOpenAI(temperature=0).bind(functions=pet_openai_functions)
model.invoke("what are three pets names")
출력 결과는 다음과 같다:
AIMessage(content='', additional_kwargs={'function_call': {'name': 'listPets', 'arguments': '{"params":{"limit":3}}'}})
AIMessage 를 보면 어떤 함수를 호출할 수 있고, 매개변수가 무엇인지에 대한 설명이 적혀있다. 이것들을 잘 파싱해서 실제로 함수를 호출할 수 있도록 하는 Router 를 만들어서 호출해보자. (LangChain 에서는 이런 Router 구현은 제공해줄 거임)
- OpenAIFunctionsAgentOutputParser 로 결과를 파싱하면, result.tool 에 함수의 이름이 있을거고 result.tool_input 에 함수를 호출할 수 있는 매개변수가 들어있을거임.
from langchain.prompts import ChatPromptTemplate
from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
functions = [
format_tool_to_openai_function(f) for f in [
search_wikipedia, get_current_temperature
]
]
prompt = ChatPromptTemplate.from_messages([
("system", "You are helpful but sassy assistant"),
("user", "{input}"),
])
model = ChatOpenAI(temperature=0).bind(functions=functions)
chain = prompt | model | OpenAIFunctionsAgentOutputParser()
result = chain.invoke({"input": "what is the weather in sf right now"})
result.tool
result.tool_input
get_current_temperature(result.tool_input)
만약 LLM 이 툴 호출을 하지 않는다고 판단한다면 텍스트 답변이 올거임.
result = chain.invoke({"input": "hi!"})
type(result)
result.return_values
이제 router 함수를 만들어서 함수 호출까지 되겍 만들어보자.
from langchain.schema.agent import AgentFinish
def route(result):
if isinstance(result, AgentFinish):
return result.return_values['output']
else:
tools = {
"search_wikipedia": search_wikipedia,
"get_current_temperature": get_current_temperature,
}
return tools[result.tool].run(result.tool_input)
chain = prompt | model | OpenAIFunctionsAgentOutputParser() | route
result = chain.invoke({"input": "What is the weather in san francisco right now?"})
result
함수 호출까지 되었으므로 출력은 다음과 같을거임.
'The current temperature is 13.7°C'
6. Conversational Agent
여기서는 OpenAI Function calling 을 이용해서 Agent 를 만들어보자.
Agent 는 LLM 에게 여러가지 도구가 합쳐진 것이고 문제를 해결하기 위해 무엇을 해야하는지 추론하고 필요하다면 각 도구를 사용해서 해결한다.
사용자가 Agent 에게 질문을 주었을 때 Agent 는 다음과 같이 동작한다:
- 1) 문제를 해결하기 위해 도구를 선택한다.
- 2) 도구를 실행한 결과를 관찰하고 기억한다.
- 3) 문제를 해결할 때까지 1-2 과정을 반복하며, 문제를 해결할 수 있겠다고 생각하면 멈춘다. (이건 종료 조건을 만나야한다. AgentFinish 가 대표적인 종료 조건이며, 원하는 경우 종료 조건을 커스터마이징 할 수 있다. 대표적인 종료 조건이 반복 횟수임)
여기 Agent 에다가 ReAct 프롬포트 기법을 이용하면 우리가 알고 있는 대표적인 Conservational Agent 가 된다.
그리고 이런 Agent 를 실행하기 편하도록 AgentExecutor 라는 것도 LangChain 에서는 제공해줌. 이걸 이용하면 추상화해서 Agent 를 만들 수 있다.
먼저 Agent 가 어떻게 동작하는지, 즉 AgentExecutor 가 어떻게 동작하는지 살펴보자.
Agent 가 사용할 도구들 부터 선언하자.
import os
import openai
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv()) # read local .env file
openai.api_key = os.environ['OPENAI_API_KEY']
from langchain.tools import tool
import requests
from pydantic import BaseModel, Field
import datetime
# Define the input schema
class OpenMeteoInput(BaseModel):
latitude: float = Field(..., description="Latitude of the location to fetch weather data for")
longitude: float = Field(..., description="Longitude of the location to fetch weather data for")
@tool(args_schema=OpenMeteoInput)
def get_current_temperature(latitude: float, longitude: float) -> dict:
"""Fetch current temperature for given coordinates."""
BASE_URL = "https://api.open-meteo.com/v1/forecast"
# Parameters for the request
params = {
'latitude': latitude,
'longitude': longitude,
'hourly': 'temperature_2m',
'forecast_days': 1,
}
# Make the request
response = requests.get(BASE_URL, params=params)
if response.status_code == 200:
results = response.json()
else:
raise Exception(f"API Request failed with status code: {response.status_code}")
current_utc_time = datetime.datetime.utcnow()
time_list = [datetime.datetime.fromisoformat(time_str.replace('Z', '+00:00')) for time_str in results['hourly']['time']]
temperature_list = results['hourly']['temperature_2m']
closest_time_index = min(range(len(time_list)), key=lambda i: abs(time_list[i] - current_utc_time))
current_temperature = temperature_list[closest_time_index]
return f'The current temperature is {current_temperature}°C'
import wikipedia
@tool
def search_wikipedia(query: str) -> str:
"""Run Wikipedia search and get page summaries."""
page_titles = wikipedia.search(query)
summaries = []
for page_title in page_titles[: 3]:
try:
wiki_page = wikipedia.page(title=page_title, auto_suggest=False)
summaries.append(f"Page: {page_title}\nSummary: {wiki_page.summary}")
except (
self.wiki_client.exceptions.PageError,
self.wiki_client.exceptions.DisambiguationError,
):
pass
if not summaries:
return "No good Wikipedia Search Result was found"
return "\n\n".join(summaries)
tools = [get_current_temperature, search_wikipedia]
우리가 이전에 살펴본 OpenAI Function caling 은 도구를 함수로 만들어놓고 이 함수를 호출할 수 있도록 function deifinition 으로 만들어서 LLM 에게 전달했었다.
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.tools.render import format_tool_to_openai_function
from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser
functions = [format_tool_to_openai_function(f) for f in tools]
model = ChatOpenAI(temperature=0).bind(functions=functions)
prompt = ChatPromptTemplate.from_messages([
("system", "You are helpful but sassy assistant"),
("user", "{input}"),
])
chain = prompt | model | OpenAIFunctionsAgentOutputParser()
result = chain.invoke({"input": "what is the weather is sf?"})
result.tool
result.tool_input
여기에다가 이제 각 도구를 실행한 결과를 기억할 수 있도록 하고, 반복문을 통해서 LLM 에게 문제를 해결해줘 라고 질문하면 대략적으로 Agent 와 가까워진다.
이렇게 하면 LLM 은 문제를 해결하기 위해서 도구를 선택했고, 도구를 실행한 결과를 종합해서, 답변을 내놓게 될거임.
도구를 실행한 결과는 agent_scratchpad 에 다음처럼 기록될거임.
from langchain.prompts import MessagesPlaceholder
prompt = ChatPromptTemplate.from_messages([
("system", "You are helpful but sassy assistant"),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
chain = prompt | model | OpenAIFunctionsAgentOutputParser()
result1 = chain.invoke({
"input": "what is the weather is sf?",
"agent_scratchpad": []
})
# 'get_current_temperature'
result1.tool
observation = get_current_temperature(result1.tool_input)
# 'The current temperature is 12.8°C'
observation
from langchain.agents.format_scratchpad import format_to_openai_functions
# [AIMessage(content='', additional_kwargs={'function_call': {'name': 'get_current_temperature', 'arguments': '{"latitude":37.7749,"longitude":-122.4194}'}})]
result1.message_log
format_to_openai_functions([(result1, observation), ])
result2 = chain.invoke({
"input": "what is the weather is sf?",
"agent_scratchpad": format_to_openai_functions([(result1, observation)])
})
format_to_openai_functions 의 결과는 다음과 같다.
[AIMessage(content='', additional_kwargs={'function_call': {'name': 'get_current_temperature', 'arguments': '{"latitude":37.7749,"longitude":-122.4194}'}}),
FunctionMessage(content='The current temperature is 12.8°C', name='get_current_temperature')]
다음은 Agent 내부 동작을 함수로 만든 코드임. 이걸 알면 어떻게 동작하는지 알 것.
from langchain.schema.agent import AgentFinish
from langchain.schema.runnable import RunnablePassthrough
agent_chain = RunnablePassthrough.assign(
agent_scratchpad= lambda x: format_to_openai_functions(x["intermediate_steps"])
) | chain
def run_agent(user_input):
intermediate_steps = []
while True:
result = agent_chain.invoke({
"input": user_input,
"intermediate_steps": intermediate_steps
})
if isinstance(result, AgentFinish):
return result
tool = {
"search_wikipedia": search_wikipedia,
"get_current_temperature": get_current_temperature,
}[result.tool]
observation = tool.run(result.tool_input)
intermediate_steps.append((result, observation))
run_agent("what is the weather is sf?")
이제 AgentExectuor 를 보자.
AgentExecutor 는 다음과 같이 사용할 수 있음.
from langchain.agents import AgentExecutor
agent_executor = AgentExecutor(agent=agent_chain, tools=tools, verbose=True)
agent_executor.invoke({"input": "what is langchain?"})
Agent 는 다음과 같이 단계적으로 접근해서 문제를 해결할거임.
- 1) search_wikipedia 도구를 써서 LangChain 에 대해 위키피디아에 검색한다.
- 2) 가져온 정보를 프롬포트에 넣어서 다시 LLM 에게 LangChain 이 뭐냐고 질문한다.
- 3) LLM 은 이제 LangChain 이 무엇인지에 대해서 답변한다.
> Entering new AgentExecutor chain...
Invoking: `search_wikipedia` with `{'query': 'Langchain'}`
Page: LangChain
Summary: LangChain is a framework designed to simplify the creation of applications using large language models (LLMs). As a language model integration framework, LangChain's use-cases largely overlap with those of language models in general, including document analysis and summarization, chatbots, and code analysis.
Page: DataStax
Summary: DataStax, Inc. is a real-time data for AI company based in Santa Clara, California. Its product Astra DB is a cloud database-as-a-service based on Apache Cassandra. DataStax also offers DataStax Enterprise (DSE), an on-premises database built on Apache Cassandra, and Astra Streaming, a messaging and event streaming cloud service based on Apache Pulsar. As of June 2022, the company has roughly 800 customers distributed in over 50 countries.
Page: Sentence embedding
Summary: In natural language processing, a sentence embedding refers to a numeric representation of a sentence in the form of a vector of real numbers which encodes meaningful semantic information.
State of the art embeddings are based on the learned hidden layer representation of dedicated sentence transformer models. BERT pioneered an approach involving the use of a dedicated [CLS] token prepended to the beginning of each sentence inputted into the model; the final hidden state vector of this token encodes information about the sentence and can be fine-tuned for use in sentence classification tasks. In practice however, BERT's sentence embedding with the [CLS] token achieves poor performance, often worse than simply averaging non-contextual word embeddings. SBERT later achieved superior sentence embedding performance by fine tuning BERT's [CLS] token embeddings through the usage of a siamese neural network architecture on the SNLI dataset.
Other approaches are loosely based on the idea of distributional semantics applied to sentences. Skip-Thought trains an encoder-decoder structure for the task of neighboring sentences predictions. Though this has been shown to achieve worse performance than approaches such as InferSent or SBERT.
An alternative direction is to aggregate word embeddings, such as those returned by Word2vec, into sentence embeddings. The most straightforward approach is to simply compute the average of word vectors, known as continuous bag-of-words (CBOW). However, more elaborate solutions based on word vector quantization have also been proposed. One such approach is the vector of locally aggregated word embeddings (VLAWE), which demonstrated performance improvements in downstream text classification tasks.LangChain is a framework designed to simplify the creation of applications using large language models (LLMs). It is a language model integration framework with use-cases including document analysis and summarization, chatbots, and code analysis.
> Finished chain.
{'input': 'what is langchain?',
'output': 'LangChain is a framework designed to simplify the creation of applications using large language models (LLMs). It is a language model integration framework with use-cases including document analysis and summarization, chatbots, and code analysis.'}
AgentExecutor 에 사용자의 Chathistory 정보까지 기억하려면 다음과 같이 ConservationBufferMemory 를 이용하면 된다.
prompt = ChatPromptTemplate.from_messages([
("system", "You are helpful but sassy assistant"),
MessagesPlaceholder(variable_name="chat_history"),
("user", "{input}"),
MessagesPlaceholder(variable_name="agent_scratchpad")
])
agent_chain = RunnablePassthrough.assign(
agent_scratchpad= lambda x: format_to_openai_functions(x["intermediate_steps"])
) | prompt | model | OpenAIFunctionsAgentOutputParser()
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory(return_messages=True,memory_key="chat_history")
agent_executor = AgentExecutor(agent=agent_chain, tools=tools, verbose=True, memory=memory)
agent_executor.invoke({"input": "my name is bob"})
Conclusion
- LLM App Serving 할 때 Fallback 을 넣는 것
- Chain 과 Agent 각각 언제 사용해야할까? 동적인 Workflow 가 있다면, Agent 를 쓰면 되고 정적인 Workflow 에 대해서는 Chain 을 이용하면 된다.
- Function calling 과 Tool 을 통해서 LLM 은 외부 세상의 도구를 이용할 수 있다. (SQL, API, Search 등)
- Function calling 으로 Unstructed text 에서 데이터를 뽑아내거나 추론으로 생성할 수 있다.
- Function calling 의 매개변수 선언에는 예시를 넣어주면 좋다.
- Function calling 을 사용하려면 강제성을 띄도록 설정하는 것과 temperature 를 0으로 설정하면 된다.
'Generative AI' 카테고리의 다른 글
| METAGPT: META PROGRAMMING FOR AMULTI-AGENT COLLABORATIVE FRAMEWORK (0) | 2024.06.20 |
|---|---|
| Building Agentic RAG with Llamaindex (0) | 2024.06.20 |
| What We’ve Learned From A Year of Building with LLMs (0) | 2024.06.11 |
| Building and Evaluating Advanced RAG (0) | 2024.06.10 |
| Fine-tuning Large Language Models (0) | 2024.05.31 |