나만의 무엇이든 만드는 방법GPT — 원하는 방식으로 응답하는 봇

여러분, 안녕하세요! 최근에는 연습 중에 오랫동안 시도해 보고 싶었던 흥미로운 솔루션을 적용했으며, 이제 다른 작업에 대해 비슷한 것을 만들 수 있는 방법을 설명할 준비가 되었습니다. 프롬프트의 크기에 따라 길이가 제한되지 않는 대규모 지식 기반을 고려하여 질문에 답하는 맞춤형 버전의 ChatGPT를 만드는 방법에 대해 이야기하겠습니다. ChatGPT에 질문).

이를 달성하기 위해 우리는 OpenAI(지식 기반에서 관련 질문을 고품질로 검색하기 위해)와 ChatGPT API 자체(자연스러운 인간 언어로 답변 형식을 지정하기 위해)의 상황별 임베딩을 사용할 것입니다.

또한, 명시적으로 언급된 Q&A 질문뿐만 아니라 해당 Q&A를 잘 아는 사람이 답변할 수 있는 질문에도 보조자가 답변할 수 있다고 가정한다. 대규모 지식 기반을 사용하여 응답하는 간단한 봇을 만드는 방법을 배우고 싶다면 자세한 내용을 확인하세요.

이 문제를 프레임워크 형태로 해결하려는 일부 라이브러리 프로젝트(예: LangChain ) 가 있다는 점을 지적하고 싶고, 저도 이를 사용해 보았습니다. 그러나 개발 초기 단계에 있는 다른 프레임워크와 마찬가지로 경우에 따라 상황을 단순화하기보다는 제한하는 경향이 있습니다. 특히, 이 작업을 해결하는 초기부터 데이터로 무엇을 하고 싶은지 이해하고 이를 스스로 수행하는 방법(컨텍스트 기반 검색, 프롬프트에 올바른 컨텍스트 설정, 정보 소스 결합 포함)을 알고 있었습니다.

그러나 허용 가능한 수준의 품질로 정확하게 이를 수행하도록 프레임워크를 구성할 수 없었고 프레임워크를 디버깅하는 것은 이 작업에 과도한 것처럼 보였습니다. 결국 나는 나만의 상용구 코드를 만들었고 이 접근 방식에 만족했습니다.

일

제가 진행하고 있던 작업에 대해 간략히 설명드리자면, 동일한 코드를 자신의 작업에 사용하여 데이터 소스와 프롬프트를 자신에게 맞는 것으로 바꾸시면 됩니다 . 귀하는 여전히 봇의 논리를 완전히 제어할 수 있습니다.

코드를 작성할 때 ChatGPT를 자주 사용합니다(부끄럽지 않습니다🙂). 그러나 2022년 이상 데이터가 부족하여 상대적으로 새로운 기술에 문제가 있는 경우가 있습니다.

특히, The Graph 프로토콜(EVM 호환 블록체인에서 색인화된 데이터를 검색하기 위해 ETL을 구축하는 가장 인기 있는 방법)에 대한 하위 그래프를 개발할 때 이에 대한 자세한 내용은 이전 기사 [ 1 ] 및 [ 2 ]에서 읽을 수 있습니다. 몇 가지 호환성이 크게 변경되었습니다. ChatGPT의 "오래된" 답변은 더 이상 도움이 되지 않으며, 부족한 문서나 최악의 경우 개발자의 Discord에서 정답을 검색해야 하는데 이는 그다지 편리하지 않습니다(StackOverflow와는 다릅니다).

문제의 두 번째 부분은 ChatGPT가 종종 하위 그래프 주제에서 벗어나 GraphQL, SQL 또는 더 높은 수학("그래프", "하위 그래프" 등)으로 이동하기 때문에 대화 컨텍스트를 올바르게 제공해야 할 때마다 있다는 것입니다. 고유한 용어가 아니며 다양한 해석과 주제가 있습니다).

따라서 하위 그래프 코드의 오류를 수정하기 위해 ChatGPT로 짧은 시간 동안 고군분투한 후, 지식 기반과 개발자 불일치의 메시지를 고려하여 항상 올바른 컨텍스트에 있고 답변을 시도하는 나만의 SubgraphGPT 봇을 만들기로 결정했습니다. .

