LlamaIndex를 활용한 Chatbot 개발

Seokchan Yoon·2024년 9월 24일

전체 흐름 정리

사용할 LLM api를 통해서 Custom Model 클래스를 생성한다.
사용할 embedding api를 통해서 Custom Embed 클래스를 생성한다.
임베딩 값을 저장할 vector db를 세팅한다.
임베딩할 문서가 임베딩 가능한 사이즈보다 크다면 parser를 사용하여 나눠준다.
알맞은 사이즈로 분할된 문서를 임베딩한다.
필요하다면 노드에 metadata를 추가한다. 표 데이터의 경우는 각 column이 어떤 의미를 갖는지를 metadata에 추가할 수 있다. 이 table_summary라는 metadata는 임베딩은 되지 않지만 LLM에게만 전달되도록 설정할 수 있다.
사용자 쿼리가 들어오면 임베딩을 한다.
Vector DB에서 similarity score가 높은 노드들을 뽑는다.
필요에 따라서 post processor를 통해서 retrieved 노드들을 가공한다.
Prompt template을 통해서 LLM에게 요청을 보낸다.
Retrieved nodes 수가 많아서 LLM이 한 번에 처리할 수 있는 데이터의 크기보다 크다면 refine 과정을 통해서 답을 만들어간다.

LlamaIndex 설치

python 3.10 이상 설치해야하는 것 같다. python3.10 부터는 openssl 1.1.1 이상이 필요하다.

yum install epel-release -y
yum install openssl11 openssl11-devel -y
yum -y install yum-utils
yum -y install mariadb-devel
yum -y install zlib zlib-devel libffi-devel bzip2-devel
yum -y install gcc gcc-c++ openssl openssl-devel
yum -y install zip unzip wget mc git net-tools

cd /opt
wget <https://www.python.org/ftp/python/3.11.3/Python-3.11.3.tgz>  --no-check-certificate
tar -zxvf Python-3.11.3.tgz
cd Python-3.11.3
./configure
make
make install

pip install llama-index

LlamaIndex High Level Concepts

Loading

텍스트 파일, PDF 파일, 웹 사이트, 데이터베이스, API 등의 데이터를 파이프라인에 넣는 것이다. LlamaHub에 다양한 connector가 있다.

Nodes and Documents
Document는 데이터 소스에 대한 컨테이너이다.
Node는 라마인덱스 데이터의 한 단위이고 Document의 chunk이다. Node에는 메타데이터가 포함된다.
Connectors
Reader라고도 불린다. 데이터를 여러 소스로부터 받아서 Document와 Node를 만든다.

Indexing

데이터를 쿼리할 수 있는 자료 구조를 생성하는 것이다. LLM에서는 주로 벡터 임베딩을 의미한다.

Indexes
데이터를 ingest하고 나면 retrieve 하기 쉬운 구조로 만들어야한다. 주로 벡터 임베딩을 생성한다. Indexes는 데이터에 대한 메타데이터도 포함할 수 있다.
Embeddings
LLM은 숫자로 표현된 데이터인 임베딩을 만들어낸다. 연관도에 따라 데이터를 골라낼 때 라마인덱스는 쿼리를 임베딩으로 바꾸게 되고, 벡터 스토어는 쿼리 임베딩과 비슷한 데이터를 찾는다.

Storing

데이터가 인덱싱되면 다시 인덱싱하지 않아도 되도록 그 인덱스를 저장한다.

Querying

LLM과 라마인덱스를 사용해서 쿼리할 수 있는 방법은 다양하다.

Retrievers
인덱스에서 얼마나 효율적으로 관련된 데이터를 뽑는지를 결정한다. retrieval strategy는 연관도와 효율성을 결정한다.
Routers
knowledge base에서 관련된 내용을 뽑는 데 어떤 retriever가 사용될지 정한다. RouterRetriever 클래스는 쿼리를 실행할 하나 혹은 그 이상의 후보 retriever를 고른다.
Node Postprocessors
retrieved node를 받아서 transformation, filtering, re-ranking과 같은 작업을 적용한다.
Response Synthesizers
사용자 쿼리와 retrieved text chunk를 통해서 LLM으로부터 response를 생성한다.

Evaluation

다른 방법에 비해 얼마나 효과적인지 확인하는 것도 중요하다. 쿼리에 대한 응답이 얼마나 믿을만하고 빠른지를 확인해준다.

문서 분할하기

문서가 한 번에 embedding 할 수 있는 크기보다 크다면 나눠서 embedding을 해야한다.

