Returning pydantic result of tool as final_result

[bbkgh]

Hi. There is a simple agent code that i suppose to return result of my_func (list[ProductsResponse]) as final result but it fails to execute this task (the model is probably powerful enough). Is there any mechanism for fixing this issue?
sth like set some tools as a final tools that their result returns as response of agent.run and execution stops at them.
or setting sth in ctx for finishing flow (e.g ctx.break = True)

also making some tools as final results to better performance (reduce one llm call for parsing output of them separately)

from dataclasses import dataclass

import logfire
from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext
from pydantic_ai.models.openai import OpenAIModel

logfire.configure(send_to_logfire="if-token-present")

@dataclass
class Deps:
    url: str


class ProductsResponse(BaseModel):
    """Use this for result of search"""

    handle: str = Field(description="handle of product")
    title: str = Field(description="title of product")
    image_url: str = Field(description="image_url of product")


_model = OpenAIModel(model_name="llama3.3", base_url="https://ollama.mysite.com/v1")

new_agent = Agent(
    _model,
    result_type=list[ProductsResponse],  # type: ignore
    deps_type=Deps,
    model_settings={"temperature": 0.0},
    retries=1,
)


@new_agent.system_prompt
async def make_general_system_prompts(ctx: RunContext[Deps]) -> str:
    result = """Always use my_func tool"""
    return result


@new_agent.tool
async def my_func(ctx: RunContext[Deps]) -> list[ProductsResponse]:
    return [ProductsResponse(title="test", handle="test", image_url="test")]

if __name__ == "__main__":
    deps = Deps(url="google.com")
    result = new_agent.run_sync("hi", deps=deps)
    print(result.data)
    print(result.new_messages_json().decode("utf8"))
    # {"message_id":}

[aristideubertas]

Also looking into how to return tool output without having the llm ingest it in its context and regurgitate it.

[HamzaFarhan]

Instead of:

@new_agent.tool
async def my_func(ctx: RunContext[Deps]) -> list[ProductsResponse]:
    return [ProductsResponse(title="test", handle="test", image_url="test")]

You could do:

# result_validator always gets called in the end but it needs to receive an argument of type `result_type`
@new_agent.result_validator
async def my_func(ctx: RunContext[Deps], result: list[ProductsResponse]) -> list[ProductsResponse]:
    return [ProductsResponse(title="test", handle="test", image_url="test")]

sth like set some tools as a final tools that their result returns as response of agent.run and execution stops at them. or setting sth in ctx for finishing flow (e.g ctx.break = True)

You can do this by returning End from a graph Node

[bbkgh]

What do you mean by the Instead of tool? I hardcoded it to simplify the issue. Do you mean I should add a new result_validator alongside the tool that returns it's input? Another issue is that my actual agent response type is list[...] | str.
Also, I think that using Graph is over-engineering for my project; it makes my code harder to understand and somehow locks me in.

[HamzaFarhan]

I mean whatever you return from the result_validator would be the exact result of the run. So you could leverage that somehow.