추신. 저는 Web3 인프라 제공업체인 chainstack.com 에서 수석 제품 관리자로 일하고 있으며 하위 그래프 호스팅 서비스 개발을 담당하고 있습니다. 그래서 저는 사용자가 상대적으로 새로운 기술을 이해하도록 돕기 위해 하위 그래프를 많이 다루어야 합니다.

최상위 솔루션

결국 이 문제를 해결하기 위해 두 가지 소스를 사용하기로 결정했습니다.

1. 세미 블라인드 모드에서 선택된 질문과 답변에 대해 수동으로 편집된 지식 기반입니다(종종 문서의 주제 제목을 질문으로 사용하고 전체 정보 단락을 답변으로 사용했습니다).

2. 프로토콜 개발자 Discord에서 지난 2년간의 메시지를 내보냈습니다(2021년 말부터 누락된 기간을 처리하기 위해).

   
   

다음으로 각 소스에 대해 ChatGPT API에 대한 요청을 작성하기 위해 다양한 접근 방식이 사용되었습니다. 특히 다음과 같습니다.

직접 작성한 Q&A의 경우,

1. 각 질문에 대해 text-embedding-ada-002 모델을 통해 얻은 상황별 임베딩(다차원 상태에서 이 질문을 설명하는 벡터)이 생성됩니다.

2. 그런 다음 코사인 거리 검색 기능을 사용하여 지식 베이스에서 가장 유사한 상위 3개의 질문을 찾습니다(3 대신 데이터세트에 가장 적합한 숫자를 사용할 수 있음).

3. 이 3가지 질문에 대한 답변은 " 주어진 질문과 관련된 경우에만 이 Q&A 스니펫을 사용하세요 ."라는 대략적인 설명과 함께 최종 프롬프트에 추가됩니다.

   
   

   Discord에서 내보낸 메시지에는 다음 알고리즘이 사용되었습니다.

4. 물음표가 포함된 각 메시지에 대해 상황별 임베딩도 생성됩니다(동일한 모델 사용).

5. 그런 다음 비슷한 방식으로 가장 유사한 질문 상위 5개를 선택하고,

6. 그리고 답변의 맥락으로 해당 질문 다음에 나오는 20개의 메시지가 추가되며, 이는 질문에 대한 답변을 포함할 특정 확률을 가지고 있다고 가정됩니다.

7. 이 정보는 대략 다음과 같이 최종 프롬프트에 추가되었습니다. " 첨부된 Q&A 스니펫에서 질문에 대한 명확한 답변을 찾지 못한 경우 개발자가 제공하는 다음 채팅 조각이 원래 질문에 답변하는 데 유용할 수 있습니다. . "

또한 주제가 명시적으로 제공되지 않은 경우 Q&A 스니펫 및 채팅이 있으면 답변이 모호해질 수 있습니다. 예를 들어 다음과 같습니다.

따라서 질문이 맥락에서 분리되었고 답변도 맥락에서 분리되어 받아들여졌다고 이해합니다. 그런 다음 그러한 데이터를 사용할 수 있다고 말했으며 이를 요약하면 다음과 같습니다.

1. 실제로 대답은 다음과 같을 수 있습니다 ...
2. 그리고 맥락을 고려해보면 이렇습니다...

이를 방지하기 위해 다음과 같이 프롬프트 시작 부분에 명시적으로 정의되고 삽입되는 주제 개념을 도입합니다.

"'그래프 하위 그래프 개발' 주제와 관련된 질문에 대한 답변을 얻고 싶습니다. {{{하위 그래프는 무엇입니까?}}}"

게다가 마지막 문장에 이렇게 덧붙입니다.

마지막으로 위의 정보가 충분하지 않은 경우에만 '그래프 하위 그래프 개발' 주제에 대한 지식을 활용하여 질문에 답할 수 있습니다.

