divided code in files and working api

app.py

@@ -1,233 +1,78 @@
-import random
-from typing import List, Dict
-import openai  # or any LLM wrapper like llama_index, langchain, etc.
-import json
+import os
+from fastapi import FastAPI, HTTPException, Header
 from pydantic import BaseModel
 from dotenv import load_dotenv
-from fastapi import FastAPI,Request, HTTPException, Header
+from schemas import GenerateInterviewsRequest, GenerateInterviewsResponse, GenerateReportRequest, GenerateReportResponse, GenerateUsersRequest, GenerateUsersResponse, GenerateRequest, GenerateResponse
 import uvicorn
-import os
+import logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
 load_dotenv()
 
 API_KEY = os.getenv("MY_API_KEY")
-latest_fleet_data = {"fleet": [], "scope": ""}
-
-### 1. Configuration ###
-LLM_MODEL = "gpt-4o-mini"
-
-### 2. Utilities ###
-def call_llm(prompt: str,response_format=None) -> str:
-    client = openai.OpenAI()
-    if response_format:
-        response = client.beta.chat.completions.parse(
-            model=LLM_MODEL,
-            messages=[{"role": "user", "content": prompt}],
-            response_format=response_format,
-            temperature=0
-        )
-    else:
-        response = client.chat.completions.create(
-            model=LLM_MODEL,
-            messages=[{"role": "user", "content": prompt}],
-            temperature=0
-        )
-    return response.choices[0].message.content
-
-### 3. Generate User Parameters ###
-def generate_user_parameters(audience: str, scope: str) -> List[str]:
-    # Combine standard parameters with additional parameters
-    standard_parameters = ["Name", "Age", "Location", "Profession"]
-    prompt = f"""
-You are an expert customer researcher.
-
-Your task is to help define user personas for a specific audience and a specific research scope.
 
-This is the desired audience: {audience} 
-This is the research scope: {scope}
-
-Start from the following 4 standard demographic parameters: {standard_parameters}
-
-Your goal is to suggest 4 additional parameters that are especially relevant for capturing behaviors, attitudes, or characteristics important for this audience and scope.
-
-Only suggest parameters that will meaningfully help differentiate users in this specific context.    
-"""
+app = FastAPI()
+""" 
+@app.post("/generate-users", response_model=GenerateUsersResponse)
+def generate_users(request: GenerateUsersRequest, x_api_key: str = Header(...)):
+    if x_api_key != API_KEY:
+        raise HTTPException(status_code=403, detail="Invalid API Key")
     
-    class Response(BaseModel):
-        additional_parameters: list[str]
-
-    response = call_llm(prompt=prompt,response_format=Response)
-    additional_parameters = json.loads(response)["additional_parameters"]
-
-    print(f"prompt: {prompt}\n response: {response}")
-
+    from core import generate_user_parameters, generate_synthetic_personas
     
-    return standard_parameters + additional_parameters
-
-### 4. Synthetic User Creation ###
-def generate_synthetic_personas(parameters: List[str], num_personas: int,audience:str) -> List[Dict]:
-    # Create a prompt for the LLM to generate diversified personas
-    if num_personas > 1:
-        prompt = f"""
-    You are an expert in user persona creation.
-
-    Generate {num_personas} diversified user personas based on the following parameters:
-    {parameters}
-
-    Each persona should be unique and realistic. Provide the output as a list. Take into account that the desired audience is the follwoing: {audience}.
-    """
-    else:
-        prompt = f"""
-    You are an expert in user persona creation.
-
-    Generate a user persona based on the following parameters:
-    {parameters}
-
-    The persona should be unique and realistic. Provide the output as a list. Take into account that the desired audience is the follwoing: {audience}.
-    """
-
-    class Parameter(BaseModel):
-        parameter_name: str
-        value: str
-
-    class User(BaseModel):
-        user_persona: list[Parameter]
+    user_parameters = generate_user_parameters(request.audience, request.scope)
+    personas = generate_synthetic_personas(parameters=user_parameters, num_personas=request.n, audience=request.audience)
+    
+    return GenerateUsersResponse(users_personas=personas)
 
-    class Response(BaseModel):
-        users_personas: list[User]
+@app.post("/generate-interviews", response_model=GenerateInterviewsResponse)
+def generate_interviews(request: GenerateInterviewsRequest, x_api_key: str = Header(...)):
+    if x_api_key != API_KEY:
+        raise HTTPException(status_code=403, detail="Invalid API Key")
+    
+    from core import generate_fleet
     
+    fleet = generate_fleet(n=request.n, parameters=request.users_personas, questions=request.questions, audience=request.scope)
     
-    # Call the LLM to generate the personas
-    response = call_llm(prompt=prompt,response_format=Response)
-    print("Generated Personas:", response)
-    # Parse the response into a list of dictionaries
-    response=json.loads(response)
+    return GenerateInterviewsResponse(fleet=fleet)
 
-    transformed_data = {
-    "users_personas": [
-        {item["parameter_name"]: item["value"] for item in user["user_persona"]}
-        for user in response["users_personas"]
-    ]
-}
+@app.post("/generate-report", response_model=GenerateReportResponse)
+def generate_report(request: GenerateReportRequest, x_api_key: str = Header(...)):
+    if x_api_key != API_KEY:
+        raise HTTPException(status_code=403, detail="Invalid API Key")
     
-    return transformed_data
+    from core import generate_report
     
-
-### 5. Q&A with Persona ###
-from concurrent.futures import ThreadPoolExecutor
-
-def ask_questions_to_persona(persona: dict, questions: List[str]) -> List[str]:
-    """
-    Ask questions to a persona in parallel using a maximum of 10 workers.
-
-    Args:
-        persona (dict): The user persona as a dictionary.
-        questions (List[str]): A list of questions to ask the persona.
-
-    Returns:
-        List[str]: A list of answers corresponding to the questions.
-    """
-    def ask_single_question(question: str) -> str:
-        # Function to ask a single question
-        prompt = f"""Act as if you are this user persona:\n{persona}\n\nAnswer the following question as if you were them:\n{question}"""
-        return call_llm(prompt, False)
-
-    # Use ThreadPoolExecutor to ask questions in parallel
-    max_workers = min(8, len(questions))  # Limit the number of workers to 8 or the number of questions
-    with ThreadPoolExecutor(max_workers=max_workers) as executor:
-        answers = list(executor.map(ask_single_question, questions))
-
-    return answers
-
-### 6. Fleet Generator ###
-def generate_fleet(n: int, parameters: Dict, questions: List[str],audience:str) -> List[Dict]:
-    fleet = []
-    users_personas = generate_synthetic_personas(parameters=parameters,num_personas=n,audience=audience)["users_personas"]
-    i=0
-    for persona_dict in users_personas:
-        i+=1
-        print("index:", i)
-        print("Persona Dictionary:", persona_dict)
-        answers = ask_questions_to_persona(persona_dict, questions)
-        print("Answers:", answers)
-        persona_dict["answers"] = answers
-        fleet.append(persona_dict)
-    print("Fleet:", fleet)
-    return fleet
-
-def generate_report() -> str:
-    fleet = latest_fleet_data["fleet"]
-    scope = latest_fleet_data["scope"]
-
-    content = f"Scope of Research:\n{scope}\n\n"
-    for i, user in enumerate(fleet, 1):
-        content += f"### User {i} ###\n"
-        # Iterate over the dictionary values to display all attributes
-        for key, value in user.items():
-            if key != "answers":  # Skip the "answers" key for now
-                content += f"{key}: {value}\n"
-        content += "\n"
-        # Add answers to the output
-        for j, answer in enumerate(user.get("answers", []), 1):
-            content += f"Q{j}: {answer}\n\n"
-        content += "\n---\n\n"
-
-    prompt = f"Write a research report based on the following scope and interviews:\n\nScope:\n{scope}\n\nQ&A Content:\n{content}\n\nInclude a section where you extract all the important insights from the interviews following the scope.\n\n"
-
-    report_text = call_llm(prompt, False)
+    report = generate_report(fleet=request.fleet, scope=request.scope)
     
-    return report_text
-
-def main(audience: str, scope: str, n: int, questions: List[str]) -> tuple[str, str]:
-    # Generate user parameters based on audience and scope
-    user_parameters = generate_user_parameters(audience, scope)
-
-    # Generate fleet
-    fleet = generate_fleet(n, user_parameters, questions, audience)
-
-    # Format interviews
-    interviews_output = ""
-    for i, user in enumerate(fleet, 1):
-        interviews_output += f"### User {i} ###\n"
-        for key, value in user.items():
-            if key != "answers":
-                interviews_output += f"{key}: {value}\n"
-        interviews_output += "\n"
-        for j, answer in enumerate(user.get("answers", []), 1):
-            interviews_output += f"Q{j}: {answer}\n\n"
-        interviews_output += "\n---\n\n"
+    return GenerateReportResponse(report=report) """
 
