함수 호출로 챗 애플리케이션 만들기
함수 호출로 챗 애플리케이션 만들기 (Function Calling)
OpenAI의 함수 호출(function calling) 기능으로 에이전트처럼 행동하는 챗 애플리케이션을 만들어 볼게요. Haystack 파이프라인을 함수 호출용 도구로 바꾸고, 여러 도구를 조합해 채팅이 가능한 앱까지 이어가는 흐름이에요.
출처: 공식문서
개요
OpenAI의 함수 호출은 LLM을 외부 도구와 연결해 줘요. API 호출에 함수 목록(tools)과 그 명세를 제공하면, 채팅 어시스턴트가 외부 API를 호출해 질문에 답하거나 텍스트에서 구조화된 정보를 뽑아낼 수 있어요.
이 튜토리얼에서는 먼저 OpenAIChatGenerator의 원시 도구 호출 메커니즘을 보고, 그다음 Haystack의 Agent 컴포넌트로 도구 호출 루프를 자동 실행해요.
설치와 환경
pip install haystack-ai
pip install sentence-transformers-haystack
import os
from getpass import getpass
if "OPENAI_API_KEY" not in os.environ:
os.environ["OPENAI_API_KEY"] = getpass("Enter OpenAI API key:")
OpenAIChatGenerator 살펴보기
OpenAIChatGenerator는 ChatMessage 객체 리스트로 통신해요. 시스템 역할과 사용자 역할 메시지를 만들어 전달해요.
from haystack.dataclasses import ChatMessage
from haystack.components.generators.chat import OpenAIChatGenerator
messages = [
ChatMessage.from_system("Always respond in German even if some input data is in other languages."),
ChatMessage.from_user("What's Natural Language Processing? Be brief."),
]
chat_generator = OpenAIChatGenerator(model="gpt-4o-mini")
chat_generator.run(messages=messages)
streaming_callback을 주면 응답을 스트리밍할 수 있어요.
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.generators.utils import print_streaming_chunk
chat_generator = OpenAIChatGenerator(model="gpt-4o-mini", streaming_callback=print_streaming_chunk)
response = chat_generator.run(messages=messages)
Haystack 파이프라인에서 함수 호출 도구 만들기
함수 호출을 쓰려면 OpenAIChatGenerator에 tools를 제공해야 해요. 여기서는 RAG 파이프라인을 도구 중 하나로 쓸 거라, 먼저 문서를 색인하고 RAG 파이프라인을 만듭니다.
문서 색인
작은 예제 데이터를 InMemoryDocumentStore에 임베딩과 함께 저장해요.
from haystack import Pipeline, Document
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.components.writers import DocumentWriter
from haystack_integrations.components.embedders.sentence_transformers import SentenceTransformersDocumentEmbedder
documents = [
Document(content="My name is Jean and I live in Paris."),
Document(content="My name is Mark and I live in Berlin."),
Document(content="My name is Giorgio and I live in Rome."),
Document(content="My name is Marta and I live in Madrid."),
Document(content="My name is Harry and I live in London."),
]
document_store = InMemoryDocumentStore()
indexing_pipeline = Pipeline()
indexing_pipeline.add_component(
instance=SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"), name="doc_embedder"
)
indexing_pipeline.add_component(instance=DocumentWriter(document_store=document_store), name="doc_writer")
indexing_pipeline.connect("doc_embedder.documents", "doc_writer.documents")
indexing_pipeline.run({"doc_embedder": {"documents": documents}})
RAG 파이프라인 만들기
리트리버와 프롬프트 빌더, LLM으로 검색 증강 생성 파이프라인을 만들고, 질문에 대한 답을 생성해요.
from haystack_integrations.components.embedders.sentence_transformers import SentenceTransformersTextEmbedder
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.builders import ChatPromptBuilder
from haystack.dataclasses import ChatMessage
from haystack.components.generators.chat import OpenAIChatGenerator
template = [
ChatMessage.from_system(
"""
Answer the questions based on the given context.
Context:
{% for document in documents %}
{{ document.content }}
{% endfor %}
Question: {{ question }}
Answer:
"""
)
]
rag_pipe = Pipeline()
rag_pipe.add_component("embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2"))
rag_pipe.add_component("retriever", InMemoryEmbeddingRetriever(document_store=document_store))
rag_pipe.add_component("prompt_builder", ChatPromptBuilder(template=template))
rag_pipe.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini"))
rag_pipe.connect("embedder.embedding", "retriever.query_embedding")
rag_pipe.connect("retriever", "prompt_builder.documents")
rag_pipe.connect("prompt_builder.prompt", "llm.messages")
파이프라인을 도구로 변환
rag_pipe.run 호출을 rag_pipeline_func 함수로 감싸고, Tool 객체로 만들어요. parameters(JSON 스키마), name, description을 정의해요.
from haystack.tools import Tool
def rag_pipeline_func(query: str):
result = rag_pipe.run({"embedder": {"text": query}, "prompt_builder": {"question": query}})
return {"reply": result["llm"]["replies"][0].text}
parameters = {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The query to use in the search. Infer this from the user's message. It should be a question or a statement",
}
},
"required": ["query"],
}
rag_pipeline_tool = Tool(
name="rag_pipeline_tool",
description="Get information about where people live",
parameters=parameters,
function=rag_pipeline_func,
)
함수에서 도구 만들기
create_tool_from_function을 쓰면 도구 파라미터의 JSON 스키마를 자동으로 추론해 줘요. Annotated 타입으로 파라미터를 설명하면 별도로 parameters를 정의할 필요가 없어요. 도시의 날씨를 돌려주는 도구를 이 방식으로 만들어요.
from typing import Annotated, Literal
from haystack.tools import create_tool_from_function
WEATHER_INFO = {
"Berlin": {"weather": "mostly sunny", "temperature": 7, "unit": "celsius"},
"Paris": {"weather": "mostly cloudy", "temperature": 8, "unit": "celsius"},
"Rome": {"weather": "sunny", "temperature": 14, "unit": "celsius"},
"Madrid": {"weather": "sunny", "temperature": 10, "unit": "celsius"},
"London": {"weather": "cloudy", "temperature": 9, "unit": "celsius"},
}
def get_weather(
city: Annotated[str, "the city for which to get the weather"] = "Berlin",
unit: Annotated[Literal["Celsius", "Fahrenheit"], "the unit for the temperature"] = "Celsius",
):
"""A simple function to get the current weather for a location."""
if city in WEATHER_INFO:
return WEATHER_INFO[city]
else:
return {"weather": "sunny", "temperature": 21.8, "unit": "fahrenheit"}
weather_tool = create_tool_from_function(get_weather)
도구를 가진 OpenAIChatGenerator 실행
도구 호출을 쓰려면 tools 리스트를 OpenAIChatGenerator에 넘겨요. 시스템 메시지로 모델에게 도구를 쓰라고 지시하고, 도구 호출이 필요한 질문을 사용자 메시지로 제공해요.
from haystack.dataclasses import ChatMessage
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.generators.utils import print_streaming_chunk
user_messages = [
ChatMessage.from_system(
"Use the tool that you're provided with. Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous."
),
ChatMessage.from_user("Can you tell me where Mark lives?"),
]
chat_generator = OpenAIChatGenerator(model="gpt-4o-mini", streaming_callback=print_streaming_chunk)
response = chat_generator.run(messages=user_messages, tools=[rag_pipeline_tool, weather_tool])
응답은 도구 이름과 인자를 담은 ToolCall 객체가 포함된 ChatMessage로 돌아와요. 간단히 말해 OpenAI API는 어떤 도구를 어떤 인자로 호출해야 하는지 알려주기만 하고, 도구 자체를 실행하지는 않아요.
Agent가 도구 호출 루프 실행하기
이 루프를 직접 작성하는 대신 Haystack의 Agent를 쓰면 도구 호출 루프를 자동으로 돌려줘요. LLM을 호출하고, 요청된 도구를 실행하고, 결과를 LLM에 다시 넣고, 모델이 일반 텍스트 답을 낼 때까지 반복해요.
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"),
tools=[rag_pipeline_tool, weather_tool],
system_prompt="Use the tools that you're provided with. Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous.",
)
result = agent.run(messages=[ChatMessage.from_user("Can you tell me where Mark lives?")])
print(result["messages"][-1].text)
Agent는 최종 답 외에도 "messages" 키 아래 실행 중 오간 전체 메시지 목록을 반환해요. 아래처럼 내부에서 어떤 도구 호출과 결과가 있었는지 확인할 수 있어요.
for message in result["messages"]:
print(message)
print("-" * 10)
챗 애플리케이션 만들기
Agent가 내부적으로 도구 호출 루프를 처리하므로, 대화를 이어가려면 메시지 히스토리만 관리하면 돼요. 새 사용자 메시지를 추가하고, Agent를 실행하고, 반환된 messages를 다음 턴으로 넘겨서 모델이 맥락을 기억하게 해요. UI는 Gradio의 채팅 인터페이스로 만들 수 있어요.
import gradio as gr
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
agent = Agent(
chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"),
tools=[rag_pipeline_tool, weather_tool],
system_prompt="Use the tools that you're provided with. Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous.",
)
agent.warm_up()
messages = [ChatMessage.from_system(agent.system_prompt)]
def chatbot_with_tc(message, history):
global messages
messages.append(ChatMessage.from_user(message))
# The Agent runs the full tool-calling loop and returns all messages exchanged during the run
result = agent.run(messages=messages)
# Carry the updated conversation history over to the next turn
messages = result["messages"]
return result["messages"][-1].text
demo = gr.ChatInterface(
fn=chatbot_with_tc,
examples=[
"Can you tell me where Giorgio lives?",
"What's the weather like in Madrid?",
"Who lives in London?",
"What's the weather like where Mark lives?",
],
title="Ask me about weather or where people live!",
)
## Uncomment the line below to launch the chat app with UI
# demo.launch()
참고로 OpenAI 모델은 가끔 답이나 도구를 환각(hallucinate)하기도 하니, 기대와 다르게 동작할 수 있다는 점만 염두에 두세요.