Asynchronously generate schema using the provided model. Args: prompt: The prompt to send to the model schema: The schema to validate the response against model: The model to use Returns: The validated schema object
(
prompt: str,
schema: BaseModel,
model: DeepEvalBaseLLM = None,
)
| 44 | |
| 45 | |
| 46 | async def a_generate( |
| 47 | prompt: str, |
| 48 | schema: BaseModel, |
| 49 | model: DeepEvalBaseLLM = None, |
| 50 | ) -> BaseModel: |
| 51 | """ |
| 52 | Asynchronously generate schema using the provided model. |
| 53 | |
| 54 | Args: |
| 55 | prompt: The prompt to send to the model |
| 56 | schema: The schema to validate the response against |
| 57 | model: The model to use |
| 58 | |
| 59 | Returns: |
| 60 | The validated schema object |
| 61 | """ |
| 62 | _, using_native_model = initialize_model(model=model) |
| 63 | |
| 64 | if using_native_model: |
| 65 | res, _ = await model.a_generate(prompt=prompt, schema=schema) |
| 66 | return res |
| 67 | else: |
| 68 | try: |
| 69 | res = await model.a_generate(prompt=prompt, schema=schema) |
| 70 | if isinstance(res, str): |
| 71 | data = trimAndLoadJson(res) |
| 72 | return schema(**data) |
| 73 | else: |
| 74 | return res |
| 75 | except TypeError: |
| 76 | res = await model.a_generate(prompt) |
| 77 | data = trimAndLoadJson(res) |
| 78 | if schema == SyntheticDataList: |
| 79 | data_list = [SyntheticData(**item) for item in data["data"]] |
| 80 | return SyntheticDataList(data=data_list) |
| 81 | else: |
| 82 | return schema(**data) |
no test coverage detected