import  osimport  numpy  as  npimport  jsonimport  fitzimport  dashscopefrom  openai  import  OpenAI
os. environ [ 'DASHSCOPE_API_KEY' ] =  "your dashscope api key"

def  extract_text_from_pdf ( pdf_path ):    # Open PDF file    document = fitz. open (pdf_path)    all_text =  ""   # Initialize an empty string to store the extracted text
    # Traverse each page in the PDF    for  page_num  in  range (document.page_count):        page = document[page_num]   # Get page        text = page.get_text( "text" )   # Extract text from the page        all_text += text   # Append the extracted text to the all_text string
    return  all_text   # Returns the extracted text

def  chunk_text ( text_input, chunk_size, overlap_size ):    text_chunks = []   # Initialize a list to store text blocks
    # Loop through the text with a step size of (chunk_size - overlap_size)    for  i  in  range ( 0 ,  len (text_input), chunk_size - overlap_size):        text_chunks.append(text_input[i:i + chunk_size])   # Append text blocks from i to i+chunk_size to the text_chunks list
    return  text_chunks   # Returns a list of text blocks

client = OpenAI( api_key=os.getenv("DASHSCOPE_API_KEY"), # If you don't have environment variables configured, please replace it with your API Key here base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" # base_url of the Bailian service)

# Define the PDF file pathpdf_path =  "knowledge_base/Intelligent Coding Assistant Tongyi Lingma.pdf"
# Extract text from PDF filesextracted_text = extract_text_from_pdf(pdf_path)
# Split the extracted text into blocks of 1000 characters, with overlaps of 100 characterstext_chunks = chunk_text(extracted_text,  1000 ,  100 )
# Print the number of text blocks createdprint ( "Number of text chunks: " ,  len (text_chunks))
# Print the first block of textprint ( "\nFirst text block: " )print ( text_chunks[ 0 ] )

Number of text blocks: 5
First block of text:What is the intelligent coding assistant Tongyi Lingma Intelligent Coding Assistant Tongyi Lingma (abbreviated as Tongyi Lingma) is an intelligent coding assistance tool provided by Alibaba Cloud.It provides capabilities such as intelligent code generation, intelligent question and answer, multi-file modification, and programming intelligence, bringing highefficient and smooth coding experience, leading a new paradigm of AI native R&D. At the same time, we provide enterprise customers withStandard and exclusive versions have the capabilities of enterprise-level scenario customization and private domain knowledge enhancement, helping enterprises develop intelligentChemical upgrade. Core Competencies Code Completion After training with a large amount of excellent open source code data, it can generate for you according to the current code file and the context across files.Line-level/function-level code, unit tests, code optimization suggestions, etc. Immersive coding flow, generation speed in seconds,You focus more on technical design and complete coding work efficiently. Ask Mode The intelligent question-answering model has a large amount of R&D documents, product documents, general R&D knowledge, etc., and combines engineering-level perception capabilitiesWe can help developers solve R&D problems encountered during the coding process and assist them in fixing and debugging code problems.Or run error troubleshooting, etc. Edit Mode The file editing mode has the ability to modify multiple file codes. When developers need to modify code files accurately,It can combine the requirements description with the current engineering environment to modify multiple files, and can perform multiple iterations and code reviews.Check to help developers complete code modification tasks efficiently and controllably. Agent Mode The intelligent agent mode has the ability to make autonomous decisions, perceive the environment, use tools, etc., and can be coded by the developer.Request, use engineering search, file editing, terminal and other tools, you can complete the coding task end to end.Developers configure MCP tools to make coding more in line with the developer workflow. Product Advantages • Multiple conversation modes: support question-and-answer mode, file editing mode, and intelligent model simultaneously in one conversation flowDevelopers can freely switch modes according to different scenarios and problem difficulties to achieve maximum work efficiencychange. • Automatic project perception: Based on the developer's task description, it can automatically perceive the project framework, technology stack, and requiredCode files, error messages and other project information, no need to manually add project context, task description is lighterThe code completion is more in line with the business scenarios of the current code base. • Project-level changes: Based on the developer's task description, you can independently perform task decomposition and multiple codes within the project.File modification can be done through multiple conversations, and can be done through gradual iteration or snapshot rollback, in collaboration with Tongyi Lingma.A coding task. • Memory perception: Supports autonomous memory capabilities based on large models.The Tongyi Lingma will gradually form rich memories related to developers, projects, problems, etc.The more you use it, the better we understand you. • Multiple enterprise edition plans, flexible choice: Provide enterprise standard edition, enterprise

# Create embedding vectors for text blocksdef  create_embeddings ( texts, model= "text-embedding-v3" ):    """    Input a set of text (string or list) and return the corresponding embedding vector list    """    ifisinstance(texts,  str ):        texts = [texts]   # ​​Ensure that the input is in list form
    completion = client.embeddings.create(        model=model,        input =texts,        encoding_format= "float"    )        # Convert the response to a dict and extract all embeddings    data = json.loads(completion.model_dump_json())    embeddings = [item[ "embedding" ]  for  item  in  data[ "data" ]]    return  embeddings

from  sklearn.metrics.pairwise  import  cosine_similarity
# Semantic search functiondef  semantic_search ( query, text_chunks, embeddings= None , k= 2 ):    """    Find the top-k text chunks in text_chunks that are most relevant to the query        parameter:        query: query statement        text_chunks: list of candidate text chunks        embeddings: list of corresponding embedding vectors (if pre-computed)        k: the number of most relevant results to return            return:        top_k_chunks: the most relevant top-k text chunks    """    if  embeddings  is  None :        embeddings = create_embeddings(text_chunks)   # Automatically generated if not provided    else :        assert  len ​​(embeddings) ==  len (text_chunks),  "embeddings and text_chunks must be the same length"
    query_embedding = create_embeddings(query)[ 0 ]   # Get the embedding of the query
    # Calculate similarity    similarity_scores = []    for  i, chunk_embedding  in  enumerate (embeddings):        score = cosine_similarity([query_embedding], [chunk_embedding])[ 0 ][ 0 ]        similarity_scores.append((i, score))
    # Sort and take top-k    similarity_scores.sort(key= lambda  x: x[ 1 ], reverse= True )    top_indices = [index  for  index, _  in  similarity_scores[:k]]    return  [text_chunks[index]  for  index  in  top_indices]

# Perform a semantic searchquery =  'What are the capabilities of Tongyi Lingma's intelligent entities?'top_chunks = semantic_search(query, text_chunks, k= 2 )
# Output resultsprint ( "Query: " , query)for  i, chunk  in  enumerate (top_chunks):    print ( f"Context  {i +  1 } :\n {chunk} \n=====================================" )

Query: What are the intelligent capabilities of Tongyi Lingma? Context 1: What is the intelligent coding assistant Tongyi Lingma? The intelligent coding assistant Tongyi Lingma (abbreviated as Tongyi Lingma) is an intelligent coding assistance tool provided by Alibaba Cloud. It provides capabilities such as intelligent code generation, intelligent question and answer, multi-file modification, and programming intelligent agents, bringing developers an efficient and smooth coding experience and leading a new paradigm of AI native research and development. At the same time, we provide enterprise customers with enterprise standard and exclusive versions, which have enterprise-level scene customization, private domain knowledge enhancement and other capabilities to help enterprises upgrade their research and development intelligence. Core Capabilities Code Completion Code Completion has been trained with a large amount of excellent open source code data, and can generate line-level/function-level code, unit tests, code optimization suggestions, etc. for you based on the current code file and cross-file context. Immersive coding flow and generation speed in seconds allow you to focus more on technical design and complete coding work efficiently. Intelligent Question and Answer Ask Mode Intelligent question and answer mode has a large amount of R&D documents, product documents, general R&D knowledge, etc., and combines engineering-level perception capabilities to help developers solve R&D problems encountered during the coding process, and assist developers in fixing code problems, debugging, or troubleshooting running errors. File Editing Edit Mode File editing mode has the ability to modify multiple files. When developers need to modify code files accurately, they can modify multiple files in combination with the requirement description and the current engineering environment, and can perform multiple iterations and code reviews to help developers complete code modification tasks efficiently and controllably. Agent Mode Agent mode has the capabilities of autonomous decision-making, environmental perception, and tool use. It can use tools such as engineering retrieval, file editing, and terminals according to the developer's coding requirements to complete coding tasks end-to-end. At the same time, it supports developers to configure MCP tools, so that coding is more in line with the developer's workflow. Product Advantages • Multiple session modes: Question and answer mode, file editing mode, and agent mode are supported in one session flow. Developers can freely switch modes according to different scenarios and problem difficulties to maximize work efficiency. • Automatic project perception: Based on the developer's task description, it can automatically perceive project information such as project framework, technology stack, required code files, error messages, etc., without manually adding project context, making task description easier and code completion more in line with the business scenario of the current code base. • Project-level changes: Based on the developer's task description, it can independently decompose tasks and modify multiple code files in the project. At the same time, it can gradually iterate or roll back snapshots through multiple conversations, and collaborate with Tongyi Lingma to complete coding tasks. • Memory perception: Supports autonomous memory capabilities based on large models. During the conversation between developers and Tongyi Lingma, Tongyi Lingma will gradually form rich memories related to developers, projects, problems, etc., and the more you use it, the better it understands you. • Multiple Enterprise Edition solutions, flexible choice: Provide Enterprise Standard Edition, Enterprise =========================================Context 2: Perception: Support autonomous memory capabilities based on large models. In the process of dialogue between developers and Tongyi Lingma, Tongyi Lingma will gradually form rich memories related to developers, projects, problems, etc., and the more you use it, the better it understands you. • Multiple Enterprise Edition solutions, flexible choice: Provide a variety of solutions for enterprise customers such as Enterprise Standard Edition and Enterprise Exclusive Edition, and provide enterprise personalized solutions, which can be flexibly selected to accelerate the large-scale implementation of intelligent R&D within the enterprise. Function introduction Interline code completion • Line-level/function-level real-time continuation: According to the current syntax and cross-file code context, it automatically perceives the current project and generates line and function-level code in real time; • Generate code by commenting: Describe the functions you want through comments, and generate code directly in the editor area, so that the coding flow is uninterrupted. Intelligent Q&A • R&D Q&A: When you encounter coding questions or technical problems, you can call up Tongyi Lingma with one click, and you can quickly get answers and solutions without leaving the IDE client. • Engineering Q&A: Through Q&A, you can quickly combine the current warehouse to understand the project, query the code, etc. At the same time, you can describe the requirements through natural language, and generate overall repair suggestions and recommended codes for simple requirements or defects in combination with the current project. • Image multimodal Q&A: Supports selecting, dragging or pasting to add images as context, automatically analyzes the image content, and generates code suggestions or problem repair suggestions based on the requirement description. • Enterprise knowledge base Q&A: Use enterprise knowledge and data to conduct Q&A, quickly build an enterprise R&D knowledge Q&A assistant, and improve the work efficiency and collaboration ability of the team. File editing • Engineering-level changes: According to the developer's task description, you can modify multiple code files in the project, and you can perform gradual iterations or snapshot rollbacks through multiple conversations. Developers and Tongyi Lingma work together to gradually complete coding tasks. • Precise editing: Complete code file modifications within the context provided by the developer, and no modifications beyond the developer's expectations will be made. • Fast execution: Strictly follow the developer's task description and the context provided to modify code files. There is no need for additional complex task planning, which completes tasks more quickly than the agent mode. • Tool usage: It has the ability to use code modification related tools such as file reading, semantic retrieval within the project, and file editing, which can help developers quickly complete code modifications. Programming agent • Project-level changes: According to the developer's task description, it can independently decompose tasks and modify multiple code files within the project. At the same time, it can perform step-by-step iteration or snapshot rollback through multiple conversations, and collaborate with Tongyi Lingma to complete coding tasks. • Automatic project perception: According to the developer's task description, it can automatically perceive project information such as project framework, technology stack, required code files, error messages, etc., without manually adding project context, making task description easier. • Tool usage: You can use more than a dozen built-in programming tools independently, such as reading and writing files, generation === ...

# Initialize the DashScope client (using Alibaba Cloud Tongyi Qianwen)client = OpenAI(    api_key=os.getenv( "DASHSCOPE_API_KEY" ),   # Make sure to set the environment variables in advance    base_url= "https://dashscope.aliyuncs.com/compatible-mode/v1")
# Set system promptSYSTEM_PROMPT = (    "You are an AI assistant and must answer strictly based on the context provided."    "If the answer is not directly inferred from the context provided, respond with: 'I can't answer this question based on the information available.'")
def  generate_response ( system_prompt, user_message, model= "qwen-max" ):    """    Generate context-based answers using DashScope's 1,000-question model.
    parameter:        system_prompt (str): System prompts that control AI behavior        user_message (str): question and context entered by the user        model (str): the model name used, the default is qwen-plus
    return:        str: the answer content generated by the model    """    response = client.chat.completions.create(        model=model,        temperature= 0.0 ,   # Set the temperature to 0 to ensure output certainty        max_tokens= 512 ,    # The maximum output length can be adjusted as needed        messages=[            { "role" :  "system" ,  "content" : system_prompt},            { "role" :  "user" ,  "content" : user_message}        ]    )        return  response.choices[ 0 ].message.content.strip()
# Example top_chunks (assuming this is what semantic_search returns)top_chunks = [    "Tongyi Lingma is an AI-based intelligent programming assistant." ,    "File editing capabilities include auto-completion, error fixing, and code refactoring."]
query =  "What are the capabilities of Tongyi Lingma's intelligent entities?"
# Build user prompt (contains context + question)user_prompt =  "\n" .join([ f"context  {i +  1 } :\n {chunk} " for  i, chunk  in  enumerate (top_chunks)])user_prompt +=  f"\n\nQuestion: {query} "
# Generate AI answeranswer = generate_response(SYSTEM_PROMPT, user_prompt)
# Output resultsprint ( "AI answers: " )print (answer)

 AI answers:The intelligent capabilities of Tongyi Lingma include the following aspects:
-  **Autonomous decision-making** : Ability to independently decompose tasks based on the developer's coding needs.-  **Environmental awareness** : It can automatically perceive project information such as project framework, technology stack, required code files, error messages, etc., without manually adding project context.-  **Tool Usage** : Ability to use more than ten built-in programming tools independently, such as reading and writing files, code editing, etc.-  **Complete coding tasks end-to-end** : Based on the needs of developers, use a variety of tools such as project retrieval, file editing, terminal, etc. to complete complete coding tasks from start to finish.-  **Support configuration of MCP tools** : Make the coding process more in line with the developer's personal workflow.

# Define prompt words for the evaluation systemevaluate_system_prompt = (    "You are an intelligent evaluation system responsible for evaluating the quality of the AI ​​assistant's answers."    "If the AI ​​assistant's answer is very close to the real answer, please give 1 point;"    "If the answer is wrong or irrelevant to the true answer, please give 0 points;"    "If the answer is partially true, please give 0.5 points."    "Please output the rating result directly: 0, 0.5 or 1.")#Build an evaluation prompt and get the scoreevaluation_prompt = f """User question: {query}AI Answer:{answer}Real answer: {ideal_answer}Please rate according to the following criteria:- If the AI ​​answer is very close to the real answer → Output 1- If the answer is wrong or irrelevant → output 0- If partial match → output 0.5" ""evaluation_result = generate_response(evaluate_system_prompt, evaluation_prompt)
# Step 5: Output the final scoreprint( "AI answer score: " , evaluation_result)

AI answer rating: 1

# Initialize the clientclient = OpenAI(    api_key=os.getenv( "DASHSCOPE_API_KEY" ),   # If you don't have environment variables configured, replace it with your API Key here    base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1"   # The base_url of the Bailian service)
# Create embedding vectors for text blocksdef  create_embeddings ( texts, model= "text-embedding-v3" ):    """    Input a set of text (string or list) and return the corresponding embedding vector list    """    if  isinstance ( texts ,  str ):        texts = [texts]   # ​​Ensure that the input is in list form
    completion = client.embeddings.create(        model=model,        input =text_chunks,        encoding_format= "float"    )        # Convert the response to a dict and extract all embeddings    data = json.loads(completion.model_dump_json())    embeddings = [item[ "embedding" ]  for  item  in  data[ "data" ]]    return  embeddings

# Preliminarily split the text into sentences based on periodssentences = extracted_text.split( "." )# Remove empty strings and leading and trailing spacessentences = [sentence.strip()  for  sentence  in  sentences  if  sentence.strip()]
# Generate embedding vectors for all sentences in batches (recommended)embeddings = create_embeddings(sentences)
print ( f"Successfully generated  embedding vectors for { len (embeddings)}  sentences." )

import  numpy  as  npfrom  sklearn.metrics.pairwise  import  cosine_similarity
def  cosine_similarity ( vec1, vec2 ):    """    Computes the cosine similarity between two vectors.
    parameter:    vec1(np.ndarray): The first vector.    vec2 (np.ndarray): The second vector.
    return:    float: cosine similarity.        abnormal:    ValueError: If the input vector is not a 1D array or the shapes do not match.    """    if  vec1.ndim !=  1 or  vec2.ndim !=  1 :        raise  ValueError( "Input vector must be a one-dimensional array" )    if  vec1.shape[ 0 ] != vec2.shape[ 0 ]:        raise  ValueError( "Input vectors must have the same dimensions" )
    return  np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
# Use sklearn's cosine_similarity function to calculate the similarity between consecutive sentencessimilarities = [cosine_similarity(embeddings[i].reshape( 1 , - 1 ), embeddings[i +  1 ].reshape( 1 , - 1 ))[ 0 ][ 0 ]  for  i  in  range ( len (embeddings) -  1 )]

def  compute_breakpoints ( similarity_scores, method= "percentile" , threshold= 90 ):    """    Calculate segment breakpoints based on similarity drop points.
    parameter:    similarity_scores(List[float]): List of similarities between sentences.    method(str): Threshold calculation method, optional 'percentile', 'standard_deviation', or 'interquartile'.    threshold(float): Threshold value (for percentile or standard deviation method).
    return:    List[int]: The index positions where the split should be done.    """    # Determine the threshold value based on the selected method    if  method ==  "percentile" :        # Calculate the similarity value of the specified percentile as the threshold        threshold_value = np.percentile(similarity_scores, threshold)    elif  method ==  "standard_deviation" :        # Calculate mean and standard deviation, and determine threshold by subtracting X standard deviations        mean = np.mean(similarity_scores)        std_dev = np.std(similarity_scores)        threshold_value = mean - (threshold * std_dev)    elif  method ==  "interquartile" :        # Use the interquartile range (IQR) rule to determine the outlier threshold        q1, q3 = np.percentile(similarity_scores, [ 25 ,  75 ])        iqr = q3 - q1        threshold_value = q1 -  1.5  * iqr    else :        # Throw an error if the method is invalid        raise  ValueError( "Invalid method. Choose 'percentile', 'standard_deviation', or 'interquartile'." )
    # Find the position where the similarity is lower than the threshold, that is, the segmentation breakpoint    return  [i  for  i, score  in  enumerate (similarity_scores)  if  score < threshold_value]
# Use the percentile method and set the threshold to 90% percentile to calculate the breakpointbreakpoints = compute_breakpoints(similarity_scores=similarities, method= "percentile" , threshold= 90 )

def  split_into_chunks ( sentence_list, break_indices ):    """    Divide the sentence list into semantic paragraphs based on the breakpoint index.
    parameter:    sentence_list(List[str]): sentence list.    break_indices(List[int]): A list of position indices that need to be segmented.
    return:    List[str]: The list of divided semantic paragraphs.    """    semantic_chunks = []   # Store the divided paragraphs    current_start_index =  0   # Current segment start index
    # Iterate through each breakpoint to create paragraphs    for  bp  in  break_indices:        # Connect the current segment from the starting index to the breakpoint position and add a period to end        semantic_chunks.append( ". " .join(sentence_list[current_start_index:bp +  1 ]) +  "." )        current_start_index = bp +  1   # Update the starting index to the next sentence
    # Add the last paragraph (remaining sentences)    semantic_chunks.append( ". " .join(sentence_list[current_start_index:]))
    return  semantic_chunks   # Returns a list of semantic paragraphs
# Use the split_into_chunks function to generate paragraphstext_chunks = split_into_chunks(sentence_list=sentences, break_indices=breakpoints)
# Print the number of generated paragraphsprint ( f"Number of semantic paragraphs generated:  { len (text_chunks)} " )
# Print the first paragraph to verify the resultprint ( "\nFirst semantic paragraph: " )print ( text_chunks[ 0 ] )

Number of semantic paragraphs generated: 4
The first semantic paragraph:What is the intelligent coding assistant Tongyi Lingma Intelligent Coding Assistant Tongyi Lingma (abbreviated as Tongyi Lingma) is an intelligent coding assistance tool provided by Alibaba Cloud.It provides capabilities such as intelligent code generation, intelligent question and answer, multi-file modification, and programming intelligence, bringing highEfficient and smooth coding experience, leading a new paradigm of AI native R&D. At the same time, we provide enterprise customers withStandard and exclusive versions have the capabilities of enterprise-level scenario customization and private domain knowledge enhancement, helping enterprises develop intelligentChemical upgrade.

# Create embedding vectors for text blocksdef  create_embeddings ( texts, model= "text-embedding-v3" ):    """    Input a set of text (string or list) and return the corresponding embedding vector list    """    ifisinstance(texts,  str ):        texts = [texts]   # ​​Ensure that the input is in list form
    completion = client.embeddings.create(        model=model,        input =text_chunks,        encoding_format= "float"    )        # Convert the response to a dict and extract all embeddings    data = json.loads(completion.model_dump_json())    embeddings = [item[ "embedding" ]  for  item  in  data[ "data" ]]    return  embeddings

from  sklearn.metrics.pairwise  import  cosine_similarity
# Semantic search functiondef  semantic_search ( query, text_chunks, embeddings= None , k= 2 ):    """    Find the top-k text chunks in text_chunks that are most relevant to the query        parameter:        query: query statement        text_chunks: list of candidate text chunks        embeddings: list of corresponding embedding vectors (if pre-computed)        k: the number of most relevant results to return            return:        top_k_chunks: the most relevant top-k text chunks    """    if  embeddings  is  None :        embeddings = create_embeddings(text_chunks)   # Automatically generated if not provided    else :        assert  len ​​(embeddings) ==  len (text_chunks),  "embeddings and text_chunks must be the same length"
    query_embedding = create_embeddings(query)[ 0 ]   # Get the embedding of the query
    # Calculate similarity    similarity_scores = []    for  i, chunk_embedding  in  enumerate (embeddings):        score = cosine_similarity([query_embedding], [chunk_embedding])[ 0 ][ 0 ]        similarity_scores.append((i, score))
    # Sort and take top-k    similarity_scores.sort(key= lambda  x: x[ 1 ], reverse= True )    top_indices = [index  for  index, _  in  similarity_scores[:k]]    return  [text_chunks[index]  for  index  in  top_indices]

# Perform a semantic searchquery =  'What is the intelligent coding assistant Tongyi Lingma? 'top_chunks = semantic_search(query, text_chunks, k= 2 )
# Output resultsprint ( "Query: " , query)for  i, chunk  in  enumerate (top_chunks):    print ( f"Context  {i +  1 } :\n {chunk} \n=====================================" )

Query: What is the intelligent coding assistant Tongyi Lingma Context 1: What is the intelligent coding assistant Tongyi Lingma Intelligent coding assistant Tongyi Lingma (abbreviated as Tongyi Lingma) is an intelligent coding assistance tool provided by Alibaba Cloud, providing capabilities such as intelligent code generation, intelligent question and answer, multi-file modification, and programming intelligence, bringing developers an efficient and smooth coding experience and leading a new paradigm of AI native research and development. At the same time, we provide enterprise customers with enterprise standard version and exclusive version, which have the capabilities of enterprise-level scene customization and private domain knowledge enhancement to help enterprises upgrade their research and development intelligence.======================================Context 2: Core Capabilities Code Completion Code Completion After training with a large amount of excellent open source code data, it can generate line-level/function-level code, unit tests, code optimization suggestions, etc. for you according to the context of the current code file and across files.=========================================

import  osfrom  openai  import  OpenAI
# Initialize the DashScope client (using Alibaba Cloud Tongyi Qianwen)client = OpenAI(    api_key=os.getenv( "DASHSCOPE_API_KEY" ),   # Make sure to set the environment variables in advance    base_url= "https://dashscope.aliyuncs.com/compatible-mode/v1")
# Set system prompt (Chinese version)SYSTEM_PROMPT = (    "You are an AI assistant and must answer strictly based on the context provided."    "If the answer is not directly inferred from the context provided, respond with: 'I can't answer this question based on the information available.'")
def  generate_response ( system_prompt, user_message, model= "qwen-max" ):    """    Generate context-based answers using DashScope's 1,000-question model.
    parameter:        system_prompt (str): System prompts that control AI behavior        user_message (str): question and context entered by the user        model (str): the model name used, the default is qwen-plus
    return:        str: the answer content generated by the model    """    response = client.chat.completions.create(        model=model,        temperature= 0.0 ,   # Set the temperature to 0 to ensure output certainty        max_tokens= 512 ,    # The maximum output length can be adjusted as needed        messages=[            { "role" :  "system" ,  "content" : system_prompt},            { "role" :  "user" ,  "content" : user_message}        ]    )        return  response.choices[ 0 ].message.content.strip()

def  context_enriched_search ( search_query, chunked_texts, chunk_embeddings, top_k= 1 , context_window_size= 1 ):    """    When searching, not only the most relevant paragraph is returned, but also the context paragraphs before and after it to provide richer background information.
    parameter:        search_query(str): user's query statement.        chunked_texts(List[str]): List of chunked texts.        chunk_embeddings(List[dict]): vector representation of each text paragraph (usually obtained from an embedding model).        top_k(int): The number of relevant paragraphs to be retrieved (here only top 1 is used to find the central paragraph).        context_window_size (int): The number of context paragraphs to include (the number of front and back paragraphs).
    return:        List[str]: List of text paragraphs containing the most relevant paragraphs and their context.    """
    # Step 1: Convert the user's question into a vector (embedding) for similarity comparison with the text paragraph    query_embedding = create_embeddings(search_query).data[ 0 ].embedding
    similarity_list = []   # Used to store the similarity score and index of each paragraph and question
    # Step 2: Traverse all paragraph vectors and calculate the cosine similarity between them and the question vector    for  i, chunk_embedding  in  enumerate (chunk_embeddings):        # Use the cosine_similarity function to calculate similarity (the closer to 1, the more similar)        similarity_score = cosine_similarity(np.array(query_embedding), np.array(chunk_embedding.embedding))        # Save the index and similarity of the paragraph, such as: (0, 0.75)        similarity_list.append((i, similarity_score))
    # Step 3: Sort all paragraphs by similarity, from high to low    similarity_list.sort(key= lambda  x: x[ 1 ], reverse= True )
    # Step 4: Get the index of the most relevant paragraph (i.e. the first paragraph)    most_relevant_index = similarity_list[ 0 ][ 0 ]   # For example: the third paragraph
    # Step 5: Determine the context range to be extracted (including the current paragraph + context_window_size paragraphs before and after)    start_index =  max ( 0 , most_relevant_index - context_window_size)   # Prevent exceeding the beginning    end_index =  min ( len (chunked_texts), most_relevant_index + context_window_size +  1 )   # Prevent exceeding the end
    # Step 6: Return a list of paragraphs containing context    return  [chunked_texts[i]  for  i  in  range (start_index, end_index)]

# Chapter 3: Basic technologies of artificial intelligence The core methods of artificial intelligence include machine learning, deep learning and natural language processing...

def  generate_chunk_header ( text_chunk, model_name= "qwen-max" ):    """    Generate a title/summary for a given text paragraph using a Large Language Model (LLM).
    parameter:        text_chunk(str): The text paragraph for which the title needs to be generated.        model_name (str): The name of the language model used to generate titles. The default value is "qwen-max".
    return:        str: Title or summary content generated by the model.    """
    # Define system prompt words to guide AI's behavior    header_system_prompt =  "Please generate a concise and informative title for the following text."
    # Call the LLM model to generate responses based on system prompts and input text    llm_response = client.chat.completions.create(        model=model_name,        temperature = 0 ,        messages=[            { "role" :  "system" ,  "content" : header_system_prompt},            { "role" :  "user" ,  "content" : text_chunk}        ]    )
    # Extract and return the content generated by the model, removing the extra whitespace characters before and after    return  llm_response.choices[ 0 ].message.content.strip()

def  chunk_text_with_headers ( input_text, chunk_size, overlap_size ):    """    Split the input text into smaller paragraphs and generate headings for each paragraph.
    parameter:        input_text(str): The complete text to be segmented.        chunk_size(int): The size of each paragraph (number of characters).        overlap_size(int): The number of overlapping characters between adjacent paragraphs.
    return:        List[dict]: A list of dictionaries containing the keys 'header' and 'text', representing the paragraph's title and content respectively.    """        text_chunks = []   # Initialize an empty list to store text paragraphs with titles
    # Iterate over the text using the specified paragraph size and overlap length    for  start_index  in  range ( 0 ,  len (input_text), chunk_size - overlap_size):        current_chunk = input_text[start_index:start_index + chunk_size]   # Extract the current paragraph        chunk_header = generate_chunk_header(current_chunk)   # Generate paragraph title using large language model        text_chunks.append({ "header" : chunk_header,  "text" : current_chunk})   # Add the title and paragraph content to the list together
    return  text_chunks   # Returns a list of paragraphs containing titles and content

# Generate embedding vectors for each text blockchunk_embeddings = []   # Initialize an empty list to store dictionaries with titles, texts, and their embedding vectors
# Iterate over each text block and generate embedding vectors (with progress bar)for current_chunk in tqdm(text_chunks, desc= "Generate embedding vector" ):    # Get the text content of the current text block and generate its embedding vector    text_embedding = create_embeddings(current_chunk[ "text" ])        # Get the title of the current text block and generate its embedding vector    header_embedding = create_embeddings(current_chunk[ "header" ])        # Store the title, text and corresponding embedding vector of the current text block into a list    chunk_embeddings.append({        "header" : current_chunk[ "header" ],        "text" : current_chunk[ "text" ],        "embedding" : text_embedding,        "header_embedding" : header_embedding    })

import  numpy  as  np
def  _calculate_similarity ( query_vec, chunk_vec ):    """    Compute the cosine similarity between the query vector and the chunk vector.        parameter:    query_vec (np.array): The embedding vector of the query.    chunk_vec (np.array): The embedding vector of the text chunk.        return:    float: cosine similarity.    """    return  cosine_similarity(np.array(query_vec), np.array(chunk_vec))
def  semantic_search ( query, chunks, top_k= 5 ):    """    Search for the most relevant text blocks based on the query semantics.        parameter:    query (str): The query statement entered by the user.    chunks (List[dict]): List of text chunks containing embedding vectors.    top_k (int): The number of most relevant results to return.        return:    List[dict]: the top_k most relevant text blocks.    """    # Generate a vector representation of the query statement    query_vector = create_embeddings(query)
    # Initialize a list to store each text block and its similarity    chunk_similarity_pairs = []
    # Traverse each text block and calculate the similarity    for  chunk  in  chunks:        text_vector = chunk[ "embedding" ]      # Get the embedding vector of the text content        header_vector = chunk[ "header_embedding" ]   # Get the embedding vector of the title
        # Calculate the similarity between the query and the text and title respectively, and take the average        similarity_text = _calculate_similarity(query_vector, text_vector)        similarity_header = _calculate_similarity(query_vector, header_vector)        avg_similarity = (similarity_text + similarity_header) /  2
        #Store text blocks and their average similarity        chunk_similarity_pairs.append((chunk, avg_similarity))
    # Sort by similarity from high to low    chunk_similarity_pairs.sort(key= lambda  pair: pair[ 1 ], reverse= True )
    # Return the top-k most relevant text blocks    return  [pair[ 0 ] forpair  in  chunk_similarity_pairs[:top_k]]

import  re
def  _extract_questions_from_response ( response_text ):    """    Extract questions ending with a question mark from the text returned by the model.
    parameter:    response_text (str): The raw text returned by the model.
    return:    List[str]: List of valid questions after cleaning.    """    questions = []    for  line  in  response_text.split( '\n' ):        cleaned_line = line.strip()   # remove leading and trailing spaces        if  cleaned_line:            # Remove any possible number prefixes (such as "1.", "2)", "•", "-", etc.)            cleaned_line = re.sub( r'^[\d\-\•\*]+\s*[\.\\)]?\s*' ,  '' , cleaned_line)            # Determine whether it contains a question mark (supports both Chinese and English)            if  '?'  in  cleaned_line  or  '? '  in  cleaned_line:                # All characters end with question mark                question = cleaned_line.rstrip( '?' ).rstrip( '?' ) +  '?'                questions.append(question)    return  questions

def  generate_questions ( text, question_count= 5 , model= "qwen-max" ):    """    Generates answerable questions based on a provided block of text.
    parameter:    text (str): The text content of the question that needs to be generated.    question_count (int): The number of questions to generate.    model (str): Name of the language model used to generate questions.
    return:    List[str]: The generated list of questions.    """    # System instructions: define the behavior rules of AI    system_instruction =  "You are an expert at generating relevant questions from text. Please use only the text provided to create concise questions that focus on key information and concepts."
    # User request template: provide specific tasks and format requirements    user_request =  f"""    Please generate  {question_count}  different questions based on the following text. These questions must be answerable by the text:
    {text}
    Please output the questions as a numbered list, without adding anything else.    """
    # Calling the large model API to generate problems    response = client.chat.completions.create(        model=model,        temperature = 0.7 ,        messages=[            { "role" :  "system" ,  "content" : system_instruction},            { "role" :  "user" ,  "content" : user_request}        ]    )
    # Extract the original response content and remove leading and trailing spaces    raw_questions_text = response.choices[ 0 ].message.content.strip()
    # Use auxiliary functions to extract and filter valid questions    filtered_questions = _extract_questions_from_response(raw_questions_text)
    return  filtered_questions

import  numpy  as  npfrom  typing  import  List ,  Dict ,  Optional

classSimpleVectorStore:    """    Simple NumPy-based vector storage implementation.    """
    def  __init__ ( self ):        """        Initialize the vector database, which contains vectors, text, and metadata lists.        """        self.vectors:  List [np.ndarray] = []    # store vectors        self.texts:  List [ str ] = []             # Store original text        self.metadata_list:  List [ Dict ] = []     #Store metadata        def  add_item ( self, text:  str , vector:  List [ float ], metadata:  Optional [ Dict ] =  None ):        """        Add an entry to the vector library.
        parameter:        text (str): original text content.        vector (List[float]): Vector embedding representation.        metadata (Dict, optional): Optional metadata information.        """        self.vectors.append(np.array(vector))        self.texts.append(text)        self.metadata_list.append(metadata  or  {})        def  similarity_search ( self, query_vector:  List [ float ], top_k:  int  =  5 ) ->  List [ Dict ]:        """        Find the top_k most similar records in the vector library according to the query vector.
        parameter:        query_vector (List[float]): query vector.        top_k (int): The number of results returned.
        return:        List[Dict]: A list of dictionaries containing similar text, metadata, and similarity scores.        """        if not self.vectors:            return  []                # Convert the query vector to a numpy array        query_array = np.array(query_vector)                # Calculate the cosine similarity between each vector and the query vector        similarities = []        for  idx, vector  in  enumerate (self.vectors):            similarity = np.dot(query_array, vector) / (                np.linalg.norm(query_array) * np.linalg.norm(vector)            )            similarities.append((idx, similarity))                # Sort by similarity in descending order        similarities.sort(key= lambda  x: x[ 1 ], reverse= True )                # Build result return        results = []        for  i  in  range ( min (top_k,  len (similarities))):            idx, score = similarities[i]            results.append({                "text" : self.texts[idx],                "metadata" : self.metadata_list[idx],                "similarity_score" :  float (score)            })                return  results

def  process_document ( pdf_path, chunk_size= 1000 , chunk_overlap= 200 , questions_per_chunk= 5 ):    """    The document is processed and question enhancements are generated.
    parameter:    pdf_path(str): PDF file path.    chunk_size (int): The number of characters per text chunk.    chunk_overlap(int): The number of overlapping characters between text chunks.    questions_per_chunk(int): The number of questions to generate per chunk of text.
    return:    Tuple[List[str], SimpleVectorStore]: Processed text chunks and vector storage.    """    print ( "Extracting text from PDF..." )    extracted_text = extract_text_from_pdf(pdf_path)
    print ( "Split text into blocks..." )    text_chunks = chunk_text(extracted_text, chunk_size, chunk_overlap)    print ( f"A total of  { len (text_chunks)}  text chunks were created" )
    vector_store = SimpleVectorStore()
    print ( "Process each text block and generate related questions..." )    for  idx, chunk  in  enumerate (tqdm(text_chunks, desc= "Processing text chunks" )):        # Generate embedding for the current text block        chunk_embedding_response = create_embeddings(chunk)        chunk_embedding = chunk_embedding_response.data[ 0 ].embedding
        # Add text blocks to the vector library        vector_store.add_item(            text=chunk,            vectors=chunk_embedding,            metadata={ "type" :  "chunk" ,  "index" : idx}        )
        # Generate multiple questions for the current text block        questions = generate_questions(chunk, num_questions=questions_per_chunk)
        # Generate embeddings for each question and add them to the vector library        for  q_idx, question  in  enumerate (questions):            question_embedding_response = create_embeddings(question)            question_embedding = question_embedding_response.data[ 0 ].embedding
            # Add the question to the vector library            vector_store.add_item(                text=question,                vectors=question_embedding,                metadata={ "type" :  "question" ,  "chunk_index" : idx,  "original_chunk" : chunk}            )
    return  text_chunks, vector_store

# Define the PDF file pathpdf_file_path =  "knowledge_base/Intelligent Coding Assistant Tongyi Lingma.pdf"
# Process documents (extract text, generate chunks, create questions, build vector library)text_chunks, vector_store = process_document(    pdf_file_path,    chunk_size = 1000 ,    chunk_overlap = 100 ,    questions_per_chunk = 3)
# Output the number of entries in the vector libraryprint ( f"The vector store contains  { len (vector_store.texts)}  entries" )

import  json
search_query =  'Which company is the intelligent coding assistance tool Tongyi Lingma produced by?'
# Find related content using semantic searchsearch_results = semantic_search(search_query, vector_store, k= 5 )
print ( "Query content: " , search_query)print ( "\nSearch results:" )
# Sort results by typedocument_chunks = []matched_questions = []
for  result  in  search_results:    if  result[ "metadata" ][ "type" ] ==  "chunk" :        document_chunks.append(result)    else :        matched_questions.append(result)
# Print document fragmentprint ( "\nRelated document fragment: " )for  index, result  in  enumerate (document_chunks):    print ( f"Context  {index +  1 }  (similarity:  {result[ 'similarity' ]: .4 f} ):" )    print (result[ "text" ][: 300 ] +  "..." )    print ( "=====================================" )
# Print matching questionsprint ( "\nMatched questions: " )for  index, result  in  enumerate (matched_questions):    print ( f"Problem  {index +  1 }  (similarity:  {result[ 'similarity' ]: .4 f} ):" )    print (result[ "text" ])    chunk_index = result[ "metadata" ][ "chunk_index" ]    print ( f"from fragment  {chunk_index} " )    print ( "=====================================" )

Query content: Which company is Tongyi Lingma an intelligent coding assistance tool?
Matching questions:Question 1 (Similarity: 0.9770):Which company provides Tongyi Lingma, an intelligent coding assistance tool, and what capabilities does it mainly provide?From segment 0=====================================Question 2 (Similarity: 0.8629):In addition to individual developers, what versions and special services does Tongyi Lingma provide for enterprises?From segment 0=====================================Question 3 (Similarity: 0.8108):What enterprise plans does Tongyi Lingma provide for customers to choose from?From Clip 1=====================================Question 4 (Similarity: 0.8078):How does Tongyi Lingma's code completion function work, and what kind of code suggestions can it generate for developers?From segment 0=====================================

def  rewrite_query ( original_query, model= "qwen-max" ):    """    Rewrite user queries to make them more specific and detailed to improve retrieval results.        parameter:        original_query(str): user's original query statement        model(str): model name used for rewriting            return:        str: optimized query statement    """    # System prompts: guide AI assistant behavior    system_prompt =  "You are an AI assistant that is good at optimizing search queries. Your task is to rewrite the user's query to be more specific, detailed, and helpful in obtaining relevant information."
    # User prompt: Provide the original query that needs to be rewritten    user_prompt =  f"""    Please rephrase the following query to make it more specific and include relevant terms and concepts that will help you find the right results.
    Original query:  {original_query}
    Rewritten query:    """
    # Generate rewrite results using the specified model    response = client.chat.completions.create(        model=model,        temperature= 0.0 ,   # Set the temperature to 0 to ensure stable output        messages=[            { "role" :  "system" ,  "content" : system_prompt},            { "role" :  "user" ,  "content" : user_prompt}        ]    )
    # Return the rewritten query content and remove the leading and trailing spaces    return  response.choices[ 0 ].message.content.strip()

def  generate_step_back_query ( original_query, model= "qwen-max" ):    """    Generate a more general "step back" query to get broader contextual information.        parameter:        original_query(str): user's original query statement        model(str): the name of the model used to generate the query            return:        str: A broader context query    """    # System prompts: guide AI assistant behavior    system_prompt =  "You are an AI assistant that excels at search strategies. Your task is to expand specific queries into more general forms to obtain relevant contextual information."
    # User prompt: Provide the original query that needs to be generalized    user_prompt =  f"""    Please generate a broader, more general version of the specific query below to gain useful context.
    Original query:  {original_query}
    One step back query:    """
    # Generate a broader query statement using the specified model    response = client.chat.completions.create(        model=model,        temperature = 0.1 ,   # The temperature is slightly higher to increase diversity        messages=[            { "role" :  "system" ,  "content" : system_prompt},            { "role" :  "user" ,  "content" : user_prompt}        ]    )
    # Return the generated query content and remove the leading and trailing spaces    return  response.choices[ 0 ].message.content.strip()

def  decompose_query ( original_query, num_subqueries= 4 , model= "qwen-max" ):    """    Break complex queries into simpler subqueries.
    parameter:        original_query(str): original complex query content        num_subqueries (int): The number of subqueries to generate        model(str): the name of the model used to decompose the query
    return:        List[str]: the subquery list after splitting    """    # System prompt words: guide the behavior logic of AI assistant    system_prompt =  "You are an AI assistant that is good at breaking down complex questions. Your task is to break down complex queries into multiple simpler questions, the answers to which together form the answer to the original question."
    # User prompt: Provide the original query that needs to be decomposed    user_prompt =  f"""    Decompose the following complex query into  {num_subqueries}  simpler subqueries. Each subquery should focus on a different aspect of the original question.
    Original query:  {original_query}
    Please output the results in the following format:    1. [First subquery]    2. [Second subquery]    ...    """
    # Generate subqueries using the specified model    response = client.chat.completions.create(        model=model,        temperature = 0.2 ,   # The temperature is slightly higher to increase diversity        messages=[            { "role" :  "system" ,  "content" : system_prompt},            { "role" :  "user" ,  "content" : user_prompt}        ]    )
    # Extract and process the response content    content = response.choices[ 0 ].message.content.strip()
    # Split the response content by line    lines = content.split( "\n" )    sub_queries = []
    # Parse each row and extract the subquery after the number    for  line  in  lines:        if  line.strip()  and  any (line.strip().startswith( f" {i} ." )  for  i  in  range ( 1 ,  10 )):            query = line.strip()            query = query[query.find( "." ) +  1 :].strip()   # Remove the serial number part and keep the actual content            sub_queries.append(query)
    return  sub_queries

def rerank_with_llm(query, search_results, top_n=3, model="qwen-max"): """ Rerank search results using LLM. Parameters: query(str): user query statement. search_results(List[Dict]): initial list of search results, each element contains document text, metadata and similarity score. top_n(int): number of documents returned after reranking. model(str): name of the LLM model used for scoring. Returns: List[Dict]: list of documents sorted by relevance score. """ print(f"Reranking {len(search_results)} documents...") scored_results = [] # Store results with relevance scores # Define system prompts to guide LLM how to score system_prompt = """You are an expert in assessing the relevance of documents to search queries. Your task is to score documents between 0 and 10 based on how relevant the document is to answer a given query. Scoring guidelines: - 0-2 points: document is completely irrelevant - 3-5 points: The document has some relevant information, but does not directly answer the query 6-8 points: The document is relevant and can partially answer the query 9-10 points: The document is highly relevant and can directly answer the query Please only output an integer score (0 to 10) as the score, do not include other content. """ # Iterate over each search result for idx, result in enumerate(search_results): if idx % 5 == 0: print(f"Scoring document {idx + 1}/{len(search_results)}...") # Construct user prompt words, enter query and document content user_prompt = f"""Query: {query}Document content:{result['text']}Please rate the relevance of this document based on the above query (0-10):""" # Call LLM API to get the score response = client.chat.completions.create( model=model, temperature=0, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ] ) # Extract the scoring results score_text = response.choices[0].message.content.strip() # Use regular expressions to extract numeric scores score_match = re.search(r'\b(10|[0-9])\b', score_text) if score_match: relevance_score = float(score_match.group(1)) else: # If the score cannot be extracted, use the similarity score as a fallback print(f"Warning: Unable to extract score from response: '{score_text}', using similarity score instead") relevance_score = result["similarity"] * 10 # Add the scoring results to the list scored_results.append({ "text": result["text"], "metadata": result["metadata"], "similarity": result["similarity"], "relevance_score": relevance_score }) # Sort the results in descending order by relevance score reranked_results = sorted(scored_results, key=lambda x: x["relevance_score"], reverse=True) # Return the top_n results return reranked_results[:top_n]

def  rerank_with_keywords ( query, doc_results, top_n= 3 ):    """    A simple re-ranking method based on keyword matching and position.    Args:        query(str): the question the user is querying        doc_results(List[Dict]): Initial search results, each dictionary contains text and other metadata        top_n(int): The number of results to return after reordering, the default is the first 3    Returns:        List[Dict]: The results are re-sorted by relevance, keeping only the top_n ones    """    def  extract_keywords ( text ):        """Extract keywords from text"""        return  [word.lower() for  word  in  text.split()iflen(word) >  3 ]    # Extract important keywords from user questions    keywords = extract_keywords(query)    ranked_docs = []   # Create a list to store the ranked documents    for  doc_result  in  doc_results:        document_text = doc_result[ "text" ].lower()   # Convert document text to lowercase for subsequent comparison        # The base score starts from vector similarity, multiplied by 0.5 to indicate that it is not the only determining factor        base_score = doc_result[ "similarity" ] *  0.5        # Initialize keyword scores        keyword_score =  0        for  keyword  in  keywords:            if  keyword  in  document_text:                # If a keyword is found, add 0.1 points                keyword_score +=  0.1                # If the keyword appears in the first 1/4 of the document, it is more likely to answer the question directly, and 0.1 points will be added                first_position = document_text.find(keyword)                if  first_position <  len (document_text) /  4 :   # In the first quarter                    keyword_score +=  0.1                # Add points based on the number of times the keyword appears, but the maximum is 0.2 points                frequency = document_text.count(keyword)                keyword_score +=  min ( 0.05  * frequency,  0.2 )   # add up to 0.2 points        # Calculate the final score: basic score plus keyword score        final_score = base_score + keyword_score        # Store this document and its related information and score in a list        ranked_docs.append({            "text" : doc_result[ "text" ],            "metadata" : doc_result[ "metadata" ],            "similarity" : doc_result[ "similarity" ],            "relevance_score" : final_score        })    # Sort all documents in descending order according to the final relevance score    reranked_docs =  sorted (ranked_docs, key= lambda  x: x[ "relevance_score" ], reverse= True )    # Return the top_n documents with the highest scores    return  reranked_docs[:top_n]

def rag_with_reranking(query, vector_store, reranking_method="llm", top_n=3, model="qwen-max"): """ Full RAG pipeline including reranking. Arguments: query(str): user query vector_store(SimpleVectorStore): vector store reranking_method(str): reranking method ('llm' or 'keywords') top_n(int): number of results to return after reranking model(str): model used to generate answers Returns: Dict: dictionary of results containing query, context and answers """ # Create query embeddings query_embedding = create_embeddings(query) # Initial search (get more results than needed for reranking later) initial_results = vector_store.similarity_search(query_embedding, k=10) # Apply reranking if reranking_method == "llm": reranked_results = rerank_with_llm(query, initial_results, top_n=top_n) elif reranking_method == "keywords": reranked_results = rerank_with_keywords(query, initial_results, top_n=top_n) else: # Do not rerank, directly use the top_n results of the initial search reranked_results = initial_results[:top_n] # Merge the reranked context context = "\n\n===\n\n".join([result["text"] for result in reranked_results]) # Generate answers based on the context response = generate_response(query, context, model) return { "query": query, "reranking_method": reranking_method, "initial_results": initial_results[:top_n], "reranked_results": reranked_results, "context": context,        "response": response }

import numpy as npclass SimpleVectorStore: """ A lightweight vector store implementation using NumPy. """ def __init__(self, dimension=1536): """ Initialize the vector store. Arguments: dimension (int): dimension of the embedding vectors """ self.dimension = dimension self.vectors = [] # store vector data self.documents = [] # store document content self.metadata_list = [] # store metadata def add_documents(self, documents, vectors=None, metadata_list=None): """ Add documents to the vector store. Arguments: documents (List[str]): list of document chunks vectors (List[List[float]], optional): list of embedding vectors metadata_list (List[Dict], optional): list of metadata dictionaries """ if vectors is None: vectors = [None] * len(documents) if metadata_list is None: metadata_list = [{} for _ in range(len(documents))] for doc, vec, metadata in zip(documents, vectors, metadata_list): self.documents.append(doc) self.vectors.append(vec) self.metadata_list.append(metadata) def search(self, query_vector, top_k=5): """ Search for most similar documents. Arguments: query_vector (List[float]): query embedding vector top_k (int): number of results to return Returns: List[Dict]: List of results containing documents, similarity scores and metadata """ if not self.vectors or not self.documents: return [] # Convert query vector to NumPy array query_array = np.array(query_vector) # Compute similarities similarities = [] for index, vector in enumerate(self.vectors): if vector is not None: # Compute cosine similarity similarity = np.dot(query_array, vector) / ( np.linalg.norm(query_array) * np.linalg.norm(vector) ) similarities.append((index, similarity)) # Sort by similarity (descending) similarities.sort(key=lambda x: x[1], reverse=True) # Get top-k results results = [] for index, score in similarities[:top_k]: results.append({ "document": self.documents[index], "score": float(score), "metadata": self.metadata_list[index] }) return results

from  typing  import  List ,  Dict ,  Tuple
def  process_document ( pdf_path:  str , chunk_size:  int  =  800 ) ->  Tuple [ List [ str ], SimpleVectorStore,  Dict ]:    """    Processes documents for use with RSE (Retrieval Enhancement Generation).        parameter:        pdf_path(str): path to the PDF document        chunk_size(int): The size of each text chunk (number of characters)            return:        Tuple[List[str], SimpleVectorStore, Dict]: a tuple containing a list of text chunks, a vector store instance, and document information    """    print ( "Extracting text from document..." )    # Extract text content from PDF files    document_text = extract_text_from_pdf(pdf_path)        print ( "Split text into non-overlapping segments..." )    # Split the extracted text into non-overlapping text blocks    text_chunks = chunk_text(document_text, chunk_size=chunk_size, overlap= 0 )    print ( f"A total of { len (text_chunks)}  text chunks were created  " )        print ( "Generate embedding vector for text block..." )    # Generate embedding vectors for each text block    chunk_embeddings = create_embeddings(text_chunks)        # Create a SimpleVectorStore instance to store vector data    vector_store = SimpleVectorStore()        # Add a document with metadata (including text block index for subsequent reconstruction)    metadata_list = [{ "chunk_index" : index,  "source" : pdf_path}  for  index  in  range ( len (text_chunks))]    vector_store.add_documents(text_chunks, chunk_embeddings, metadata_list)        # Record the original document structure for subsequent splicing    document_info = {        "chunks" : text_chunks,        "source" : pdf_path,    }        return  text_chunks, vector_store, document_info

def  calculate_chunk_values ​​( query:  str , chunks:  List [ str ], vector_store, irrelevant_chunk_penalty:  float  =  0.2 ) ->  List [ float ]:    """    Calculate the value of each document slice, combining its relevance score and location information.
    parameter:        query(str): query text entered by the user        chunks(List[str]): List of text chunks after the document is segmented        vector_store: vector database, containing vector representations of document blocks        irrelevant_chunk_penalty (float): The penalty imposed on irrelevant document chunks. The default value is 0.2            return:        List[float]: List of values ​​corresponding to each document block (floating point number)    """
    # Convert user queries into embedding vectors for semantic matching    query_embedding = create_embeddings([query])[ 0 ]
    # Get the number of all text blocks and search for similarity results    total_chunks =  len (chunks)    search_results = vector_store.search(query_embedding, top_k=total_chunks)
    # Build a mapping dictionary from chunk_index to relevance score    relevance_scores = {        result[ "metadata" ][ "chunk_index" ]: result[ "score" ]        for  result  in  search_results    }
    # Calculate the value based on the relevance score and apply the penalty mechanism for irrelevant blocks    chunk_values ​​= []    for  i  in  range (total_chunks):        score = relevance_scores.get(i,  0.0 )        value = score - irrelevant_chunk_penalty        chunk_values.append(value)
    return  chunk_values

def  reconstruct_segments ( document_chunks:  List [ str ], best_segment_indices:  List [ Tuple [ int ,  int ]] ) ->  List [ Dict ]:    """    Reconstruct the text paragraph based on the optimal slice index.        parameter:        document_chunks(List[str]): All text chunks of the original document        best_segment_indices(List[Tuple[int, int]]): List of start and end indices of the best segment            return:        List[Dict]: List of dictionaries containing the reconstructed paragraphs and their ranges    """    reconstructed_segments = []
    for  start_idx, end_idx  in  best_segment_indices:        segment_text =  " " .join(document_chunks[start_idx:end_idx])        reconstructed_segments.append({            "text" : segment_text,            "segment_range" : (start_idx, end_idx),        })
    return  reconstructed_segments

def  format_segments_for_context ( segments:  List [ Dict ] ) ->  str :    """    Format a paragraph of text into a context string that is usable by a language model.        parameter:        segments(List[Dict]): List of dictionaries containing paragraph text and index ranges            return:        str: formatted context text    """    context_lines = []
    for  index, segment  in  enumerate (segments):        header =  f"SEGMENT  {index +  1 }  (Chunks  {segment[ 'segment_range' ][ 0 ]} - {segment[ 'segment_range' ][ 1 ] -  1 } ):"        context_lines.append(header)        context_lines.append(segment[ "text" ])        context_lines.append( "-"  *  80 )
    return "\n\n" .join(context_lines)

segments = [ { "segment_range": [2, 5], "text": "Artificial intelligence is a branch of computer science that aims to enable machines to simulate human intelligent behavior." }, { "segment_range": [7, 9], "text": "Deep learning is a special type of machine learning method that is particularly good at processing image and speech data." }]

SEGMENT 1 (Chunks 2-4):Artificial intelligence is a branch of computer science that aims to enable machines to simulate human intelligent behavior.--------------------------------------------------------------------------------
SEGMENT 2 (Chunks 7-8):Deep learning is a special type of machine learning that is particularly good at processing image and speech data.--------------------------------------------------------------------------------

def  rag_with_rse ( pdf_path:  str , query:  str , chunk_size:  int  =  800 , penalty:  float  =  0.2 ) ->  Dict :    """    The complete RAG process uses the Relevant Paragraph Extraction (RSE) strategy to filter out the most useful document content.        parameter:        pdf_path(str): PDF document path        query(str): user query        chunk_size(int): text slice size        penalty(float): penalty coefficient for irrelevant slices            return:        Dict: A dictionary containing the query, selected paragraphs, and generated answers    """    print ( "\n=== Start executing the RAG process based on relevant paragraph extraction===" )    print ( f"Query content:  {query} " )
    # Step 1: Process documents and generate vector storage    text_chunks, vector_store, doc_info = process_document(pdf_path, chunk_size)
    # Step 2: Calculate the relevance score and value of each text block    print ( "\nCalculating text block relevance score and value..." )    chunk_values ​​= calculate_chunk_values(query, text_chunks, vector_store, penalty)
    # Step 3: Select the best paragraph based on value    best_segments, scores = find_best_segments(        chunk_values=chunk_values,        max_segment_length = 20 ,        total_max_length = 30 ,        min_segment_value = 0.2    )
    # Step 4: Reconstruct the best paragraph    print ( "\nRebuilding best text paragraph..." )    selected_segments = reconstruct_segments(text_chunks, best_segments)
    # Step 5: Format the context for use with the large model    formatted_context = format_segments_for_context(selected_segments)
    # Step 6: Call the large model to generate the final response    response = generate_response(query, formatted_context)
    # Arrange the output results    result = {        "query" : query,        "segments" : selected_segments,        "response" : response    }
    print ( "\n=== The final reply is as follows===" )    print (response)
    return  result

scores = [model.similarity(query_embedding, sentence_embedding) for sentence in sentences]

def  compress_chunk ( chunk:  str , query:  str , compression_type:  str  =  "selective" , model:  str  =  "qwen-max" ) ->  Tuple [ str ,  float ]:    """    Compresses the retrieved text blocks, keeping only the parts relevant to the query.    parameter:        chunk(str): the text block to be compressed        query(str): user query        compression_type (str): compression method ("selective", "summary", or "extraction")        model(str): LLM model name used    return:        Tuple[str, float]: compressed text block and compression ratio (percentage)    """    # Build system prompts based on different compression types    if  compression_type ==  "selective" :        system_prompt =  """You are an expert at filtering information.        Your task is to analyze the document snippet and extract the sentences or paragraphs that are directly related to the user query. Remove all irrelevant content.        Output requirements:        1. Include only the text that helps answer the question        2. Keep the original wording of relevant sentences (do not paraphrase)        3. Maintain the original order        4. Include all relevant content, even if it appears to be repetitive        5. Exclude any text that is not relevant to the question        Please output in plain text format without adding additional instructions. """        elif  compression_type ==  "summary" :        system_prompt =  """You are a summary expert.        Your task is to succinctly summarize the given document snippet, focusing only on the information relevant to the user's query.        Output requirements:        1. Be concise but cover all relevant content        2. Focus on information relevant to your query        3. Ignore irrelevant details        4. Write in a neutral, objective tone        Please output in plain text format without adding additional instructions. """        else :   # extraction        system_prompt =  """You are an information extraction expert.        Your task is to extract the exact sentences that contain relevant information from the document fragment to answer the user's query.        Output requirements:        1. Include only relevant sentences from the original text        2. Keep the original sentence unchanged (do not modify it)        3. Include only sentences directly related to the question        4. Separate each sentence with a line break        5. Do not add any comments or other content        Please output in plain text format without adding additional instructions. """    # Build user prompt    user_prompt =  f"""        Query: {query}        Document snippet:        {chunk}        Extract content relevant to the query.    """    # Call the large model API for compression    response = client.chat.completions.create(        model=model,        messages=[            { "role" :  "system" ,  "content" : system_prompt},            { "role" :  "user" ,  "content" : user_prompt}        ],        temperature = 0    )    # Get the compressed content    compressed_content = response.choices[ 0 ].message.content.strip()    # Calculate the compression ratio    original_length =  len (chunk)    compressed_length =  len (compressed_content)    compression_ratio = (original_length - compressed_length) / original_length *  100
    return  compressed_content, compression_ratio

def  rag_with_compression ( pdf_path:  str , query:  str , k:  int  =  10 , compression_type:  str  =  "selective" , model:  str  =  "qwen-max" ) ->  Dict :    """    The complete RAG process,using the context compression strategy to reduce the input length.
    parameter:        pdf_path(str): PDF file path        query(str): user query        k (int): the number of text blocks to retrieve initially        compression_type(str): compression method        model(str): the name of the large model used
    return:        Dict: A dictionary containing the query, the content before and after compression, the response, etc.    """
    print ( "\n=== Context compression RAG process starts===" )    print ( f"Query content:  {query} " )    print ( f"Compression type:  {compression_type} " )
    # Load the document and create vector storage    vector_store = process_document(pdf_path)
    # Create query embeddings    query_embedding = create_embeddings(query)
    # Retrieve the top k most relevant text blocks    print ( f"Retrieving the first  {k}  related text blocks..." )    results = vector_store.similarity_search(query_embedding, k=k)    retrieved_chunks = [result[ "text" ]  for  result  in  results]
    # Compress each text block    compressed_results = batch_compress_chunks(retrieved_chunks, query, compression_type, model)    compressed_chunks = [result[ 0 ]  for  result  in  compressed_results]    compression_ratios = [result[ 1 ]  for  result  in  compressed_results]
    # Filter empty content    valid_compressed_data = [(chunk, ratio)  for  chunk, ratio  in  zip (compressed_chunks, compression_ratios)  if  chunk.strip()]
    ifnot valid_compressed_data:        # Fallback to the original text block when all compressed text is empty        print ( "Warning: All text blocks were compressed to nothing. The original text blocks will be used." )        valid_compressed_data = [(chunk,  0.0 )  for  chunk  in  retrieved_chunks]    else :        compressed_chunks, compression_ratios =  zip (*valid_compressed_data)
    # Build context    context =  "\n\n---\n\n" .join(compressed_chunks)
    # Generate final response    print ( "Generate reply based on compressed text block..." )    response = generate_response(query, context, model)
    # Return results    result = {        "query" : query,        "original_chunks" : retrieved_chunks,        "compressed_chunks" : compressed_chunks,        "compression_ratios" : compression_ratios,        "context_length_reduction" :  f" { sum (compression_ratios) /  len (compression_ratios): .2 f} %" ,        "response" : response    }
    print ( "\n=== Final reply===" )    print (response)
    return  result    

from  typing  import  List ,  Dict ,  Optional ,  Tuple ,  Callableimport  numpy  as  np

classSimpleVectorStore:    """    A simple vector database implementation based on NumPy.
    This class provides an in-memory vector storage and retrieval system, supporting basic similarity searches using cosine similarity.    """
    def  __init__ ( self ):        """        Initialize the vector database, which contains three parallel lists:        - vectors: stores embedding vectors (NumPy arrays)        - texts: stores raw text blocks        - metadata: stores metadata for each text block        """        self.vectors:  List [np.ndarray] = []   # Embedding vector list        self.texts:  List [ str ] = []            # Text content list        self.metadata:  List [ Dict ] = []        # metadata list
    def  add_item ( self, text:  str , embedding:  List [ float ], metadata:  Optional [ Dict ] =  None ) ->  None :        """        Adds an entry to the vector database.
        parameter:            text (str): the text content to be stored            embedding (List[float]): represents the embedding vector of the text            metadata (Dict, optional): Optional metadata dictionary        """        self.vectors.append(np.array(embedding))        self.texts.append(text)        self.metadata.append(metadata  or  {})
    def  similarity_search (        self,        query_embedding:  List [ float ],        k:  int  =  5 ,        filter_func:  Optional [ Callable [[ Dict ],  bool ]] =  None    ) ->  List [ Dict ]:        """        Use cosine similarity to find the entries that are most similar to the query vector.
        parameter:            query_embedding (List[float]): query embedding            k (int): number of results to return            filter_func (Callable, optional): filter function used to filter results based on metadata
        return:            List[Dict]: A list of results containing text, metadata, and relevance scores        """        if not self.vectors:            return  []
        query_vector = np.array(query_embedding)        similarities = []
        for  i, vector  in  enumerate (self.vectors):            if  filter_func andnot filter_func(self.metadata[i]):                continue
            similarity = np.dot(query_vector, vector) / (                np.linalg.norm(query_vector) * np.linalg.norm(vector)            )            similarities.append((i, similarity))
        similarities.sort(key= lambda  x: x[ 1 ], reverse= True )
        results = []        for  i  in  range ( min (k,  len (similarities))):            idx, score = similarities[i]            results.append({                "text" : self.texts[idx],                "metadata" : self.metadata[idx],                "similarity" : score,                "relevance_score" : self.metadata[idx].get( "relevance_score" , score)            })
        return  results

def  get_user_feedback ( query:  str , response:  str , relevance:  int , quality:  int , comments:  str  =  "" ) ->  Dict :    """    Format user feedback as a dictionary.
    parameter:        query(str): user's question        response(str): the system's answer        relevance(int): relevance score (1-5)        quality(int): answer quality score (1-5)        comments(str): optional comments
    return:        Dict: Formatted feedback dictionary    """    return {        "query" : query,        "response" : response,        "relevance" :  int (relevance),        "quality" :  int (quality),        "comments" : comments,        "timestamp" : datetime.now().isoformat()    }

def  store_feedback ( feedback:  Dict , feedback_file:  str  =  "feedback_data.json" ) ->  None :    """    Save the feedback data into a JSON file.
    parameter:        feedback(Dict): feedback data        feedback_file(str): file path    """    with  open (feedback_file,  "a" )  as  f:        json.dump(feedback, f)        f.write( "\n" )

def  load_feedback_data ( feedback_file:  str  =  "feedback_data.json" ) ->  List [ Dict ]:    """    Load feedback data from a file.
    parameter:        feedback_file(str): file path
    return:        List[Dict]: Feedback data list    """    feedback_data = []    try :        with  open (feedback_file,  "r" )  as  f:            for  line  in  f:                if  line.strip():                    feedback_data.append(json.loads(line.strip()))    except  FileNotFoundError:        print ( "Feedback file not found. Starting with empty feedback." )
    return  feedback_data

def  process_document ( pdf_path:  str , chunk_size:  int  =  1000 , chunk_overlap:  int  =  200 ) ->  Tuple [ List [ str ], SimpleVectorStore]:    """    Process documents for RAG process.
    step:    1. Extract text from PDF    2. Segment the text    3. Create an embed    4. Store in vector database
    parameter:        pdf_path(str): PDF file path        chunk_size(int): size of each slice        chunk_overlap(int): the number of overlapping characters between slices
    return:        Tuple[List[str], SimpleVectorStore]: text slices and vector database    """    print ( "Extracting text from PDF..." )    extracted_text = extract_text_from_pdf(pdf_path)
    print ( "Splitting text..." )    chunks = chunk_text(extracted_text, chunk_size, chunk_overlap)    print ( f"Generated  { len (chunks)}  text slices" )
    print ( "Creating embedding for slice..." )    chunk_embeddings = create_embeddings(chunks)
    store = SimpleVectorStore()
    for  i, (chunk, embedding)  in  enumerate ( zip (chunks, chunk_embeddings)):        store.add_item(            text=chunk,            embedding=embedding,            metadata={                "index" : i,                "source" : pdf_path,                "relevance_score" :  1.0 ,                "feedback_count" :  0            }        )
    print ( f"  { len (chunks)}  slices have been added to the vector database" )    return  chunks, store