Structured Response¶
By default, agents will generate plane text and JSON outputs, and store them in the TaskOutput
object.
- Ref.
TaskOutput
class
But you can choose to generate Pydantic class or specifig JSON object as response.
[var]
response_schema: Optional[Type[BaseModel] | List[ResponseField]] = None
1. Pydantic¶
Add a custom Pydantic class
to the response_schema
field to generate a structured response.
The custom class can accept one layer of a nested child as you can see in the following code snippet:
import versionhq as vhq
from pydantic import BaseModel
from typing import Any
# 1. Define Pydantic class with a description (optional), annotations and field names.
class Demo(BaseModel):
"""
A demo pydantic class to validate the outcome with various nested data types.
"""
demo_1: int
demo_2: float
demo_3: str
demo_4: bool
demo_5: list[str]
demo_6: dict[str, Any]
demo_nest_1: list[dict[str, Any]] # 1 layer of nested child is ok.
demo_nest_2: list[list[str]]
demo_nest_3: dict[str, list[str]]
demo_nest_4: dict[str, dict[str, Any]]
# error_1: list[list[dict[str, list[str]]]] # <- Trigger 400 error due to 2+ layers of nested child.
# error_2: InstanceOf[AnotherPydanticClass] # <- Trigger 400 error due to non-typing annotation.
# error_3: list[InstanceOf[AnotherPydanticClass]] # <- Trigger 400 error due to non-typing annotation as a nested child.
# 2. Define a task
task = vhq.Task(
description="generate random output that strictly follows the given format",
response_schema=Demo,
)
# 3. Execute
res = task.execute()
assert isinstance(res, vhq.TaskOutput)
assert res.raw and res.json
assert isinstance(res.raw, str) and isinstance(res.json_dict, dict)
assert [
getattr(res.pydantic, k) and v.annotation == Demo.model_fields[k].annotation
for k, v in res.pydantic.model_fields.items()
]
2. JSON¶
Similar to Pydantic, JSON output structure can be defined by using a list of ResponseField
objects.
The following code snippet demonstrates how to use ResponseField
to generate output with a maximum of one level of nesting.
Custom JSON outputs can accept one layer of a nested child.
[NOTES]
-
demo_response_fields
in the following case is identical to the previous Demo class, except that titles are specified for nested fields. -
Agents generate JSON output by default, whether or not
response_schema
is given. -
However,
response_schema
is required to specify the key value sets.
import versionhq as vhq
# 1. Define a list of ResponseField objects.
demo_response_fields = [
vhq.ResponseField(title="demo_1", data_type=int),
vhq.ResponseField(title="demo_2", data_type=float),
vhq.ResponseField(title="demo_3", data_type=str),
vhq.ResponseField(title="demo_4", data_type=bool),
vhq.ResponseField(title="demo_5", data_type=list, items=str),
vhq.ResponseField(
title="demo_6",
data_type=dict,
properties=[vhq.ResponseField(title="demo-item", data_type=str)]
),
# nesting
vhq.ResponseField(
title="demo_nest_1",
data_type=list,
items=dict,
properties=([
vhq.ResponseField(
title="nest1",
data_type=dict,
properties=[vhq.ResponseField(title="nest11", data_type=str)]
)
])
),
vhq.ResponseField(title="demo_nest_2", data_type=list, items=list),
vhq.ResponseField(title="demo_nest_3", data_type=dict, properties=[
vhq.ResponseField(title="nest1", data_type=list, items=str)
]),
vhq.ResponseField(title="demo_nest_4", data_type=dict, properties=[
vhq.ResponseField(
title="nest1",
data_type=dict,
properties=[vhq.ResponseField(title="nest12", data_type=str)]
)
])
]
# 2. Define a task
task = vhq.Task(
description="Output random values strictly following the data type defined in the given response format.",
response_schema=demo_response_fields
)
# 3. Execute
res = task.execute()
assert isinstance(res, vhq.TaskOutput)
assert [v and type(v) == task.response_schema[i].data_type for i, (k, v) in enumerate(res.json_dict.items())]
- Ref.
ResponseField
class
Structuring reponse format
-
Higlhy recommends assigning agents optimized for
gemini-x
orgpt-x
to produce structured outputs with nested items. -
To generate response with more than 2 layers of nested items, seperate them into multipe tasks or utilize nodes.
The following case demonstrates to returning a Main
class that contains a nested Sub
class.
[NOTES]
-
Using
callback
functions to format the final response. (You can try other functions suitable for your use case.) -
Passing parameter:
sub
to the callback function via thecallback_kwargs
variable. -
By default, the outputs of
main_task
are automatically passed to the callback function; you do NOT need to explicitly define them. -
Callback results will be stored in the
callback_output
field of theTaskOutput
class.
import versionhq as vhq
from pydantic import BaseModel
from typing import Any
# 1. Defines a sub task
class Sub(BaseModel):
sub1: list[dict[str, Any]]
sub2: dict[str, Any]
sub_task = vhq.Task(
description="generate random values that strictly follows the given format.",
response_schema=Sub
)
sub_res = sub_task.execute()
# 2. Defines a main task with callbacks
class Main(BaseModel):
main1: list[Any] # <= assume expecting to store Sub object.
main2: dict[str, Any]
def format_response(sub, **kwargs) -> Main:
main1 = kwargs["main1"] if kwargs and "main1" in kwargs else None
if main1:
main1.append(sub)
main2 = kwargs["main2"] if kwargs and "main2" in kwargs else None
main = Main(main1=main1, main2=str(main2))
return main
# 3. Executes
main_task = vhq.Task(
description="generate random values that strictly follows the given format.",
response_schema=Main,
callback=format_response,
callback_kwargs=dict(sub=sub_res.json_dict),
)
res = main_task.execute(context=sub_res.raw) # [Optional] Adding sub_task's response as context.
assert res.callback_output.main1 is not None
assert res.callback_output.main2 is not None
To automate these manual setups, refer to AgentNetwork class.