결국 전체 프롬프트(채팅에서 얻은 부분 제외)는 다음과 같습니다.

 ==I need to get an answer to the question related to the topic of "The Graph subgraph development": {{{what is a subgraph?}}}.== ==Possibly, you might find an answer in these Q&As \[use the information only if it is actually relevant and useful for the question answering\]:== ==Q: <What is a subgraph?>== ==A: <A subgraph is a custom API built on blockchain data. Subgraphs are queried using the GraphQL query language and are deployed to a Graph Node using the Graph CLI. Once deployed and published to The Graph's decentralized network, Indexers process subgraphs and make them available to be queried by subgraph consumers.>== ==Q: <Am I still able to create a subgraph if my smart contracts don't have events?>== ==A: <It is highly recommended that you structure your smart contracts to have events associated with data you are interested in querying. Event handlers in the subgraph are triggered by contract events and are by far the fastest way to retrieve useful data. If the contracts you are working with do not contain events, your subgraph can use call and block handlers to trigger indexing. Although this is not recommended, as performance will be significantly slower.>== ==Q: <How do I call a contract function or access a public state variable from my subgraph mappings?>== ==A: <Take a look at Access to smart contract state inside the section AssemblyScript API. https://thegraph.com/docs/en/developing/assemblyscript-api/>== ==Finally, only if the information above was not enough you can use your knowledge in the topic of "The Graph subgraph development" to answer the question.==

입력 시 반자동 생성된 프롬프트가 포함된 위 요청에 대한 응답은 처음부터 올바르게 보입니다.

이 경우 봇은 즉시 올바른 키로 응답하고 관련 정보를 더 추가하므로 Q&A처럼 답변이 간단해 보이지는 않습니다(이 질문은 정확히 질문과 답변 목록에 있음을 상기시켜 드립니다). 다음 질문을 부분적으로 다루는 합리적인 설명.

소스 코드

마지막에 저장소에 대한 링크가 있다는 점을 바로 알아두어야 합니다. 따라서 봇을 있는 그대로 실행할 수 있으며, "주제"를 자신의 것으로 바꾸고, Q&A 지식 기반 파일을 자신의 것으로 바꾸고, 자신만의 API 키를 제공할 수 있습니다. OpenAI 및 Telegram 봇용. 따라서 여기 설명은 GitHub의 소스 코드와 완전히 일치하기 위한 것이 아니라 코드의 주요 측면을 강조하기 위한 것입니다.

1 - 가상 환경 준비

새로운 가상 환경을 생성하고 요구사항.txt에서 종속성을 설치해 보겠습니다.

 virtualenv -p python3.8 .venv source .venv/bin/activate pip install -r requirements.txt

2 - 수동으로 수집된 기술 자료

위에서 언급한 대로 질문과 답변 목록이 있다고 가정합니다. 이 경우에는 다음 유형의 Excel 파일 형식입니다.

주어진 질문과 가장 유사한 질문을 찾으려면 이 파일의 각 줄에 질문 임베딩(상태 공간의 다차원 벡터)을 추가해야 합니다. 이를 위해 add_embeddings.py 파일을 사용하겠습니다. 스크립트는 몇 가지 간단한 부분으로 구성됩니다.

라이브러리 가져오기 및 명령줄 인수 읽기:

 import pandas as pd import openai import argparse # Create an Argument Parser object parser = argparse.ArgumentParser(description='Adding embeddings for each line of csv file') # Add the arguments parser.add_argument('--openai_api_key', type=str, help='API KEY of OpenAI API to create contextual embeddings for each line') parser.add_argument('--file', type=str, help='A source CSV file with the text data') parser.add_argument('--colname', type=str, help='Column name with the texts') # Parse the command-line arguments args = parser.parse_args() # Access the argument values openai.api_key = args.openai_api_key file = args.file colname = args.colname

다음으로 파일을 pandas 데이터 프레임으로 읽고 물음표가 있는지에 따라 질문을 필터링합니다. 이 코드 조각은 지식 기반과 Discord의 원시 메시지 스트림을 처리하는 데 일반적이므로 질문이 자주 중복된다는 가정 하에 대략적인 질문이 아닌 필터링의 간단한 방법을 유지하기로 결정했습니다.

 if file[-4:] == '.csv': df = pd.read_csv(file) else: df = pd.read_excel(file) # filter NAs df = df[~df[colname].isna()] # Keep only questions df = df[df[colname].str.contains('\?')]

