Interviews

Interview class

This module contains the Interview class, which is responsible for conducting an interview asynchronously.

class edsl.jobs.interviews.Interview.Interview(agent: Agent, survey: Survey, scenario: Scenario, model: Type['LanguageModel'], debug: bool | None = False, iteration: int = 0, cache: 'Cache' | None = None, sidecar_model: 'LanguageModel' | None = None, skip_retry: bool = False, raise_validation_errors: bool = True, indices: dict = None)

Bases: object

An ‘interview’ is one agent answering one survey, with one language model, for a given scenario.

The main method is async_conduct_interview, which conducts the interview asynchronously. Most of the class is dedicated to creating the tasks for each question in the survey, and then running them.

static _extract_valid_results(tasks, invigilators: List['InvigilatorABC'], exceptions) → Generator['Answers', None, None]

Extract the valid results from the list of results.

It iterates through the tasks and invigilators, and yields the results of the tasks that are done. If a task is not done, it raises a ValueError. If an exception is raised in the task, it records the exception in the Interview instance except if the task was cancelled, which is expected behavior.

>>> i = Interview.example()
>>> result, _ = asyncio.run(i.async_conduct_interview())

async async_conduct_interview(model_buckets: ModelBuckets | None = None, stop_on_exception: bool = False, sidecar_model: 'LanguageModel' | None = None, raise_validation_errors: bool = True) → tuple['Answers', List[dict[str, Any]]]

Conduct an Interview asynchronously. It returns a tuple with the answers and a list of valid results.

Parameters:

• model_buckets – a dictionary of token buckets for the model.

• debug – run without calls to LLM.

• stop_on_exception – if True, stops the interview if an exception is raised.

• sidecar_model – a sidecar model used to answer questions.

Example usage:

>>> i = Interview.example()
>>> result, _ = asyncio.run(i.async_conduct_interview())
>>> result['q0']
'yes'

>>> i = Interview.example(throw_exception = True)
>>> result, _ = asyncio.run(i.async_conduct_interview())
>>> i.exceptions
{'q0': ...
>>> i = Interview.example()
>>> result, _ = asyncio.run(i.async_conduct_interview(stop_on_exception = True))
Traceback (most recent call last):
...
asyncio.exceptions.CancelledError

duplicate(iteration: int, cache: Cache) → Interview

Duplicate the interview, but with a new iteration number and cache.

>>> i = Interview.example()
>>> i2 = i.duplicate(1, None)
>>> i.iteration + 1 == i2.iteration
True

classmethod example(throw_exception: bool = False) → Interview

Return an example Interview instance.

classmethod from_dict(d: dict[str, Any]) → Interview

Return an Interview instance from a dictionary.

property has_exceptions: bool

Return True if there are exceptions.

property interview_status: InterviewStatusDictionary

Return a dictionary mapping task status codes to counts.

property task_status_logs: InterviewStatusLog

Return the task status logs for the interview.

The keys are the question names; the values are the lists of status log changes for each task.

to_dict(include_exceptions=True, add_edsl_version=True) → dict[str, Any]

Return a dictionary representation of the Interview instance. This is just for hashing purposes.

>>> i = Interview.example()
>>> hash(i)
193593189022259693

property token_usage: InterviewTokenUsage

Determine how many tokens were used for the interview.