-    # Generate report using the modified generate_report function
-    latest_fleet_data["fleet"] = fleet
-    latest_fleet_data["scope"] = scope
-    report = generate_report()
-
-    return interviews_output, report
+@app.post("/generate", response_model=GenerateResponse)
+def generate_all(request: GenerateRequest, x_api_key: str = Header(...)):
+    if x_api_key != API_KEY:
+        logger.warning("Unauthorized access attempt.")
+        raise HTTPException(status_code=403, detail="Invalid API Key")
 
-app = FastAPI()
+    from core import generate_user_parameters, generate_synthetic_personas, generate_fleet, generate_report
 
-class InterviewRequest(BaseModel):
-    audience: str
-    scope: str
-    n: int
-    questions: list[str]
+    logger.info(f"Generating user personas for audience: {request.audience}, scope: {request.scope}, n: {request.n}")
+    user_parameters = generate_user_parameters(request.audience, request.scope)
+    logger.info(f"User parameters: {user_parameters}")
+    users = generate_synthetic_personas(parameters=user_parameters, num_personas=request.n, audience=request.audience)["users_personas"]
+    logger.info(f"Generated personas:\n{users}")
 
-class InterviewResponse(BaseModel):
-    interviews_output: str
-    report: str
+    logger.info("Generating interviews for each persona.")
+    fleet = generate_fleet(n=request.n, parameters=user_parameters, questions=request.questions, audience=request.audience)
+    logger.info(f"Generated fleet:\n{fleet}")
 
