CLOVA Studio 운영자 Posted June 14 공유하기 Posted June 14 들어가며 CLOVA Studio에서 제공하는 Chat Completions API만을 단독으로 사용하면 최대 4096 토큰까지만 대화를 진행할 수 있어, 맥락 유지에 한계가 있습니다. 하지만 이번 cookbook에서 다룰 요약 API와 Chat Completions API를 함께 활용하면, 이전 대화 내용을 요약해가며 맥락을 보존하기 때문에 훨씬 더 길고 깊이 있는 대화가 가능합니다. 요약 API는 CLOVAStudio에서 제공하는 API로, 많은 양의 텍스트를 효과적으로 요약할 수 있습니다. 작동 원리 사용자와 어시스턴트의 대화는 일정량까지 누적되어 저장됩니다. 이 일정량은 모델이 한 번에 처리할 수 있는 토큰 수에서 Chat Completions의 'maxTokens'을 뺀 값을 기준으로 결정됩니다. 만약 사용자의 요청이 이 기준을 초과할 경우, 요약 API가 자동으로 작동하여 대화를 요약하고, 요약된 내용은 시스템 프롬프트에 포함됩니다. 이후 대화는 다시 기준까지 누적되며 진행되고, 기준을 초과한 사용자의 요청에 대한 어시스턴트의 답변은 요약 이후에 화면에 출력됩니다. 이를 통해 대화의 맥락을 유지하면서도 토큰 제한을 효율적으로 관리할 수 있습니다. 전체 구조 Quote Python 3.12.2 필요한 API들을 별도의 파일로 분리하여 모듈화하였고, 이를 필요에 따라 불러올 수 있도록 구성하였습니다. 관리의 편의성을 위해 요약 API와 Chat Completions API 기능을 각각 별도의 파일로 분리하고, main.py에서 필요한 모듈을 불러와 사용할 수 있도록 구성하였습니다. project-root/ │ ├── clovastudio_executor.py ├── completion_executor.py ├── summary_executor.py └── main.py 각 모듈 설명 1. 상위 클래스 정의 1부에서 정의한 CLOVAStudioExecutor입니다. 여러 API들의 호출에 필요한 공통 기능을 묶어 상위 클래스로 정의한 것으로, 1부에서 소개된 코드와 동일합니다. import json import http.client from http import HTTPStatus class CLOVAStudioExecutor: def __init__(self, host, api_key, api_key_primary_val, request_id): self._host = host self._api_key = api_key self._api_key_primary_val = api_key_primary_val self._request_id = request_id def _send_request(self, completion_request, endpoint): headers = { 'Content-Type': 'application/json; charset=utf-8', 'X-NCP-CLOVASTUDIO-API-KEY': self._api_key, 'X-NCP-APIGW-API-KEY': self._api_key_primary_val, 'X-NCP-CLOVASTUDIO-REQUEST-ID': self._request_id } conn = http.client.HTTPSConnection(self._host) conn.request('POST', endpoint, json.dumps(completion_request), headers) response = conn.getresponse() status = response.status result = json.loads(response.read().decode(encoding='utf-8')) conn.close() return result, status def execute(self, completion_request, endpoint): res, status = self._send_request(completion_request, endpoint) if status == HTTPStatus.OK: return res, status else: error_message = res.get("status", {}).get("message", "Unknown error") if isinstance(res, dict) else "Unknown error" raise ValueError(f"오류 발생: HTTP {status}, 메시지: {error_message}") 2. Chat Completions API Chat Completions API를 정의하는 클래스입니다. 1부와 역시 동일합니다. 전체 프로젝트 폴더 내에 completion_executor.py로 저장합니다. from clovastudio_executor import CLOVAStudioExecutor from http import HTTPStatus import requests class ChatCompletionExecutor(CLOVAStudioExecutor): def __init__(self, host, api_key, api_key_primary_val, request_id): super().__init__(host, api_key, api_key_primary_val, request_id) def execute(self, completion_request, stream=True): headers = { 'X-NCP-CLOVASTUDIO-API-KEY': self._api_key, 'X-NCP-APIGW-API-KEY': self._api_key_primary_val, 'X-NCP-CLOVASTUDIO-REQUEST-ID': self._request_id, 'Content-Type': 'application/json; charset=utf-8', 'Accept': 'text/event-stream' if stream else 'application/json' } with requests.post(self._host + '/testapp/v1/chat-completions/HCX-003', headers=headers, json=completion_request, stream=stream) as r: if stream: if r.status_code == HTTPStatus.OK: response_data = "" for line in r.iter_lines(): if line: decoded_line = line.decode("utf-8") print(decoded_line) response_data += decoded_line + "\n" return response_data else: raise ValueError(f"오류 발생: HTTP {r.status_code}, 메시지: {r.text}") else: if r.status_code == HTTPStatus.OK: return r.json() else: raise ValueError(f"오류 발생: HTTP {r.status_code}, 메시지: {r.text}") 3. 요약 API 요약 API를 정의하는 클래스입니다. 전체 프로젝트 폴더 내에 summary_executor.py로 저장합니다. from clovastudio_executor import CLOVAStudioExecutor from http import HTTPStatus class SummarizationExecutor(CLOVAStudioExecutor): def execute(self, completion_request): endpoint = '/testapp/v1/api-tools/summarization/v2/{테스트앱 식별자}' res, status = super().execute(completion_request, endpoint) if status == HTTPStatus.OK and "result" in res: return res["result"]["text"] else: error_message = res.get("status", {}).get("message", "Unknown error") if isinstance(res, dict) else "Unknown error" raise ValueError(f"오류 발생: HTTP {status}, 메시지: {error_message}") Quote 요약 API를 테스트앱으로 발급받은 후, 이에 해당하는 테스트앱 식별자를 {테스트앱 식별자} 부분에 넣어야합니다. 4. main.py 앞서 정의한 3개의 모듈을 활용하여 토큰 한도를 초과하는 대화를 가능하게 하는 로직이 담긴 main 파일입니다. 아래는 main.py의 전체 코드이며, 각 부분에 대한 자세한 설명을 이어서 하겠습니다. import json from completion_executor import ChatCompletionExecutor from summary_executor import SummarizationExecutor # 세션 state 정의 session_state = { "preset_messages": [], "total_tokens": 0, "chat_log": [], "started": False, "summary_messages": [], "last_user_input": "", "last_response": "", "last_user_message": {}, "last_assistant_message": {}, "previous_messages": [], "system_message_changed": False, "system_message": "" } def log_preset_messages(): print("preset_messages:", json.dumps(session_state['preset_messages'], ensure_ascii=False, indent=2)) print("system_message:", session_state['system_message']) print("total_tokens:", session_state['total_tokens']) def summarize_and_reset(summarization_executor): text_to_summarize = " ".join([msg.get('content', '') for msg in session_state['preset_messages'][1:] if msg['role'] != 'system']) # 첫 system 메시지 제외 print(f"Text to summarize: {text_to_summarize}") summary_request_data = { "texts": [text_to_summarize], "autoSentenceSplitter": True, "segCount": -1, "segMaxSize": 1000, "segMinSize": 300, "includeAiFilters": False } summary_response = summarization_executor.execute(summary_request_data) print(f"Summary response: {summary_response}") if summary_response and isinstance(summary_response, str): summary_text = summary_response session_state['summary_messages'].append({"role": "system", "content": summary_text}) session_state['preset_messages'] = [{"role": "system", "content": session_state['system_message']}, {"role": "system", "content": summary_text}] session_state['preset_messages'].append(session_state.get('last_user_message', {})) session_state['preset_messages'].append(session_state.get('last_assistant_message', {})) session_state['total_tokens'] = len(summary_text.split()) + \ len(session_state.get('last_user_message', {}).get('content', '').split()) + \ len(session_state.get('last_assistant_message', {}).get('content', '').split()) log_preset_messages() def parse_stream_response(response): input_length = 0 output_length = 0 content_parts = [] for line in response.splitlines(): if line.startswith('data:'): data = json.loads(line[5:]) if 'message' in data and 'content' in data['message']: content_parts.append(data['message']['content']) if 'inputLength' in data: input_length = data['inputLength'] if 'outputLength' in data: output_length = data['outputLength'] content = content_parts[-1] return { 'inputLength': input_length, 'outputLength': output_length, 'content': content } def parse_non_stream_response(response): result = response.get('result', {}) message = result.get('message', {}) content = message.get('content', '') input_length = result.get('inputLength', 0) output_length = result.get('outputLength', 0) return { 'inputLength': input_length, 'outputLength': output_length, 'content': content } def main(): system_message = input("Enter the initial system message (예시: 친절하게 답변하는 클로바 AI 어시스턴트입니다.): ") session_state['system_message'] = system_message session_state['system_message_changed'] = True session_state['started'] = True if session_state['system_message_changed']: session_state['preset_messages'] = [msg for msg in session_state['preset_messages'] if msg['role'] != 'system'] session_state['preset_messages'].insert(0, {"role": "system", "content": session_state['system_message']}) session_state['system_message_changed'] = False if session_state['started']: completion_executor = ChatCompletionExecutor( host='https://clovastudio.stream.ntruss.com', api_key='<api_key>', api_key_primary_val='<api_key_primary_val>', request_id='<request_id>' ) summarization_executor = SummarizationExecutor( host='clovastudio.apigw.ntruss.com', api_key='<api_key>', api_key_primary_val='<api_key_primary_val>', request_id='<request_id>' ) while True: user_input = input("사용자: ") if user_input.lower() in ['exit', 'quit']: break session_state['last_user_input'] = user_input session_state['last_user_message'] = {"role": "user", "content": user_input} session_state['preset_messages'].append(session_state['last_user_message']) session_state['chat_log'].append(session_state['last_user_message']) request_data = { "messages": session_state['preset_messages'], "maxTokens": 256, "temperature": 0.5, "topK": 0, "topP": 0.6, "repeatPenalty": 1.2, "stopBefore": [], "includeAiFilters": True, "seed": 0, } def execute_request(stream): try: response = completion_executor.execute(request_data, stream=stream) if stream: parsed_response = parse_stream_response(response) else: parsed_response = parse_non_stream_response(response) return parsed_response except Exception as e: print(f"Error occurred: {e}") return None # 여기서 stream 옵션을 True나 False로 설정 response = execute_request(stream=True) if response: response_text = response['content'] input_length = response.get('inputLength', 0) output_length = response.get('outputLength', 0) session_state['last_response'] = response_text session_state['last_assistant_message'] = {"role": "assistant", "content": response_text} session_state['preset_messages'].append(session_state['last_assistant_message']) session_state['chat_log'].append(session_state['last_assistant_message']) session_state['total_tokens'] = input_length + output_length log_preset_messages() token_limit = 4096 - request_data["maxTokens"] # 토큰 한도를 고려하여 설정 if session_state['total_tokens'] > token_limit: print("Token limit exceeded. Starting summarization.") summarize_and_reset(summarization_executor) else: print("Error occurred. Starting summarization.") summarize_and_reset(summarization_executor) session_state['preset_messages'].append(session_state['last_user_message']) if __name__ == "__main__": main() main.py 호출 구조 이제 main.py의 자세한 호출 구조를 설명하겠습니다. ▼ 이 단계에서는 앞서 별도의 Python 파일로 저장한 모듈들을 불러옵니다. 또한, 대화 로그를 관리하고 토큰 수를 계산하기 위한 세션을 정의합니다. 이렇게 필요한 모듈과 세션을 준비함으로써, 원활한 대화 진행과 요약 기능 활용을 위한 기반을 마련하게 됩니다. import json from completion_executor import ChatCompletionExecutor from summary_executor import SummarizationExecutor # 세션 state 정의 session_state = { "preset_messages": [], "total_tokens": 0, "chat_log": [], "started": False, "summary_messages": [], "last_user_input": "", "last_response": "", "last_user_message": {}, "last_assistant_message": {}, "previous_messages": [], "system_message_changed": False, "system_message": "" } ▼ 이 함수는 대화가 진행되는 동안 터미널 상에서 대화 로그가 정상적으로 기록되고, 토큰 수가 올바르게 계산되고 있는지 확인하기 위해 사용됩니다. 이 정보는 사용자의 대화 화면에는 표시되지 않습니다. def log_preset_messages(): print("preset_messages:", json.dumps(session_state['preset_messages'], ensure_ascii=False, indent=2)) print("system_message:", session_state['system_message']) print("total_tokens:", session_state['total_tokens']) ▼ 진행된 대화 내용은 preset_messages 변수에 저장됩니다. 대화 내역이 최대 토큰 수에 도달하면, 이 함수를 통해 preset_message에 저장된 모든 대화 내역을 요약된 내용으로 대체합니다. 요약된 내용은 시스템 프롬프트("role": "system")에 입력되어, 어시스턴트가 이전 대화 맥락을 기억한 채로 사용자와 계속해서 대화를 이어갈 수 있도록 합니다. 이를 통해 대화의 연속성과 일관성을 유지하면서도 토큰 제한을 효과적으로 관리할 수 있습니다. def summarize_and_reset(summarization_executor): text_to_summarize = " ".join([msg.get('content', '') for msg in session_state['preset_messages'][1:] if msg['role'] != 'system']) # 첫 system 메시지 제외 print(f"Text to summarize: {text_to_summarize}") summary_request_data = { "texts": [text_to_summarize], "autoSentenceSplitter": True, "segCount": -1, "segMaxSize": 1000, "segMinSize": 300, "includeAiFilters": False } summary_response = summarization_executor.execute(summary_request_data) print(f"Summary response: {summary_response}") if summary_response and isinstance(summary_response, str): summary_text = summary_response session_state['summary_messages'].append({"role": "system", "content": summary_text}) session_state['preset_messages'] = [{"role": "system", "content": session_state['system_message']}, {"role": "system", "content": summary_text}] session_state['preset_messages'].append(session_state.get('last_user_message', {})) session_state['preset_messages'].append(session_state.get('last_assistant_message', {})) session_state['total_tokens'] = len(summary_text.split()) + \ len(session_state.get('last_user_message', {}).get('content', '').split()) + \ len(session_state.get('last_assistant_message', {}).get('content', '').split()) Quote 요약 API의 파라미터인 autoSentenceSplitter, segCount, segMaxSize, segMinSize 등은 사용자가 원하는 대로 조정할 수 있습니다. 요약된 내용은 시스템 프롬프트로 사용되기 때문에, 요약문의 길이가 길어질수록 더 많은 토큰이 소모됩니다. 따라서 대화형 애플리케이션의 목적과 특성에 맞게 파라미터를 적절히 설정하는 것이 중요합니다. ▼ 다음은 stream 옵션을 실행했을 때와 실행하지 않았을 때 각각의 요청을 처리하기 위한 함수입니다. 토큰 계산 확인을 위해 Chat Completions API의 응답 바디중 inputLength와 outputLength를 추가적으로 가져옵니다. def parse_stream_response(response): input_length = 0 output_length = 0 content_parts = [] for line in response.splitlines(): if line.startswith('data:'): data = json.loads(line[5:]) if 'message' in data and 'content' in data['message']: content_parts.append(data['message']['content']) if 'inputLength' in data: input_length = data['inputLength'] if 'outputLength' in data: output_length = data['outputLength'] content = content_parts[-1] return { 'inputLength': input_length, 'outputLength': output_length, 'content': content } def parse_non_stream_response(response): result = response.get('result', {}) message = result.get('message', {}) content = message.get('content', '') input_length = result.get('inputLength', 0) output_length = result.get('outputLength', 0) return { 'inputLength': input_length, 'outputLength': output_length, 'content': content } ▼ 이 부분은 대화형 애플리케이션의 핵심 로직을 담고 있는 중요한 코드 블록으로, 앞서 소개한 함수들을 포함하여 토큰 수를 계산하고 답변을 출력하며, 최대 토큰 수에 도달했을 때 요약을 수행하는 등의 기능을 수행합니다. def main(): system_message = input("Enter the initial system message (예시: 친절하게 답변하는 클로바 AI 어시스턴트입니다.): ") session_state['system_message'] = system_message session_state['system_message_changed'] = True session_state['started'] = True if session_state['system_message_changed']: session_state['preset_messages'] = [msg for msg in session_state['preset_messages'] if msg['role'] != 'system'] session_state['preset_messages'].insert(0, {"role": "system", "content": session_state['system_message']}) session_state['system_message_changed'] = False if session_state['started']: completion_executor = ChatCompletionExecutor( host='https://clovastudio.stream.ntruss.com', api_key='<api_key>', api_key_primary_val='<api_key_primary_val>', request_id='<request_id>' ) summarization_executor = SummarizationExecutor( host='clovastudio.apigw.ntruss.com', api_key='<api_key>', api_key_primary_val='<api_key_primary_val>', request_id='<request_id>' ) while True: user_input = input("사용자: ") if user_input.lower() in ['exit', 'quit']: break session_state['last_user_input'] = user_input session_state['last_user_message'] = {"role": "user", "content": user_input} session_state['preset_messages'].append(session_state['last_user_message']) session_state['chat_log'].append(session_state['last_user_message']) request_data = { "messages": session_state['preset_messages'], "maxTokens": 256, "temperature": 0.5, "topK": 0, "topP": 0.6, "repeatPenalty": 1.2, "stopBefore": [], "includeAiFilters": True, "seed": 0, } def execute_request(stream): try: response = completion_executor.execute(request_data, stream=stream) if stream: parsed_response = parse_stream_response(response) else: parsed_response = parse_non_stream_response(response) return parsed_response except Exception as e: print(f"Error occurred: {e}") return None # 여기서 stream 옵션을 True나 False로 설정 response = execute_request(stream=True) if response: response_text = response['content'] input_length = response.get('inputLength', 0) output_length = response.get('outputLength', 0) session_state['last_response'] = response_text session_state['last_assistant_message'] = {"role": "assistant", "content": response_text} session_state['preset_messages'].append(session_state['last_assistant_message']) session_state['chat_log'].append(session_state['last_assistant_message']) session_state['total_tokens'] = input_length + output_length log_preset_messages() token_limit = 4096 - request_data["maxTokens"] # 토큰 한도를 고려하여 설정 if session_state['total_tokens'] > token_limit: print("Token limit exceeded. Starting summarization.") summarize_and_reset(summarization_executor) else: print("Error occurred. Starting summarization.") summarize_and_reset(summarization_executor) session_state['preset_messages'].append(session_state['last_user_message']) Quote system_message는 대화 시작 시에 사용되는 시스템 프롬프트로, 대화 도중에도 변경할 수 있습니다. 요약은 모델의 최대 토큰 수인 4096에서 Chat Completions API의 maxTokens 파라미터 값을 뺀 token_limit이 현재 대화의 전체 토큰 수인 total_tokens를 초과할 때 발동됩니다. 따라서 maxTokens 값에 따라 요약이 실행되는 토큰 수 한도가 달라지므로, 대화의 목적에 맞게 이 값을 설정해야 합니다. 추가적으로, 답변을 토큰 단위로 출력받는 stream 옵션의 기본값은 True로, stream 여부는 코드 중간 부분에서 확인할 수 있듯 사용자가 직접 설정할 수 있습니다. Chat Completions API와 요약 API의 각각 테스트앱에 해당하는 <request_id>를 입력해야 정상적으로 각 API가 작동합니다. main 함수를 실행하는 것까지 main.py에 포함됩니다. if __name__ == "__main__": main() 요약 활용 대화 실행 결과 이제 요약 기능을 활용하여 대화 맥락을 유지하는 방법에 대한 실행 결과를 보여드리겠습니다. 대화의 진행 과정을 보다 명확하게 확인할 수 있도록 Streamlit을 사용하여 채팅 화면을 구현했습니다. 앞서 정의한 executor들은 그대로 유지하며, main.py 파일에 아래 제시된 코드를 그대로 저장하면 Streamlit으로 포장된 대화를 실행할 수 있습니다. Quote Streamlit은 Python 기반의 오픈소스 웹 어플리케이션 라이브러리입니다. 손쉽게 UI 인터페이스를 만들고 프로토타이핑, 시각화 등이 가능합니다. cd 프로젝트/폴더/경로로/이동 streamlit run main.py Streamlit Code 열어보기 ▼ 대화가 진행되는 동안, 터미널에는 모델이 이후 대화에 참고할 수 있는 정보들이 출력됩니다. 여기에는 이전 대화 내용을 담고 있는 대화 로그(preset_messages), 현재 설정된 시스템 프롬프트(system_message), 그리고 누적된 총 토큰 수(total_tokens)가 포함됩니다. ▼ 대화 내역이 한도를 초과하기 전까지는 'Text to summarize'에 저장되며, 이를 요약한 내용은 'Summary response'로 출력됩니다. 이 요약된 내용은 시스템 프롬프트("role: system")로 사용되어 대화의 맥락을 유지하는 데 활용됩니다. 터미널에서의 가독성을 높이기 위해, 요약이 실행되는 시점(한도 초과 시)에만 누적된 대화 로그를 'Text to summarize'로 표시하고, 요약된 내용이 시스템 프롬프트로 사용된 것은 'preset_messages'를 통해 확인할 수 있도록 구성하였습니다. ▼ 요약이 실행되면 토큰 수도 초기화되어 다시 누적 계산됩니다. 맺음말 본 cookbook에서는 CLOVAStudio의 요약 API를 활용하여 토큰 한도를 넘어서는 대화를 가능하게 하는 방법에 대해 알아보았습니다. 요약 API의 파라미터와 Chat Completions API의 파라미터 설정에 따라 대화 맥락의 보존 정도와 연속 대화 가능 길이가 달라질 수 있습니다. 따라서 대화 서비스의 목적과 특성에 맞게 이러한 파라미터들을 적절히 조정하는 것이 중요합니다. 앞으로도 CLOVA Studio는 다양한 API와 도구를 제공하여, 사용자들이 더욱 풍부하고 인상적인 대화형 AI 서비스를 구현할 수 있도록 지원하겠습니다. 1 링크 복사 다른 사이트에 공유하기 More sharing options...
Recommended Posts
게시글 및 댓글을 작성하려면 로그인 해주세요.
로그인