참조가 있는 웹 검색
참조가 있는 웹 검색 (Web Search with References)
Mistral Large 2 모델로 웹 검색을 수행하고, 관련 출처를 응답에 참조(reference)로 포함시키는 방법을 배우는 문서예요. 환각(hallucination)과 잘못된 URL을 방지하는 기능을 다룹니다.
출처: 문서
본문
이 쿡북의 주된 목표는 최신 [Mistral Large 2] 모델을 효과적으로 사용해 웹 검색을 수행하고 관련 소스를 응답에 통합하는 방법을 보여주는 거예요. 챗봇과 RAG(검색 증강 생성) 시스템의 흔한 문제 중 하나는 소스를 환각하거나 URL을 잘못 포맷하는 경향이에요. Mistral의 고급 기능은 이러한 문제를 해결해 정확하고 신뢰할 수 있는 정보 검색을 보장합니다.
Mistral의 웹 검색 능력
새 Mistral 모델 mistral-large-latest는 웹 검색 기능을 통합해 응답에서 출처를 정확히 참조할 수 있게 해요. 이 기능을 사용하면 소스 콘텐츠를 검색해 응답에 올바르게 표시할 수 있어, 제공되는 정보의 신뢰성과 신용도를 높여요. Mistral의 고급 자연어 처리와 웹 검색 통합을 활용하면 더 견고하고 신뢰할 수 있는 애플리케이션을 만들 수 있어요.
프로세스의 단계별 설명:
- Query Initiation: 사용자 쿼리로 시작해요.
- Function Calling with Mistral Large: 쿼리가 Mistral Large 모델에서 처리되고, 더 많은 정보를 얻기 위해 함수 호출을 수행해야 한다고 식별해요. 이 단계는 쿼리에 적합한 도구를 결정합니다.
- Tool Identification: Mistral 모델이 쿼리에 관련된 도구를 식별해요. 이 경우
web_search_wikipedia이에요. 도구는 사용자 쿼리를 인자로 가져요. - Wikipedia Search: 도구가 호출되어 쿼리로 Wikipedia에서 검색해요.
- Extract Relevant Chunks: Wikipedia 검색 결과를 처리해 관련 정보 청크를 추출해요. 이 청크들은 최종 답변의 참조로 사용되도록 준비됩니다.
- Final Answer with References: 채팅 기록이 Mistral Large 모델로 보내지고, 추출된 청크를 사용해 최종 답변을 생성해요. 답변에는 Wikipedia 문서에 대한 참조가 포함되어 정보가 정확하고 잘 출처가 명시되도록 합니다.
!pip install mistralai==1.2.3 wikipedia==1.4.0
Step 1: Mistral 클라이언트 초기화
API 키로 Mistral 클라이언트를 초기화해요. [Mistral API 대시보드]에서 API 키를 얻거나 만들 수 있어요. (경고: API 키 활성화에 최대 1분이 걸릴 수 있어요.)
from mistralai.client import Mistral
from mistralai.client.models import UserMessage, SystemMessage
import os
client = Mistral(
api_key=os.environ["MISTRAL_API_KEY"],
)
query = "Who won the Nobel Peace Prize in 2024?"
#Add the user message to the chat_history
chat_history = [
SystemMessage(content="You are a helpful assistant that can search the web for information. Use context to answer the question."),
UserMessage(content=query),
]
print(chat_history)
Step 2: Wikipedia를 검색하는 함수 호출 도구 정의
[함수 호출(Function calling)]을 사용하면 Mistral 모델이 외부 도구에 연결할 수 있어요. Mistral 모델을 사용자 정의 함수나 API 같은 외부 도구와 통합하면 특정 사용 사례와 실용적 문제를 해결하는 애플리케이션을 쉽게 만들 수 있어요.
먼저 Wikipedia API를 검색하고 결과를 특정 형식으로 반환하는 도구를 만들어요. 도구가 준비되면 Mistral에 대한 채팅 완성 요청에서 사용할 수 있어요. 결과에는 다음이 포함되어야 해요:
- 도구의 이름
- 도구 호출 ID(Tool call ID)
- 사용자 쿼리를 포함하는 인자(Arguments)
web_search_tool = {
"type": "function",
"function": {
"name": "web_search",
"description": "Search the web for a query for which you do not know the answer",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Query to search the web in keyword form.",
}
},
"required": ["query"],
},
},
}
chat_response = client.chat.complete(
model="mistral-large-latest",
messages=chat_history,
tools=[web_search_tool],
)
if hasattr(chat_response.choices[0].message, 'tool_calls'):
tool_call = chat_response.choices[0].message.tool_calls[0]
chat_history.append(chat_response.choices[0].message)
print(tool_call)
else:
print("No tool call found in the response")
Step 3: 도구와 연결된 Wikipedia 검색 메서드 정의
이전 단계에서 web_search_wikipedia라는 도구를 만들었어요. 도구 호출 ID와 인자를 받아 특정 형식으로 결과를 반환하는 함수를 만들어야 해요.
결과 형식은 다음과 같아야 해요:
{
"url": str | None, # Page URL
"title": str | None, # Page title
"description": str | None, # Page description
"snippets": List[str], # Relevant text snippets in a list
"date": str | None, # date
"source": str | None, # Source/reference
"metadata": Dict[str, Any] # Metadata
}
import wikipedia
import json
from datetime import datetime
def get_wikipedia_search(query: str) -> str:
"""
Search Wikipedia for a query and return the results in a specific format.
"""
result = wikipedia.search(query, results = 5)
data={}
for i, res in enumerate(result):
pg= wikipedia.page(res, auto_suggest=False)
data[i]={
"url": pg.url,
"title": pg.title,
"snippets": [pg.summary.split('.')],
"description": None,
"date": datetime.now().isoformat(),
"source": "wikipedia"
}
return json.dumps(data, indent=2)
Step 4: 도구 호출 수행 및 Wikipedia 검색
이제 도구 호출 ID와 인자가 있으니, 도구 호출을 수행하고 Wikipedia를 검색할 수 있어요.
import json
from mistralai import ToolMessage
query = json.loads(tool_call.function.arguments)["query"]
wb_result = get_wikipedia_search(query)
tool_call_result = ToolMessage(
content=wb_result,
tool_call_id=tool_call.id,
name=tool_call.function.name,
)
# Append the tool call message to the chat_history
chat_history.append(tool_call_result)
#See chunks in the response
print(json.dumps(json.loads(wb_result), indent=2))
Step 5: 도구 호출 결과로 Mistral 호출
이제 채팅 기록에는 다음이 포함돼요:
- 어시스턴트에 대한 지침을 포함한
System메시지 - 원래 질문을 포함한
User메시지 - Wikipedia 검색을 위한 도구 호출을 포함한
Assistant메시지 - Wikipedia 검색 결과를 포함한
Tool call결과
for msg in chat_history:
print(msg,end='\n')
출력을 포맷해 답변과 참조를 함께 보여주는 format_response 함수를 만들어요. 응답 콘텐츠가 TextChunk면 텍스트를 출력하고, ReferenceChunk면 참조 ID를 수집해 출처 목록으로 표시합니다.
from mistralai.client.models import TextChunk, ReferenceChunk
def format_response(chat_response: list, wb_result:dict):
print("\n🤖 Answer:\n")
refs_used = []
# Print the main response
for chunk in chat_response.choices[0].message.content:
if isinstance(chunk, TextChunk):
print(chunk.text, end="")
elif isinstance(chunk, ReferenceChunk):
refs_used += chunk.reference_ids
# Print references
if refs_used:
print("\n\n📚 Sources:")
for i, ref in enumerate(set(refs_used), 1):
reference = json.loads(wb_result)[str(ref)]
print(f"\n{i}. {reference['title']}: {reference['url']}")
# Use the formatter
chat_response = client.chat.complete(
model="mistral-large-latest",
messages=chat_history,
tools=[web_search_tool],
)
format_response(chat_response, wb_result)
Step 6: 참조가 있는 스트리밍 완성 (Streaming completion with references)
스트리밍 중에도 참조를 처리해 실시간으로 URL을 표시할 수 있어요.
stream_response = client.chat.stream(
model="mistral-large-2411",
messages=chat_history,
tools=[web_search_tool],
)
last_reference_index = 0
if stream_response is not None:
for event in stream_response:
chunk = event.data.choices[0]
if chunk.delta.content:
if isinstance(chunk.delta.content, list):
# Check if TYPE of chunk is a reference
references_ids = [
ref_id
for chunk_elem in chunk.delta.content
if chunk_elem.TYPE == "reference"
for ref_id in chunk_elem.reference_ids
]
last_reference_index += len(references_ids)
# Map the references ids to the references data stored in the chat history
references_data = [json.loads(wb_result)[str(ref_id)] for ref_id in references_ids]
urls = " " + ", ".join(
[
f"[{i}]({reference['url']})"
for i, reference in enumerate(
references_data,
start=last_reference_index - len(references_ids) + 1,
)
]
)
print(urls, end="")
else:
print(chunk.delta.content, end="")
스트리밍 결과로 "The 2024 Nobel Peace Prize was awarded to Nihon Hidankyo ... in 1945 [1], [2]."처럼 번호 매겨진 참조 [1], [2]가 URL로 표시되는 걸 볼 수 있어요.