from typing import Any, Dict, List import copy import json import os import time import traceback from openai import OpenAI import tiktoken from lpm_kernel.api.services.user_llm_config_service import UserLLMConfigService from lpm_kernel.configs.config import Config from lpm_kernel.L0.models import InsighterInput, SummarizerInput from lpm_kernel.L0.prompt import * from lpm_kernel.utils import ( DataType, IntentType, TokenParagraphSplitter, TokenTextSplitter, cal_upperbound, chunk_filter, equidistant_filter, get_safe_content_turncate, get_summarize_title_keywords, select_language_desc, ) from lpm_kernel.configs.logging import get_train_process_logger logger = get_train_process_logger() class L0Generator: def __init__(self, preferred_language="English"): """Initialize L0Generator with language preference. Args: preferred_language: The language to use for generation, defaults to English. """ self.preferred_language = preferred_language # Initialize tokenizer self._tokenizer = tiktoken.get_encoding("cl100k_base") # OpenAI default tokenizer self.lf_prompt_image_parser = insight_image_parser self.lf_prompt_image_overview = insight_image_overview self.lf_prompt_image_breakdown = insight_image_breakdown self.lf_prompt_audio_parser = insight_audio_parser self.lf_prompt_audio_overview = insight_audio_overview self.lf_prompt_audio_breakdown = insight_audio_breakdown self.lf_prompt_doc_overview = insight_doc_overview self.lf_prompt_doc_breakdown = insight_doc_breakdown self.max_retries_summarize = 2 self.timeout_summarize = 30 self.user_llm_config_service = UserLLMConfigService() self.user_llm_config = self.user_llm_config_service.get_available_llm() if self.user_llm_config is None: self.client = None self.model_name = None else: self.client = OpenAI( api_key=self.user_llm_config.chat_api_key, base_url=self.user_llm_config.chat_endpoint, ) self.model_name = self.user_llm_config.chat_model_name def _insighter_image( self, bio: Dict[str, str], content: str, max_retries: int, request_timeout: int, file_content: str ) -> tuple[str, str]: """Process image content to generate insights. Args: bio: Dictionary containing user biography information content: Text content related to the image max_retries: Maximum number of API call retries request_timeout: Timeout for API calls in seconds file_content: URL or base64 content of the image Returns: Tuple of (summary, title) strings """ hint_prompt = f"# Hint #\n{content}\n# Instruction #\n" language_desc = select_language_desc(self.preferred_language) segment_list = [ self.lf_prompt_image_parser, self.lf_prompt_image_overview, self.lf_prompt_image_breakdown, ] messages_list = [] for i in range(len(segment_list)): image_parser_prompt = segment_list[i] if "__global_bio__" in image_parser_prompt: image_parser_prompt = image_parser_prompt.replace( "__about_me__", bio["about_me"] ) image_parser_prompt = image_parser_prompt.replace( "__global_bio__", bio["global_bio"] ) image_parser_prompt = image_parser_prompt.replace( "__status_bio__", bio["status_bio"] ) # system prompt language = language_desc if i != 0 else "English" messages = [ {"role": "system", "content": image_parser_prompt}, { "role": "user", "content": [ { "type": "text", "text": hint_prompt + "Here are some images and their Hint. Please follow the WorkFlow and do your best. Ensure that your response is in a parseable JSON format." + language, } ], }, ] if i == 0: new_messages = copy.deepcopy(messages) new_messages[-1]["content"].append( { "type": "image_url", "image_url": { "url": file_content, # file_content is the image url }, } ) messages_list.append(new_messages) else: messages[-1]["content"].append( { "type": "image_url", "image_url": { "url": file_content, # file_content is the image url }, } ) messages_list.append(messages) results = [] for messages in messages_list: response = self.client.chat.completions.create( model=self.model_name, messages=messages, max_tokens=4096, temperature=0.0, max_retries=max_retries, timeout=request_timeout, response_format={"type": "json_object"}, ) results.append(response.choices[0].message.content) try: images_intent_list = [] for image_id in range(len(results) - 2): images_intent_list.append(results[image_id]["image"].get("Step 3", "")) title = results[-2].get("Title", "") opening = results[-2].get("Opening", "") insight = results[-1].get("Insight", []) insight = "- " + "\n- ".join(insight) if insight else "" summary = "\n\n".join([opening, insight]) return summary, title except Exception as e: logger.error(f"Unexpected error: {e}") raise RuntimeError(f"Unexpected error: {e}") def _insighter_audio( self, bio: str, content: str, max_retries: int, request_timeout: int, file_content: Dict[str, Any] ) -> tuple[str, str]: """Process audio content to generate insights. Args: bio: User biography information content: Text content related to the audio max_retries: Maximum number of API call retries request_timeout: Timeout for API calls in seconds file_content: Dictionary containing audio metadata and content Returns: Tuple of (insight, title) strings """ user_info = """# Hint # "{content}" # Speech # "{speech}" # User Instruction # '{user_input}' """ user_input = "Here are some speech and their hint. Please follow the WorkFlow and do your best. Ensure that your response is in a parseable JSON format. " language_desc = select_language_desc(self.preferred_language) speech_dict = file_content["metadata"]["audio"].get("segmentList", []) speech = "" end_point = 0 # Raise exception if speech is empty or too short if not speech_dict: raise ValueError("Invalid input: speech must not be empty") for segment in speech_dict: start_time = int(segment["segmentStartTime"]) end_time = int(segment["segmentEndTime"]) segment_content = segment["segmentContent"] tmp = f"[{start_time}-{end_time}]: {segment_content}\n" speech += tmp end_point = int(end_time) logger.info(f"length of speech: {end_point}") # Split speech over 1200s into segments, maximum 1200s each num_segments = 1 if end_point > 1200: num_segments = max(2, int(round(end_point / 1200.0))) segment_duration = end_point / num_segments speech_segments = ["" for _ in range(num_segments)] for segment in speech_dict: start_time = int(segment["segmentStartTime"]) end_time = int(segment["segmentEndTime"]) segment_content = segment["segmentContent"] segment_index = min( num_segments - 1, int(start_time // segment_duration) ) speech_segments[ segment_index ] += f"[{start_time}-{end_time}]: {segment_content}\n" user_info_overall = user_info.format( content=content, speech=speech, user_input=user_input ) audio_parser_prompt_overview = self.lf_prompt_audio_overview.replace( "__bio__", bio ) messages_overall = [ {"role": "system", "content": audio_parser_prompt_overview}, { "role": "user", "content": [ {"type": "text", "text": user_info_overall + language_desc} ], }, ] message_list = [messages_overall] max_retry_list = [2] for i in range(num_segments): user_info_segment = user_info.format( content=content, speech=speech_segments[i], user_input=user_input ) messages_segment = [ {"role": "system", "content": self.lf_prompt_audio_breakdown}, { "role": "user", "content": [ {"type": "text", "text": user_info_segment + language_desc} ], }, ] message_list.append(messages_segment) max_retry_list.append(2) results = [] for messages in message_list: response = self.client.chat.completions.create( model=self.model_name, messages=messages, max_tokens=4096, temperature=0.0, max_retries=max_retries, timeout=request_timeout, response_format={"type": "json_object"}, ) results.append(response.choices[0].message.content) try: title = results[0].get("Title", "") overview = results[0].get("Overview", "") breakdown = {} for res_p in results[1:]: breakdown = {**breakdown, **res_p.get("Breakdown", {})} tmpl = "{}\n{}" formated_breakdown = "" for subtitle, key_points in breakdown.items(): formated_breakdown += f"\n**{subtitle}**\n" for key_point in key_points: if len(key_point) != 3: raise ValueError( f"Unexpected length of key_point: {key_point}" ) timestamps = ( key_point[2].replace(",", ",").replace(" ", "").split(",") ) std_timestamps = "".join( [ f"[_TIMESTAMP_]('{timestamp}')" for timestamp in timestamps ] ) formated_breakdown += ( f"- **{key_point[0]}**: {key_point[1]}{std_timestamps}\n" ) insight = tmpl.format(overview, formated_breakdown) return insight, title except Exception as e: logger.error(f"Unexpected error: {e}") raise RuntimeError(f"Unexpected error: {e}") else: user_info = user_info.format( content=content, speech=speech, user_input=user_input ) prompt_audio_parser = self.lf_prompt_audio_parser.replace("__bio__", bio) messages = [ {"role": "system", "content": prompt_audio_parser}, { "role": "user", "content": [{"type": "text", "text": user_info + language_desc}], }, ] response = self.client.chat.completions.create( model=self.model_name, messages=messages, max_tokens=4096, temperature=0.0, max_retries=max_retries, timeout=request_timeout, response_format={"type": "json_object"}, ) api_res_dict = response.choices[0].message.content try: title = api_res_dict.get("Title", "") overview = api_res_dict.get("Overview", "") breakdown = api_res_dict.get("Breakdown", {}) tmpl = "{}\n{}" formated_breakdown = "" for subtitle, key_points in breakdown.items(): formated_breakdown += f"\n**{subtitle}**\n" for key_point in key_points: if len(key_point) != 3: raise ValueError( f"Unexpected length of key_point: {key_point}" ) timestamps = ( key_point[2].replace(",", ",").replace(" ", "").split(",") ) std_timestamps = "".join( [ f"[_TIMESTAMP_]('{timestamp}')" for timestamp in timestamps ] ) formated_breakdown += ( f"- **{key_point[0]}**: {key_point[1]}{std_timestamps}\n" ) insight = tmpl.format(overview, formated_breakdown) return insight, title except Exception as e: logger.error(f"Unexpected error: {e}") raise RuntimeError(f"Unexpected error: {e}") def _insighter_doc( self, bio: Dict[str, str], content: str, max_retries: int, request_timeout: int, file_content: Dict[str, Any], max_tokens: int = 3000, filter=equidistant_filter, ) -> tuple[str, str]: """Process document content to generate insights. Args: bio: Dictionary containing user biography information content: Text content or hint about the document max_retries: Maximum number of API call retries request_timeout: Timeout for API calls in seconds file_content: Dictionary containing document content max_tokens: Maximum tokens for generation filter: Function to filter document chunks Returns: Tuple of (insight, title) strings """ user_info = """# Hint # "{hint}" # Content # "{content}" # User Instruction # "{user_input}" """ user_input = "Here are some content and their hint. Please follow the WorkFlow and do your best. Ensure that your response is in a parseable JSON format. " language_desc = select_language_desc(self.preferred_language) segment_list = [self.lf_prompt_doc_overview, self.lf_prompt_doc_breakdown] messages_list = [] max_retry_list = [] alarm_mesg_list = [] for i in range(len(segment_list)): DOC_PARSER_PROMPT = segment_list[i] raw_text = DOC_PARSER_PROMPT + user_input + user_info + language_desc upper_bound = cal_upperbound( model_limit=7000 + max_tokens, generage_limit=max_tokens, tolerance=500, raw=raw_text, ) # Chunk and truncate chunk_size = 512 chunk_num = upper_bound // chunk_size + 1 if self.model_name is None: self.user_llm_config = self.user_llm_config_service.get_available_llm() self.client = OpenAI( api_key=self.user_llm_config.chat_api_key, base_url=self.user_llm_config.chat_endpoint, ) self.model_name = self.user_llm_config.chat_model_name spliter = TokenTextSplitter( chunk_size=chunk_size, chunk_overlap=0, model_name=self.model_name.replace("openai/", ""), ) tmp = file_content.get("content", "") doc_content = "\n".join(tmp) splits = spliter.split_text(doc_content) use_content = chunk_filter( splits, filter, filtered_chunks_n=chunk_num, separator="\n", spacer="\n" ) doc_content = get_safe_content_turncate( use_content, self.model_name.replace("openai/", ""), max_tokens=upper_bound ) user_content = user_info.format( hint=content, content=doc_content, user_input=user_input ) if "__global_bio__" in DOC_PARSER_PROMPT: DOC_PARSER_PROMPT = DOC_PARSER_PROMPT.replace( "__about_me__", bio["about_me"] ) DOC_PARSER_PROMPT = DOC_PARSER_PROMPT.replace( "__global_bio__", bio["global_bio"] ) DOC_PARSER_PROMPT = DOC_PARSER_PROMPT.replace( "__status_bio__", bio["status_bio"] ) messages = [ {"role": "system", "content": DOC_PARSER_PROMPT}, {"role": "user", "content": user_content + language_desc}, ] messages_list.append(messages) results = [] for messages in messages_list: response = self.client.chat.completions.create( model=self.model_name, messages=messages, max_tokens=max_tokens, temperature=0.0, timeout=request_timeout, response_format={"type": "json_object"}, ) results.append(json.loads(response.choices[0].message.content)) try: title = results[0].get("Title") overview = results[0].get("Overview") breakdown = results[1].get("Breakdown", {}) tmpl = "{}\n{}" formated_breakdown = "" for subtitle, key_points in breakdown.items(): formated_breakdown += f"\n**{subtitle}**\n" if not isinstance(key_points, list): raise RuntimeError( f"Unexpected generated result: {json.dumps(breakdown)}" ) for key_point in key_points: if isinstance(key_point, list) and len(key_point) == 2: formated_breakdown += f"- **{key_point[0]}**: {key_point[1]}\n" else: raise RuntimeError( f"Unexpected generated result in key_points: {json.dumps(breakdown)} expected a list of length 2." ) insight = tmpl.format(overview, formated_breakdown) return insight, title except Exception as e: logger.error(traceback.format_exc()) raise RuntimeError(f"Unexpected error: {e}") def insighter(self, inputs: InsighterInput) -> Dict[str, str]: """Generate insights from document inputs. Args: inputs: Structured input parameters containing file and bio information Returns: Dictionary containing title and insight """ try: datatype = DataType(inputs.file_info.data_type) except ValueError: logger.warning( "Unsupported dataType: %s. Processing as DOCUMENT by default", inputs.file_info.data_type, ) datatype = DataType.DOCUMENT logger.info("input filename=%s", inputs.file_info.filename) logger.info( "input content=%s (first 100 characters)", inputs.file_info.content.strip()[:100], ) bio = { "global_bio": inputs.bio_info.global_bio.split("### Conclusion ###")[ -1 ].strip("\n ") if inputs.bio_info.global_bio else "User has no biography right now", "status_bio": inputs.bio_info.status_bio.split( "** User Activities Overview **" )[-1] .strip("** Physical and mental health status **")[0] .strip("\n") if inputs.bio_info.status_bio else "", "about_me": inputs.bio_info.about_me.strip("\n") if inputs.bio_info.about_me else "", } text_len = len(self._tokenizer.encode(inputs.file_info.content)) if text_len > 20 or inputs.file_info.file_content: if datatype == DataType.IMAGE: insight, title = self._insighter_image( bio=bio, content=inputs.file_info.content, max_retries=self.max_retries_summarize, request_timeout=30, file_content=inputs.file_info.file_content, ) elif datatype == DataType.AUDIO: insight, title = self._insighter_audio( bio=bio, content=inputs.file_info.content, max_retries=self.max_retries_summarize, request_timeout=45, file_content=inputs.file_info.file_content, ) else: insight, title = self._insighter_doc( bio=bio, content=inputs.file_info.content, max_retries=self.max_retries_summarize, request_timeout=45, file_content=inputs.file_info.file_content, ) else: logger.warning("less than 20 characters, use filename as title") title, insight = inputs.file_info.content, inputs.file_info.content if inputs.file_info.filename: logger.info("use filename as title") title = inputs.file_info.filename t1 = time.time() logger.warning( "Insighter: title=%s, summary=%s", title, insight, ) return { "title": title, "insight": insight, } def __serial_summary_filter( self, summaries: List[str], chunks_list: List[List[str]], separator: str = "", filtered_chunks_n: int = 6 ) -> List[str]: """Filter and combine summaries with relevant chunks. Args: summaries: List of summary strings chunks_list: List of lists containing text chunks separator: String to join chunks and summaries filtered_chunks_n: Maximum number of chunks to filter Returns: List of combined content strings """ # Skip summary when chunks length is 0, otherwise combine summary with some adjacent chunks use_contents = [] for summary, chunks in zip(summaries, chunks_list): # When chunks exceed filtered_chunks_n-1, this is not the final summarization round if len(chunks) > filtered_chunks_n - 1: use_content = separator.join([summary, *chunks[:5]]) # When chunks are between 0 and filtered_chunks_n-1, this is the final round elif len(chunks) > 0: use_content = separator.join([summary, *chunks]) else: # When chunks are 0, summary is done, skip this round to avoid using resources continue use_contents.append(use_content) return use_contents def _summarize_title_abstract_keywords( self, content: str or List[str], filename: str, file_type: str, request_timeout: int, max_retries: int, preferred_language: str, filter=equidistant_filter, ) -> tuple[str, str, List[str]] or List[tuple[str, str, List[str]]]: """Generate title, abstract and keywords from content. Args: content: String or list of strings to summarize filename: Name of the file being summarized file_type: Type of file (document, image, audio, etc.) request_timeout: Timeout for API calls in seconds max_retries: Maximum number of API call retries preferred_language: Language to use for generation filter: Function to filter content chunks Returns: Single tuple or list of tuples containing (title, summary, keywords) """ upper_limit = 8192 filtered_chunks_n = 14 max_tokens = 512 if isinstance(content, str): inputs = [content] else: inputs = content filename = filename or "" if not filename: filename_desc = "" else: filename_desc = f"Filename: {filename}\n" def get_text_generate(_requests): language_desc = "" prompt = NOTE_SUMMARY_PROMPT.replace("{language_desc}", language_desc) messages = [ [ {"role": "user", "content": prompt.format(**_request)}, { "role": "system", "content": f"""User Preferred Language: {preferred_language}, you should use this language to generate the title, summary. Don't to start the summary section with sentences like "This document", "This text" or "This article", but describe the content directly.""", }, ] for _request in _requests ] logger.info("generate inputs: %s", _requests) responses = [ self.client.chat.completions.create( model=self.model_name, messages=msg, max_tokens=max_tokens, temperature=0.0, timeout=request_timeout, ) for msg in messages ] return responses spliter = TokenParagraphSplitter(chunk_size=512, chunk_overlap=0) if filter is self.__serial_summary_filter: # Serial fine-grained full-text summary chunks_list = [spliter.split_text(each) for each in inputs] # Maximum number of summaries needed [K summaries can handle docs with 5K+1 chunks] max_summary_times = int( (max([len(chunks) for chunks in chunks_list]) + 4) / 5 ) results = [() for i in range(len(inputs))] # Initialize summaries with first chunk content # Set to empty string if chunks length is 0 summaries = [chunks[0] if len(chunks) > 0 else "" for chunks in chunks_list] # When chunks length is 1, set to [""], requires one summary # When chunks length is 0, set to empty list, no summary needed chunks_list = [ [] if len(chunks) == 0 else ([""] if len(chunks) == 1 else chunks[1:]) for chunks in chunks_list ] for i in range(max_summary_times): use_contents = self.__serial_summary_filter(summaries, chunks_list) requests = [ { "content": use_content, "file_type": file_type, "filename_desc": filename_desc, } for use_content in use_contents ] responses = get_text_generate(requests) tmp_results = get_summarize_title_keywords(responses) for doc_id, chunks in enumerate(chunks_list): index = 0 # Documents participating in this round of summaries if len(chunks) > 0: # Update result (title, abstract, keywords) results[doc_id] = tmp_results[index] # Update summary list summaries[doc_id] = tmp_results[index][1] # Update chunks list to be summarized chunks_list[doc_id] = chunks_list[doc_id][5:] index += 1 else: requests = [] for each in inputs: splits = spliter.split_text(each) # Sampling-based full text summary approach # Keep beginning and end, can skip middle. End is useful for company signatures and information, reducing model hallucination # Also keep one extra chunk at the end to avoid issues with short final chunks providing insufficient information use_content = chunk_filter( splits, filter, filtered_chunks_n=filtered_chunks_n, separator="\n", spacer="\n……\n……\n……\n", ) if self.model_name is None: self.user_llm_config = self.user_llm_config_service.get_available_llm() self.client = OpenAI( api_key=self.user_llm_config.chat_api_key, base_url=self.user_llm_config.chat_endpoint, ) self.model_name = self.user_llm_config.chat_model_name requests.append( { "content": get_safe_content_turncate( use_content, self.model_name.replace("openai/", ""), max_tokens=upper_limit, ), "file_type": file_type, "filename_desc": filename_desc, } ) responses = get_text_generate(requests) results = get_summarize_title_keywords(responses) logger.debug("results: %s", results) if isinstance(content, str): return results[0] else: return results def summarizer(self, inputs: SummarizerInput) -> Dict[str, Any]: """Generate summary from document inputs. Args: inputs: Structured input parameters containing file information and insight Returns: Dictionary containing title, summary and keywords """ bottom_summary_len = 200 datatype = inputs.file_info.data_type filename = inputs.file_info.filename md = inputs.file_info.content # hint inner_content = inputs.file_info.file_content.get("content") insight = inputs.insight md = md + "\n" + inner_content md = f"insight: {insight}\ncontent: {md}" try: datatype = DataType(datatype) except ValueError: logger.warning("Unsupported dataType: %s. Processing as DOCUMENT by default", datatype) datatype = DataType.DOCUMENT logger.info("input filename=%s", filename) logger.info("input content=%s (first 100 characters)", md.strip()[:100]) t0 = time.time() bottom_summary = self._tokenizer.decode( self._tokenizer.encode(insight)[:bottom_summary_len] ) if len(self._tokenizer.encode(md)) > 20: title, summary, keywords = self._summarize_title_abstract_keywords( md, filename=filename, file_type=datatype.value, request_timeout=self.timeout_summarize, max_retries=self.max_retries_summarize, preferred_language=self.preferred_language, ) if not (title or summary or keywords): logger.warning("summary failed, use insight as summary") title, summary, keywords = filename, bottom_summary, [] if filename: title = filename else: logger.warning("less than 20 characters, use filename as title") title, summary, keywords = md, md, [] if filename: title = filename t1 = time.time() logger.warning( "MarkdownChunkAPI summarize_title_abstract_keywords(): time spent %.2f seconds, title=%s, summary=%s", t1 - t0, title, summary, ) return {"title": title, "summary": summary, "keywords": keywords}