마지막으로, 모델 text-embedding-ada-002 의 API를 호출하여 임베딩을 생성하는 함수입니다. API가 때때로 오버로드되어 오류로 응답할 수 있으므로 몇 번의 반복 요청이 있으며 이 함수를 각 행에 적용합니다. 데이터프레임의.

 def get_embedding(text, model="text-embedding-ada-002"): i = 0 max_try = 3 # to avoid random OpenAI API fails: while i < max_try: try: text = text.replace("\n", " ") result = openai.Embedding.create(input=[text], model=model)['data'][0]['embedding'] return result except: i += 1 def process_row(x): return get_embedding(x, model='text-embedding-ada-002') df['ada_embedding'] = df[colname].apply(process_row) df.to_csv(file[:-4]+'_question_embed.csv', index=False)

결국 이 스크립트는 다음 명령을 사용하여 호출할 수 있습니다.

 python add_embeddings.py \ --openai_api_key="xxx" \ --file="./subgraphs_faq.xlsx" \ --colname="Question"

OpenAI API 키, 지식베이스가 포함된 파일, 질문 텍스트가 있는 열 이름을 설정합니다. 최종 생성된 파일인 subgraphs_faq._question_embed.csv에는 "Question", "Answer" 및 "ada_embedding " 열이 포함되어 있습니다.

3 - Discord에서 데이터 수집(선택 사항)

수동으로 수집한 지식 기반만을 기반으로 응답하는 간단한 봇에 관심이 있는 경우 이 섹션과 다음 섹션을 건너뛸 수 있습니다. 하지만 여기에서는 Discord 채널과 Telegram 그룹 모두에서 데이터를 수집하기 위한 코드 예제를 간략하게 제공하겠습니다. discord-channel-data-collection.py 파일은 두 부분으로 구성됩니다. 첫 번째 부분에는 라이브러리 가져오기 및 명령줄 인수 초기화가 포함됩니다.

 import requests import json import pandas as pd import argparse # Create an Argument Parser object parser = argparse.ArgumentParser(description='Discord Channel Data Collection Script') # Add the arguments parser.add_argument('--channel_id', type=str, help='Channel ID from the URL of a channel in browser https://discord.com/channels/xxx/{CHANNEL_ID}') parser.add_argument('--authorization_key', type=str, help='Authorization Key. Being on the discord channel page, start typing anything, then open developer tools -> Network -> Find "typing" -> Headers -> Authorization.') # Parse the command-line arguments args = parser.parse_args() # Access the argument values channel_id = args.channel_id authorization_key = args.authorization_key

두 번째는 채널에서 데이터를 검색하고 이를 Pandas 데이터 프레임에 저장하는 기능과 지정된 매개변수를 사용한 호출입니다.

 def retrieve_messages(channel_id, authorization_key): num = 0 limit = 100 headers = { 'authorization': authorization_key } last_message_id = None # Create a pandas DataFrame df = pd.DataFrame(columns=['id', 'dt', 'text', 'author_id', 'author_username', 'is_bot', 'is_reply', 'id_reply']) while True: query_parameters = f'limit={limit}' if last_message_id is not None: query_parameters += f'&before={last_message_id}' r = requests.get( f'https://discord.com/api/v9/channels/{channel_id}/messages?{query_parameters}', headers=headers ) jsonn = json.loads(r.text) if len(jsonn) == 0: break for value in jsonn: is_reply = False id_reply = '0' if 'message_reference' in value and value['message_reference'] is not None: if 'message_id' in value['message_reference'].keys(): is_reply = True id_reply = value['message_reference']['message_id'] text = value['content'] if 'embeds' in value.keys(): if len(value['embeds'])>0: for x in value['embeds']: if 'description' in x.keys(): if text != '': text += ' ' + x['description'] else: text = x['description'] df_t = pd.DataFrame({ 'id': value['id'], 'dt': value['timestamp'], 'text': text, 'author_id': value['author']['id'], 'author_username': value['author']['username'], 'is_bot': value['author']['bot'] if 'bot' in value['author'].keys() else False, 'is_reply': is_reply, 'id_reply': id_reply, }, index=[0]) if len(df) == 0: df = df_t.copy() else: df = pd.concat([df, df_t], ignore_index=True) last_message_id = value['id'] num = num + 1 print('number of messages we collected is', num) # Save DataFrame to a CSV file df.to_csv(f'../discord_messages_{channel_id}.csv', index=False) if __name__ == '__main__': retrieve_messages(channel_id, authorization_key)

