Spaces:
Sleeping
Sleeping
File size: 5,566 Bytes
c1d0c23 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | import os
import json
import chevron
import logging
from pydantic import BaseModel
from typing import Optional, List
from tinytroupe import openai_utils
from tinytroupe.agent import TinyPerson
from tinytroupe import config
import tinytroupe.utils as utils
default_max_content_display_length = config["OpenAI"].getint("MAX_CONTENT_DISPLAY_LENGTH", 1024)
class ValidationResponse(BaseModel):
"""Response structure for the validation process"""
questions: Optional[List[str]] = None
next_phase_description: Optional[str] = None
score: Optional[float] = None
justification: Optional[str] = None
is_complete: bool = False
class TinyPersonValidator:
@staticmethod
def validate_person(person, expectations=None, include_agent_spec=True, max_content_length=default_max_content_display_length) -> tuple[float, str]:
"""
Validate a TinyPerson instance using OpenAI's LLM.
This method sends a series of questions to the TinyPerson instance to validate its responses using OpenAI's LLM.
The method returns a float value representing the confidence score of the validation process.
If the validation process fails, the method returns None.
Args:
person (TinyPerson): The TinyPerson instance to be validated.
expectations (str, optional): The expectations to be used in the validation process. Defaults to None.
include_agent_spec (bool, optional): Whether to include the agent specification in the prompt. Defaults to False.
max_content_length (int, optional): The maximum length of the content to be displayed when rendering the conversation.
Returns:
float: The confidence score of the validation process (0.0 to 1.0), or None if the validation process fails.
str: The justification for the validation score, or None if the validation process fails.
"""
# Initiating the current messages
current_messages = []
# Generating the prompt to check the person
check_person_prompt_template_path = os.path.join(os.path.dirname(__file__), 'prompts/check_person.mustache')
with open(check_person_prompt_template_path, 'r', encoding='utf-8', errors='replace') as f:
check_agent_prompt_template = f.read()
system_prompt = chevron.render(check_agent_prompt_template, {"expectations": expectations})
# use dedent
import textwrap
user_prompt = textwrap.dedent(\
"""
Now, based on the following characteristics of the person being interviewed, and following the rules given previously,
create your questions and interview the person. Good luck!
""")
if include_agent_spec:
user_prompt += f"\n\n{json.dumps(person._persona, indent=4)}"
# TODO this was confusing the expectations
#else:
# user_prompt += f"\n\nMini-biography of the person being interviewed: {person.minibio()}"
logger = logging.getLogger("tinytroupe")
logger.info(f"Starting validation of the person: {person.name}")
# Sending the initial messages to the LLM
current_messages.append({"role": "system", "content": system_prompt})
current_messages.append({"role": "user", "content": user_prompt})
message = openai_utils.client().send_message(current_messages, response_format=ValidationResponse, enable_pydantic_model_return=True)
max_iterations = 10 # Limit the number of iterations to prevent infinite loops
cur_iteration = 0
while cur_iteration < max_iterations and message is not None and not message.is_complete:
cur_iteration += 1
# Check if we have questions to ask
if message.questions:
# Format questions as a text block
if message.next_phase_description:
questions_text = f"{message.next_phase_description}\n\n"
else:
questions_text = ""
questions_text += "\n".join([f"{i+1}. {q}" for i, q in enumerate(message.questions)])
current_messages.append({"role": "assistant", "content": questions_text})
logger.info(f"Question validation:\n{questions_text}")
# Asking the questions to the persona
person.listen_and_act(questions_text, max_content_length=max_content_length)
responses = person.pop_actions_and_get_contents_for("TALK", False)
logger.info(f"Person reply:\n{responses}")
# Appending the responses to the current conversation and checking the next message
current_messages.append({"role": "user", "content": responses})
message = openai_utils.client().send_message(current_messages, response_format=ValidationResponse, enable_pydantic_model_return=True)
else:
# If no questions but not complete, something went wrong
logger.warning("LLM did not provide questions but validation is not complete")
break
if message is not None and message.is_complete and message.score is not None:
logger.info(f"Validation score: {message.score:.2f}; Justification: {message.justification}")
return message.score, message.justification
else:
logger.error("Validation process failed to complete properly")
return None, None |