-
Notifications
You must be signed in to change notification settings - Fork 1.3k
Support overriding agent instructions #2926
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Conversation
else: | ||
system_prompts = self._system_prompts | ||
|
||
if override_instructions := self._override_instructions.get(): |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could we move this to some top-level method, like the other override context var usages?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I didn't follow this one -- I think we're doing the same as the existing ones?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What I mean is that we're currently only calling self._override_<foo>.get()
from private getters on Agent
that return either the overridden value, or the original, and then those getters are used in the place where we actually need the value.
So I suggest making _get_instructions_literal_and_functions
a method that doesn't take an argument, but itself checks whether to use the overridden or original value. We can make that easier by merging the self._instructions
and self._instructions_functions
variables into just self._instructions
, so that we don't need to call _get_instructions_literal_and_functions
from __init__
anymore, just here.
And actually, if we move this up above the get_instructions
function, we can reference the same instructions
and instructions_functions
variables inside there, so we only call _instructions_literal_and_functions
once.
if isinstance(instruction, str): | ||
literal_parts.append(instruction) | ||
elif callable(instruction): | ||
func = cast(_system_prompt.SystemPromptFunc[AgentDepsT], instruction) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't love this, but not sure how to appease the type checker. This was marked as "unknown" otherwise.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Does it work if we change elif callable(instruction):
to just else:
, like we had in the original code?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Nope, i think because we have an explicit:
functions: list[_system_prompt.SystemPromptRunner[AgentDepsT]] = []
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@mwildehahn Hmm ok, once this PR is otherwise ready I'll have a look to see if I can clean up the typing here a bit.
if isinstance(instruction, str): | ||
literal_parts.append(instruction) | ||
elif callable(instruction): | ||
func = cast(_system_prompt.SystemPromptFunc[AgentDepsT], instruction) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Does it work if we change elif callable(instruction):
to just else:
, like we had in the original code?
else: | ||
system_prompts = self._system_prompts | ||
|
||
if override_instructions := self._override_instructions.get(): |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What I mean is that we're currently only calling self._override_<foo>.get()
from private getters on Agent
that return either the overridden value, or the original, and then those getters are used in the place where we actually need the value.
So I suggest making _get_instructions_literal_and_functions
a method that doesn't take an argument, but itself checks whether to use the overridden or original value. We can make that easier by merging the self._instructions
and self._instructions_functions
variables into just self._instructions
, so that we don't need to call _get_instructions_literal_and_functions
from __init__
anymore, just here.
And actually, if we move this up above the get_instructions
function, we can reference the same instructions
and instructions_functions
variables inside there, so we only call _instructions_literal_and_functions
once.
dbostest.sqlite
Outdated
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please remove this file, it should've gotten cleaned up automatically 🤔
else: | ||
self._instructions_functions.append(_system_prompt.SystemPromptRunner(instruction)) | ||
self._instructions = self._instructions.strip() or None | ||
self._instructions, self._instructions_functions = self._instructions_literal_and_functions(instructions) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I was thinking we could store self._instructions = instructions
, and then call _get_instructions_literal_and_functions
where we need it. I don't think we need the 2 private vars
self._entered_count = 0 | ||
self._exit_stack = None | ||
|
||
def _get_instructions_literal_and_functions( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Let's just call this _get_instructions
instructions, instructions_functions = self._instructions_literal_and_functions(override_instructions.value) | ||
return instructions, instructions_functions | ||
|
||
def _instructions_literal_and_functions( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
With my suggestion above, we should not need this as a separate method anymore, so we can move its contents into the get method.
if isinstance(instruction, str): | ||
literal_parts.append(instruction) | ||
elif callable(instruction): | ||
func = cast(_system_prompt.SystemPromptFunc[AgentDepsT], instruction) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@mwildehahn Hmm ok, once this PR is otherwise ready I'll have a look to see if I can clean up the typing here a bit.
usage_limits = usage_limits or _usage.UsageLimits() | ||
|
||
async def get_instructions(run_context: RunContext[AgentDepsT]) -> str | None: | ||
literal, functions = self._get_instructions_literal_and_functions() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
With the changes I suggested above, this can be:
literal, functions = self._get_instructions_literal_and_functions() | |
instructions, instructions_functions = self._get_instructions() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
And can we move this out of the function, and then use the same instructions
and instructions_functions
when we build the UserPromptNode
below, to save calling this method twice?
@DouweM i think i've addressed everything and fixed the weird type errors -- issue was I wasn't passing generic with Instruction alias |
self._entered_count = 0 | ||
self._exit_stack = None | ||
|
||
def _get_instructions( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Minor thing, but please move this to where the other _get_
methods are -- it's not important enough to be right at the top of the class :)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please move the relevant tests to tests/test_agent.py
, where the other overrides are tested
def test_first_request_skips_non_requests(): | ||
"""Helper ignores non-request messages until it finds a request.""" | ||
response = ModelResponse(parts=()) | ||
request = ModelRequest(parts=()) | ||
assert _first_request([response, request]) is request | ||
|
||
|
||
def test_first_request_raises_without_model_request(): | ||
"""Helper raises when no model request is present.""" | ||
response = ModelResponse(parts=()) | ||
with pytest.raises(AssertionError, match='no ModelRequest found'): | ||
_first_request([response]) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
No need to test the helpers :)
def _first_request(messages: list[ModelMessage]) -> ModelRequest: | ||
"""Helper to extract the first ModelRequest from captured messages.""" | ||
assert messages, 'no messages captured' | ||
for m in messages: | ||
if isinstance(m, ModelRequest): | ||
return m | ||
raise AssertionError('no ModelRequest found in captured messages') | ||
|
||
|
||
def _system_prompt_texts(parts: Sequence[ModelRequestPart]) -> list[str]: | ||
"""Helper to extract system prompt text content from message parts.""" | ||
return [p.content for p in parts if isinstance(p, SystemPromptPart)] |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do we really need these helpers? I'd rather just repeat this line, and use assert isinstance(messages[0], ModelRequest)
to ensure the first message is a request.
elif callable(self._instructions): | ||
instructions_list = [self._instructions, instruction] | ||
else: | ||
instructions_list = [*self._instructions, instruction] # pragma: no cover |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sorry for continuing to come up with more ways to refactor this, but what do you think about making self._instructions
and self._override_instructions
always hold a list[str | _system_prompt.SystemPromptFunc[AgentDepsT]]
, and change __init__
and override
to store list(instructions)
so that any single items automatically get wrapped in a list, and this method and the one above can be much simpler?
I just started working on a library: https://github.com/mwildehahn/pydantic-ai-gepa to integrate https://github.com/gepa-ai/gepa into pydantic-ai. This would provide similar functionality to https://dspy.ai/ where you can provide a signature and then let an LLM handle constructing the prompt.
This is very experimental and I just started on this yesterday, but I wanted to propose a small extension to pydantic-ai that would allow us to override system_prompt and instructions in the same way we can override toolsets etc. With that minimal surface area, I can hook in something like pydantic-ai-gepa to optimize the prompts and then run those optimized prompts.
LMK if there are better ways to override the system prompts on demand.