여기에 있는 유용한 정보에는 필요할 때마다 찾을 수 없는 세부 정보인 인증 키 획득이 있습니다. Channel_id는 브라우저에 열린 Discord 채널의 URL(링크의 마지막 긴 숫자)에서 얻을 수 있다는 점을 고려하면, Authorization_key는 채널에 메시지를 입력하기 시작한 다음 개발자 도구를 사용하여 찾아야만 찾을 수 있습니다. 네트워크 섹션에 " typing "이라는 이벤트를 입력하고 헤더에서 매개변수를 추출합니다.

이러한 매개변수를 받은 후 다음 명령을 실행하여 채널에서 모든 메시지를 수집할 수 있습니다(사용자 고유의 값으로 대체).

 python discord-channel-data-collection.py \ --channel_id=123456 \ --authorization_key="123456qwerty"

4 - 텔레그램에서 데이터 수집

텔레그램의 채팅/채널에서 다양한 데이터를 자주 다운로드하기 때문에 비슷한 형식( add_embeddings.py 스크립트 측면에서 호환 가능) CSV 파일을 생성하는 코드도 제공하기로 결정했습니다. 따라서 telegram-group-data-collection.py 스크립트는 다음과 같습니다. 명령줄에서 라이브러리 가져오기 및 인수 초기화:

 import pandas as pd import argparse from telethon import TelegramClient # Create an Argument Parser object parser = argparse.ArgumentParser(description='Telegram Group Data Collection Script') # Add the arguments parser.add_argument('--app_id', type=int, help='Telegram APP id from https://my.telegram.org/apps') parser.add_argument('--app_hash', type=str, help='Telegram APP hash from https://my.telegram.org/apps') parser.add_argument('--phone_number', type=str, help='Telegram user phone number with the leading "+"') parser.add_argument('--password', type=str, help='Telegram user password') parser.add_argument('--group_name', type=str, help='Telegram group public name without "@"') parser.add_argument('--limit_messages', type=int, help='Number of last messages to download') # Parse the command-line arguments args = parser.parse_args() # Access the argument values app_id = args.app_id app_hash = args.app_hash phone_number = args.phone_number password = args.password group_name = args.group_name limit_messages = args.limit_messages

보시다시피, 자신을 첫 번째 사람으로 인증하지 않으면 채팅에서 모든 메시지를 다운로드할 수 없습니다. 즉, https://my.telegram.org/apps를 통해 앱을 생성하는 것(APP_ID 및 APP_HASH 획득) 외에도 Telethon 라이브러리에서 TelegramClient 클래스의 인스턴스를 생성하려면 전화번호와 비밀번호를 사용해야 합니다. .

또한 텔레그램 채팅의 공개 group_name이 필요하고 검색할 최신 메시지 수를 명시적으로 지정해야 합니다. 전반적으로 한 계정에서 메시지를 너무 자주 보내는 경우와는 달리 Telegram API에서 임시 또는 영구 금지를 받지 않고 내보낸 메시지 수에 관계없이 이 절차를 여러 번 수행했습니다.