+    logger.info("Generating report based on all interviews.")
+    report_text = generate_report(fleet=fleet, scope=request.scope)
+    logger.info("Report generation completed.")
 
-@app.post("/generate", response_model=InterviewResponse)
-def generate_interviews(request: InterviewRequest, x_api_key: str = Header(...)):
-    if x_api_key != API_KEY:
-        raise HTTPException(status_code=403, detail="Invalid API Key")
-        
-    interviews_output, report = main(
-        audience=request.audience,
-        scope=request.scope,
-        n=request.n,
-        questions=request.questions
-    )
-    return InterviewResponse(interviews_output=interviews_output, report=report)
+    return {
+        "users": fleet,
+        "report": report_text
+    }
 
-#if __name__ == "__main__": uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=False)
+if __name__ == "__main__": uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=False)

core.py

@@ -0,0 +1,115 @@
+import json
+from typing import List, Dict
+from pydantic import BaseModel
+from concurrent.futures import ThreadPoolExecutor
+from utils import call_llm
+
+LLM_MODEL = "gpt-4o-mini"
+
+def generate_user_parameters(audience: str, scope: str) -> List[str]:
+    standard_parameters = ["Name", "Age", "Location", "Profession"]
+    prompt = f"""
+You are an expert customer researcher.
+
+Your task is to help define user personas for a specific audience and a specific research scope.
+
+This is the desired audience: {audience} 
+This is the research scope: {scope}
+
+Start from the following 4 standard demographic parameters: {standard_parameters}
+
+Your goal is to suggest 4 additional parameters that are especially relevant for capturing behaviors, attitudes, or characteristics important for this audience and scope.
+
+Only suggest parameters that will meaningfully help differentiate users in this specific context. The parameters should be tailored for the audience and scope but also not too specific.    
+"""
+
+    class Response(BaseModel):
+        additional_parameters: list[str]
+
+    response = call_llm(prompt=prompt, response_format=Response)
+    additional_parameters = json.loads(response)["additional_parameters"]
+
+    return standard_parameters + additional_parameters
+
+
+def generate_synthetic_personas(parameters: List[str], num_personas: int, audience: str) -> Dict:
+    prompt = f"""
+You are an expert in user persona creation.
+
+Generate {num_personas} diversified user personas based on the following parameters:
+{parameters}
+
+Each persona should be unique and realistic, ensure variability. Provide the output as a list. Take into account that the desired audience is the following: {audience}.
+Ensure that the list contains exactly {num_personas} personas.
+"""
+
+    class Parameter(BaseModel):
+        parameter_name: str
+        value: str
+
+    class User(BaseModel):
+        user_persona: list[Parameter]
+
+    class Response(BaseModel):
+        users_personas: list[User]
+
+    response = call_llm(prompt=prompt, response_format=Response)
+    response = json.loads(response)
+
+    transformed_data = {
+        "users_personas": [
+            {item["parameter_name"]: item["value"] for item in user["user_persona"]}
+            for user in response["users_personas"]
+        ]
+    }
+
+    return transformed_data
+
+
+def ask_questions_to_persona(persona: dict, questions: List[str]) -> List[str]:
+    def ask_single_question(question: str) -> str:
+        prompt = f"""
+Act as if you were this user persona:
+{persona}
+You need to impersonate this user persona and answer the following question as if you were that user:
+{question}
+Never sart the sentences in the following way: As "name of the person"...
+Try to sound natural and authentic, as if you were the user persona.
+Make sure to answer the question in a way that is relevant to the user persona.
+"""
+        return call_llm(prompt, False)
+
+    max_workers = min(5, len(questions))
+    with ThreadPoolExecutor(max_workers=max_workers) as executor:
+        answers = list(executor.map(ask_single_question, questions))
+
+    return answers
+
+
+def generate_fleet(n: int, parameters: Dict, questions: List[str], audience: str) -> List[Dict]:
+    fleet = []
+    users_personas = generate_synthetic_personas(parameters=parameters, num_personas=n, audience=audience)["users_personas"]
+    for persona_dict in users_personas:
+        answers = ask_questions_to_persona(persona_dict, questions)
+        persona_dict["answers"] = answers
+        fleet.append(persona_dict)
+    return fleet
+
+
+def generate_report(fleet,scope) -> str:
+
+    content = f"Scope of Research:\n{scope}\n\n"
+    for i, user in enumerate(fleet, 1):
+        content += f"### User {i} ###\n"
+        for key, value in user.items():
+            if key != "answers":
+                content += f"{key}: {value}\n"
+        content += "\n"
+        for j, answer in enumerate(user.get("answers", []), 1):
+            content += f"Q{j}: {answer}\n\n"
+        content += "\n---\n\n"
+
+    prompt = f"Write a research report based on the following scope and interviews:\n\nScope:\n{scope}\n\nQ&A Content:\n{content}\n\nInclude a section where you extract all the important insights from the interviews following the scope.\n\n"
+    report_text = call_llm(prompt, False)
+
+    return report_text

