chatGPT에 검색엔진을 붙여보자! feat. Naver search api

chatGPT를 이용한 프레임워크중 가장 많이 사용되고 있는 프레임워크가 Langchain 일 것입니다.

 

Langchain에서는 다양한 기능을 미리 만들어서 제공하고 있고,

 

다양한 검색엔진을 연계하는 Agent 또한 준비되어져 있습니다.

 

하지만 업무시스템에 구축되어져있는 상용검색엔진 혹은 업무시스템에 맞게 커스텀되어진 엘라스틱서치를 연계하는

 

맞춤형 Agent는 직접 개발해야 합니다.

 

langchain 공식 문서의 예제 코드인

 

CustomSearchTool 을 활용하여 네이버 API를 호출하는 NaverSearchAPIWrapper 을 만들고 

 

langchain과 연계 할수 있도록 해보겠습니다.

 

NaverSearchAPIWrapper 코드를 각자 사용할 검색엔진을 호출하는 방식으로 커스터마이징 하면

 

손쉽게 langchain을 이용하여 원하는 검색엔진을 붙여 chatGPT 혹은 llm모델을 통해 시멘틱 검색을 구현할 수 있을것입니다.

 

from typing import Any, Dict, List, Optional

import aiohttp
import requests
from pydantic.class_validators import root_validator
from pydantic.main import BaseModel
from typing_extensions import Literal

from langchain.utils import get_from_dict_or_env

from langchain.callbacks.manager import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun
)

class NaverSearchAPIWrapper(BaseModel):
    display: int = 5
    start: int = 1
    sort: str = "date"    
    type: Literal["news","blog","webkr","kin","doc"] = "kin"

    X_Naver_Client_Id: str = "X_Naver_Client_Id"
    X_Naver_Client_Secret: str = "X_Naver_Client_Secret"

    aiosession: Optional[aiohttp.ClientSession] = None

    class Config:
        arbitrary_types_allowed = True
    
    
    
    def results(self, query:str, **kwargs: Any) -> Dict:
        return self._naver_search_api_results(
            search_term = query,
            display= self.display,
            start = self.start,
            sort = self.sort,
            search_type=self.type,
            **kwargs,
        )
    
    def run(self, query: str, **kwargs: Any) -> str:
        results = self._naver_search_api_results(
            search_term = query,
            display= self.display,
            start = self.start,
            sort = self.sort,
            search_type=self.type,
            **kwargs,
        )
        return self._parse_results(results)
    
    async def aresults(self, query: str, **kwargs: Any) -> Dict:
        results = await self._async_naver_search_api_results(
            search_term = query,
            display= self.display,
            start = self.start,
            sort = self.sort,
            search_type=self.type,
            **kwargs,
        )
        return results
    
    async def arun(self, query: str, **kwargs: Any) -> str:
        results = await self._async_naver_search_api_results(
            search_term = query,
            display= self.display,
            start = self.start,
            sort = self.sort,
            search_type=self.type,
            **kwargs,
        )
        return self._parse_results(results)
    
    def _parse_descriptions(self, results: dict) -> List[str] :
        descriptions = []
        for result in results["items"]:
            if "description" in result:
                descriptions.append(result["description"])            

        if len(descriptions) == 0:
            return ["No good Naver Search Result was found"]
        return descriptions

    def _parse_results(self, results: dict) -> str:
        result = " ".join(self._parse_descriptions(results))
        return result    

    def _naver_search_api_results(
            self, search_term: str, search_type: str = "kin", **kwargs: Any
    ) -> dict:
        headers = {
            "X-Naver-Client-Id" : self.X_Naver_Client_Id,
            "X-Naver-Client-Secret" : self.X_Naver_Client_Secret
        }
        params = {
            "query" : search_term,
            **{key: value for key, value in kwargs.items() if value is not None},
        }
        response = requests.get(
            f"https://openapi.naver.com/v1/search/{search_type}.json", headers=headers, params=params
        )
        response.raise_for_status()
        search_results = response.json()
        return search_results
    
    async def _async_naver_search_api_results(
            self, search_term: str, search_type: str = "kin", **kwargs: Any
    ) -> dict:
        headers = {
            "X-Naver-Client-Id" : self.X_Naver_Client_Id,
            "X-Naver-Client-Secret" : self.X_Naver_Client_Secret
        }
        url = f"https://openapi.naver.com/v1/search/{search_type}.json"
        params = {
            "query" : search_term,
            **{key: value for key, value in kwargs.items() if value is not None},
        }

        if not self.aiosession:
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    url, params=params, headers=headers, raise_for_status=False
                ) as response:
                    search_results = await response.json()
        else:
            async with self.aiosession.get(
                url, params=params, headers=headers, raise_for_status=True
            ) as response:
                search_results = await response.json()

        return search_results




search = NaverSearchAPIWrapper()

class CustomSearchTool(BaseTool):
    name = "custom_search"
    description = "useful for when you need to answer questions about current events"

    def _run(
            self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> str:
        result = search.run(query)
        return result
    
    async def _arun(
            self, query: str, run_manager: Optional[AsyncCallbackManagerForToolRun] = None
    ) -> str:
        result = search.arun(query)
        return result




llm = OpenAI(temperature=0, openai_api_key=openAIKey)
tools = [CustomSearchTool()]
agent = initialize_agent(tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True)
agent("여기에 질문할 내용을 입력")

 

이보다 더 간단하게 구현할수 있는 방법이 있다면 댓글로 공유해주세요.

 

감사합니다.