스크립트의 두 번째 부분에는 메시지 내보내기 및 실행을 위한 실제 기능이 포함되어 있습니다(수집을 중간에 중지하는 심각한 오류를 방지하기 위해 필요한 필터링 포함).

 async def main(): messages = await client.get_messages(group_name, limit=limit_messages) df = pd.DataFrame(columns=['date', 'user_id', 'raw_text', 'views', 'forwards', 'text', 'chan', 'id']) for m in messages: if m is not None: if 'from_id' in m.__dict__.keys(): if m.from_id is not None: if 'user_id' in m.from_id.__dict__.keys(): df = pd.concat([df, pd.DataFrame([{'date': m.date, 'user_id': m.from_id.user_id, 'raw_text': m.raw_text, 'views': m.views, 'forwards': m.forwards, 'text': m.text, 'chan': group_name, 'id': m.id}])], ignore_index=True) df = df[~df['user_id'].isna()] df = df[~df['text'].isna()] df['date'] = pd.to_datetime(df['date']) df = df.sort_values('date').reset_index(drop=True) df.to_csv(f'../telegram_messages_{group_name}.csv', index=False) client = TelegramClient('session', app_id, app_hash) client.start(phone=phone_number, password=password) with client: client.loop.run_until_complete(main())

결국 이 스크립트는 다음 명령으로 실행될 수 있습니다(값을 원하는 값으로 바꾸십시오).

 python telegram-group-data-collection.py \ --app_id=123456 --app_hash="123456qwerty" \ --phone_number="+xxxxxx" --password="qwerty123" \ --group_name="xxx" --limit_messages=10000

5 - 실제로 질문에 답하는 텔레그램 봇 스크립트

대부분의 경우 애완동물 프로젝트를 Telegram 봇으로 래핑합니다. 실행하는 데 최소한의 노력이 필요하고 즉시 잠재력을 보여주기 때문입니다. 이 경우에는 나도 똑같이 했습니다. 봇 코드에는 SubgraphGPT 봇의 프로덕션 버전에서 사용하는 모든 특수 사례가 포함되어 있지 않습니다. 왜냐하면 봇 코드에는 다른 애완 동물 프로젝트에서 상속받은 논리가 상당히 많기 때문입니다. 대신 필요에 따라 쉽게 수정할 수 있는 최소한의 기본 코드만 남겨 두었습니다.

telegram-bot.py 스크립트는 여러 부분으로 구성됩니다. 먼저 이전과 마찬가지로 라이브러리를 가져오고 명령줄 인수가 초기화됩니다.

 import threading import telegram from telegram.ext import Updater, CommandHandler, MessageHandler, Filters import openai from openai.embeddings_utils import cosine_similarity import numpy as np import pandas as pd import argparse import functools # Create an Argument Parser object parser = argparse.ArgumentParser(description='Run the bot which uses prepared knowledge base enriched with contextual embeddings') # Add the arguments parser.add_argument('--openai_api_key', type=str, help='API KEY of OpenAI API to create contextual embeddings for each line') parser.add_argument('--telegram_bot_token', type=str, help='A telegram bot token obtained via @BotFather') parser.add_argument('--file', type=str, help='A source CSV file with the questions, answers and embeddings') parser.add_argument('--topic', type=str, help='Write the topic to add a default context for the bot') parser.add_argument('--start_message', type=str, help="The text that will be shown to the users after they click /start button/command", default="Hello, World!") parser.add_argument('--model', type=str, help='A model of ChatGPT which will be used', default='gpt-3.5-turbo-16k') parser.add_argument('--num_top_qa', type=str, help="The number of top similar questions' answers as a context", default=3) # Parse the command-line arguments args = parser.parse_args() # Access the argument values openai.api_key = args.openai_api_key token = args.telegram_bot_token file = args.file topic = args.topic model = args.model num_top_qa = args.num_top_qa start_message = args.start_message

이 경우 OpenAI API 키도 필요합니다. 지식 베이스에서 사용자가 방금 입력한 질문과 가장 유사한 질문을 찾으려면 먼저 다음을 호출하여 해당 질문의 임베딩을 얻어야 합니다. 지식 기반 자체에 대해 했던 것처럼 API도 마찬가지입니다.

또한 다음이 필요합니다.