schemas.py

@@ -0,0 +1,36 @@
+from pydantic import BaseModel
+from typing import List, Dict
+
+class GenerateUsersRequest(BaseModel):
+    audience: str
+    scope: str
+    n: int
+
+class GenerateUsersResponse(BaseModel):
+    users_personas: List[Dict[str, str]]
+
+class GenerateInterviewsRequest(BaseModel):
+    users_personas: List[Dict[str, str]]
+    questions: List[str]
+    scope: str
+
+class GenerateInterviewsResponse(BaseModel):
+    fleet: List[Dict[str, str]]
+
+class GenerateReportRequest(BaseModel):
+    fleet: List[Dict[str, str]]
+    scope: str
+
+class GenerateReportResponse(BaseModel):
+    report: str
+
+class GenerateRequest(BaseModel):
+    audience: str
+    scope: str
+    n: int
+    questions: List[str]
+
+class GenerateResponse(BaseModel):
+    users_personas: List[dict]
+    interviews: List[dict]
+    report: str

utils.py

@@ -0,0 +1,21 @@
+import openai
+import os
+
+LLM_MODEL = "gpt-4o-mini"
+
+def call_llm(prompt: str, response_format=None) -> str:
+    client = openai.OpenAI()
+    if response_format:
+        response = client.beta.chat.completions.parse(
+            model=LLM_MODEL,
+            messages=[{"role": "user", "content": prompt}],
+            response_format=response_format,
+            temperature=0
+        )
+    else:
+        response = client.chat.completions.create(
+            model=LLM_MODEL,
+            messages=[{"role": "user", "content": prompt}],
+            temperature=0
+        )
+    return response.choices[0].message.content