MarkdownNodeParser
- header와 code block 기준으로 문서를 노드로 분할
- 노드의 metadata에 어떤 파일에 대한 정보인지 기록됨
MarkdownElementNodeParser
- 더 세부적인 markdown 문법으로 노드를 분할 (e.g. text, table, image)
노드 parsing 시 테이블 데이터면 LLM을 통해 table summary 및 column 정보 생성

CustomLLM 구현하기

class ApiTool:
    _api_cnt = 0
    _segmentation_api_cnt = 0
    _completion_api_cnt = 0
    _embedding_api_cnt = 0

    def __init__(self, access_token=None):
        self._base_url = conf.llm_base_url
        self._client_id = os.getenv("CLIENT_ID")
        self._client_secret = os.getenv("CLIENT_SECRET")
        if not self._client_id or not self._client_secret:
            logger.error(f"Credential is required.")
            sys.exit(1)

        self._chat_model_name = conf.llm_model
        self._embed_model_name = conf.embedding_model

        if access_token is None:
            self._refresh_access_token()
        else:
            self._access_token = access_token
        
        self.docstore = SimpleDocumentStore()

    def _refresh_access_token(self):
        # If existingToken is true, it returns a token that has the longest expiry time among existing tokens.
        u = f'{self._base_url}/v1/auth/token?existingToken=true'
        ...
        r = requests.get(url=u, headers=headers)
        self._api_cnt += 1
        logger.info(f"Access Token refreshed: {self._access_token}")

    def _request_post(self, url, json_body, headers, is_retry=False):
        r = requests.post(url=url, json=json_body, headers=headers)
        self._api_cnt += 1
        try:
            if r.statuc_code == 401:
                # refresh token and retry
            if r.status_code != 200:
                self._refresh_access_token()
                headers['Authorization'] = f'Bearer {self._access_token}'
                raise Exception(f"{r.status_code} {r.text}")
        except Exception as e:
            logger.error(f"API  failed.")
        return r

    def get_completion(self, query):
        u = f'{self._base_url}/v1/completion/{self._chat_model_name}'
        headers = {'Authorization': f'Bearer {self._access_token}'}
        d = {...} # input query and parameters

        r = self._request_post(url=u, json_body=d, headers=headers)
        return r.json()['result']['message']['content']

class MyLLM(CustomLLM):
    context_window: int = 2048
    num_output: int = 256
    model_name: str = "my-model"
    chunk_size: int = 1024
    dummy_response: str = "My Dummy Response"
    error_response: str = "Something went wrong! Please contact administrator."
    api_tool: ApiTool = None

    def __init__(self, input_system_prompt: str = '', api_tool: ApiTool = None, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        if input_system_prompt:
            self.system_prompt = input_system_prompt
        else:
            self.system_prompt = system_prompt  # llama_index/core/llms/llm 에서 _extend_prompt 해준다.
        if api_tool:
            self.api_tool = api_tool
        else:
            self.api_tool = ApiTool()

    @property
    def metadata(self) -> LLMMetadata:
        return LLMMetadata(
            context_window=self.context_window,
            num_output=self.num_output,
            model_name=self.model_name
        )

    @llm_completion_callback()
    def complete(self, prompt: str, **kwargs: Any) -> CompletionResponse:
        logger.debug(f"{prompt=}\n\n")
        try:
            completion = self.api_tool.get_completion(prompt)
        except Exception as e:
            logger.error(e)
            return CompletionResponse(text=self.error_response)
        return CompletionResponse(text=completion)

    @llm_completion_callback()
    def stream_complete(self, prompt: str, **kwargs: Any) -> CompletionResponseGen:
        # todo: Not implemented yet
        response = ""
        for token in self.dummy_response:
            response += token
            yield CompletionResponse(text=response, delta=token)
        # raise NotImplementedError()

응답 생성하기

response_mode: refine
노드 리스트가 주어졌을 때, 노드를 여러 개 묶어서 prompt를 생성하고 LLM에 요청
LLM이 반환한 응답과 남은 노드 중 일부를 묶어서 다시 LLM에 요청
모든 노드를 사용할 때까지 이 과정을 반복
예시
- 쿼리: q, 노드 리스트: n1, n2, n3, n4
- 1st iteration: (q, n1, n2) + LLM => r1
- 2nd iteration: (q, r1, n3, n4) + LLM => r2 => 최종 response

References

https://docs.llamaindex.ai/en/stable