• telegram_bot_token - BotFather의 텔레그램 봇용 토큰
• file - 지식 기반 파일의 경로(여기서 Discord에서 보낸 메시지는 틈새 작업이라고 가정하기 때문에 의도적으로 건너뛰었지만 필요한 경우 코드에 쉽게 통합할 수 있습니다.)
• topic - 봇이 작동할 주제(기사 시작 부분에 언급됨)의 텍스트 형식입니다.
• start_message - /start를 클릭한 사용자에게 표시되는 메시지(기본적으로 "Hello, World!")
• model - 모델 선택(기본적으로 설정됨)
• num_top_qa - ChatGPT 요청의 컨텍스트로 사용될 지식 베이스의 가장 유사한 질문-답변 수

그런 다음 지식 기반 파일을 로드하고 질문 임베딩을 초기화합니다.

 # reading QA file with embeddings df_qa = pd.read_csv(file) df_qa['ada_embedding'] = df_qa.ada_embedding.apply(eval).apply(np.array)

ChatGPT API에 요청을 하기 위해 가끔 과부하로 인해 오류로 응답하는 것을 알고, 오류 발생 시 자동으로 요청을 재시도하는 기능을 사용합니다.

 def retry_on_error(func): @functools.wraps(func) def wrapper(*args, **kwargs): max_retries = 3 for i in range(max_retries): try: return func(*args, **kwargs) except Exception as e: print(f"Error occurred, retrying ({i+1}/{max_retries} attempts)...") # If all retries failed, raise the last exception raise e return wrapper @retry_on_error def call_chatgpt(*args, **kwargs): return openai.ChatCompletion.create(*args, **kwargs)

OpenAI의 권장 사항에 따르면 텍스트를 임베딩으로 변환하기 전에 새 줄을 공백으로 바꿔야 합니다.

 def get_embedding(text, model="text-embedding-ada-002"): text = text.replace("\n", " ") return openai.Embedding.create(input=[text], model=model)['data'][0]['embedding']

가장 유사한 질문을 검색하기 위해 openai 라이브러리에서 직접 가져온 두 질문의 임베딩 사이의 코사인 거리를 계산합니다.

 def search_similar(df, question, n=3, pprint=True): embedding = get_embedding(question, model='text-embedding-ada-002') df['similarities'] = df.ada_embedding.apply(lambda x: cosine_similarity(x, embedding)) res = df.sort_values('similarities', ascending=False).head(n) return res

주어진 것과 가장 유사한 질문-답변 쌍의 목록을 받은 후 이를 하나의 텍스트로 컴파일하여 ChatGPT가 무엇이 무엇인지 명확하게 결정할 수 있는 방식으로 표시할 수 있습니다.

 def collect_text_qa(df): text = '' for i, row in df.iterrows(): text += f'Q: <'+row['Question'] + '>\nA: <'+ row['Answer'] +'>\n\n' print('len qa', len(text.split(' '))) return text

그 후에는 기사의 시작 부분에 설명된 프롬프트의 "조각"을 하나의 전체로 모아야 합니다.

 def collect_full_prompt(question, qa_prompt, chat_prompt=None): prompt = f'I need to get an answer to the question related to the topic of "{topic}": ' + "{{{"+ question +"}}}. " prompt += '\n\nPossibly, you might find an answer in these Q&As [use the information only if it is actually relevant and useful for the question answering]: \n\n' + qa_prompt # edit if you need to use this also if chat_prompt is not None: prompt += "---------\nIf you didn't find a clear answer in the Q&As, possibly, these talks from chats might be helpful to answer properly [use the information only if it is actually relevant and useful for the question answering]: \n\n" + chat_prompt prompt += f'\nFinally, only if the information above was not enough you can use your knowledge in the topic of "{topic}" to answer the question.' return prompt

이 경우 Discord의 메시지를 사용하여 해당 부분을 제거했지만 chat_prompt != None이면 논리를 계속 따를 수 있습니다.

또한 ChatGPT API에서 받은 응답을 텔레그램 메시지(4096자 이하)로 분할하는 기능이 필요합니다.

 def telegram_message_format(text): max_message_length = 4096 if len(text) > max_message_length: parts = [] while len(text) > max_message_length: parts.append(text[:max_message_length]) text = text[max_message_length:] parts.append(text) return parts else: return [text]

봇은 일반적인 단계 순서로 시작하여 /start 명령으로 실행될 두 가지 기능을 할당하고 사용자로부터 개인 메시지를 받습니다.

 bot = telegram.Bot(token=token) updater = Updater(token=token, use_context=True) dispatcher = updater.dispatcher dispatcher.add_handler(CommandHandler("start", start, filters=Filters.chat_type.private)) dispatcher.add_handler(MessageHandler(~Filters.command & Filters.text, message_handler)) updater.start_polling()

/start 에 응답하는 코드는 간단합니다.

 def start(update, context): user = update.effective_user context.bot.send_message(chat_id=user.id, text=start_message)

자유 형식 메시지에 대한 응답도 명확하지 않습니다.

첫째 , 다른 사용자의 스레드를 차단하지 않으려면 스레딩 라이브러리를 사용하여 스레드를 독립적인 프로세스로 즉시 "분리"해 보겠습니다.

 def message_handler(update, context): thread = threading.Thread(target=long_running_task, args=(update, context)) thread.start()

둘째 , 모든 로직은 long_running_task 함수 내에서 발생합니다. 봇의 코드를 수정할 때 오류를 쉽게 지역화할 수 있도록 try/exc 에 주요 조각을 의도적으로 래핑했습니다.

• 먼저, 사용자가 메시지 대신 파일이나 이미지를 보내는 경우 메시지를 검색하고 오류를 처리합니다.
• 그런 다음 search_similar 를 사용하여 가장 유사한 질문-답변을 검색합니다.
• 그 후, Collect_text_qa를 사용하여 모든 질문-답변을 하나의 텍스트로 수집합니다.
• 그리고 Collect_full_prompt 를 사용하여 ChatGPT API에 대한 최종 프롬프트를 생성합니다.

 def long_running_task(update, context): user = update.effective_user context.bot.send_message(chat_id=user.id, text='🕰️⏰🕙⏱️⏳...') try: question = update.message.text.strip() except Exception as e: context.bot.send_message(chat_id=user.id, text=f"🤔It seems like you're sending not text to the bot. Currently, the bot can only work with text requests.") return try: qa_found = search_similar(df_qa, question, n=num_top_qa) qa_prompt = collect_text_qa(qa_found) full_prompt = collect_full_prompt(question, qa_prompt) except Exception as e: context.bot.send_message(chat_id=user.id, text=f"Search failed. Debug needed.") return

예를 들어 형식 지정으로 인해 지식 베이스와 주제를 자신의 것으로 바꾸는 경우 오류가 발생할 수 있으므로 사람이 읽을 수 있는 오류가 표시됩니다.

다음으로 요청은 이미 입증된 주요 시스템 메시지와 함께 ChatGPT API로 전송됩니다. " 당신은 도움이 되는 조수입니다. " 결과 출력은 필요한 경우 여러 메시지로 나누어 사용자에게 다시 전송됩니다.

 try: print(full_prompt) completion = call_chatgpt( model=model, n=1, messages=[{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": full_prompt}] ) result = completion['choices'][0]['message']['content'] except Exception as e: context.bot.send_message(chat_id=user.id, text=f'It seems like the OpenAI service is responding with errors. Try sending the request again.') return parts = telegram_message_format(result) for part in parts: update.message.reply_text(part, reply_to_message_id=update.message.message_id)

이것으로 코드 부분을 마칩니다.

원기

이제 이러한 봇의 프로토타입이 다음 링크 에서 제한된 형식으로 제공됩니다. API가 유료라서 하루 최대 3개까지 요청이 가능하지만 누구에게나 제한은 없을 것 같아요. 가장 흥미로운 건 좁은 주제에 집중하는 전문 봇이 아니라 AnythingGPT 프로젝트의 코드이기 때문이죠. , 이 예제를 기반으로 한 지식 기반으로 특정 작업을 해결하기 위해 자신만의 봇을 만드는 방법에 대한 간단한 지침과 함께 GitHub 에서 사용할 수 있습니다. 끝까지 읽으셨다면, 이 글이 도움이 되셨기를 바랍니다.