agent/agent-framework
1
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: [BREAKING] Make response_format validation errors visible to users (#3274)

Browse Source 
* Make response_format validation errors visible to users

* Small fix

* Addressed comments
This commit is contained in:
Dmytro Struk
2026-01-19 09:20:46 -08:00
committed by GitHub
Unverified
parent 3c1be2a713
commit 83e8965c8e
12 changed files with 298 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/core/agent_framework/_types.py
+124 -24
View File
@@ -2861,14 +2861,13 @@ class ChatResponse(SerializationMixin):
self.created_at = created_at
self.finish_reason = finish_reason
self.usage_details = usage_details
self.value = value
self._value: Any | None = value
self._response_format: type[BaseModel] | None = response_format
self._value_parsed: bool = value is not None
self.additional_properties = additional_properties or {}
self.additional_properties.update(kwargs or {})
self.raw_representation: Any | list[Any] | None = raw_representation
if response_format:
self.try_parse_value(output_format_type=response_format)
@classmethod
def from_chat_response_updates(
cls: type[TChatResponse],
@@ -2933,12 +2932,13 @@ class ChatResponse(SerializationMixin):
Keyword Args:
output_format_type: Optional Pydantic model type to parse the response text into structured data.
"""
msg = cls(messages=[])
response_format = output_format_type if isinstance(output_format_type, type) else None
msg = cls(messages=[], response_format=response_format)
async for update in updates:
_process_update(msg, update)
_finalize_response(msg)
if output_format_type and isinstance(output_format_type, type) and issubclass(output_format_type, BaseModel):
msg.try_parse_value(output_format_type)
if response_format and issubclass(response_format, BaseModel):
msg.try_parse_value(response_format)
return msg
@property
@@ -2946,16 +2946,64 @@ class ChatResponse(SerializationMixin):
"""Returns the concatenated text of all messages in the response."""
return ("\n".join(message.text for message in self.messages if isinstance(message, ChatMessage))).strip()
@property
def value(self) -> Any | None:
"""Get the parsed structured output value.
If a response_format was provided and parsing hasn't been attempted yet,
this will attempt to parse the text into the specified type.
Raises:
ValidationError: If the response text doesn't match the expected schema.
"""
if self._value_parsed:
return self._value
if (
self._response_format is not None
and isinstance(self._response_format, type)
and issubclass(self._response_format, BaseModel)
):
self._value = self._response_format.model_validate_json(self.text)
self._value_parsed = True
return self._value
def __str__(self) -> str:
return self.text
def try_parse_value(self, output_format_type: type[BaseModel]) -> None:
"""If there is a value, does nothing, otherwise tries to parse the text into the value."""
if self.value is None and isinstance(output_format_type, type) and issubclass(output_format_type, BaseModel):
try:
self.value = output_format_type.model_validate_json(self.text) # type: ignore[reportUnknownMemberType]
except ValidationError as ex:
logger.debug("Failed to parse value from chat response text: %s", ex)
def try_parse_value(self, output_format_type: type[_T] | None = None) -> _T | None:
"""Try to parse the text into a typed value.
This is the safe alternative to accessing the value property directly.
Returns the parsed value on success, or None on failure.
Args:
output_format_type: The Pydantic model type to parse into.
If None, uses the response_format from initialization.
Returns:
The parsed value as the specified type, or None if parsing fails.
"""
format_type = output_format_type or self._response_format
if format_type is None or not (isinstance(format_type, type) and issubclass(format_type, BaseModel)):
return None
# Cache the result unless a different schema than the configured response_format is requested.
# This prevents calls with a different schema from polluting the cached value.
use_cache = (
self._response_format is None or output_format_type is None or output_format_type is self._response_format
)
if use_cache and self._value_parsed and self._value is not None:
return self._value # type: ignore[return-value, no-any-return]
try:
parsed_value = format_type.model_validate_json(self.text) # type: ignore[reportUnknownMemberType]
if use_cache:
self._value = parsed_value
self._value_parsed = True
return parsed_value # type: ignore[return-value]
except ValidationError as ex:
logger.warning("Failed to parse value from chat response text: %s", ex)
return None
# region ChatResponseUpdate
@@ -3141,6 +3189,7 @@ class AgentResponse(SerializationMixin):
created_at: CreatedAtT | None = None,
usage_details: UsageDetails | MutableMapping[str, Any] | None = None,
value: Any | None = None,
response_format: type[BaseModel] | None = None,
raw_representation: Any | None = None,
additional_properties: dict[str, Any] | None = None,
**kwargs: Any,
@@ -3153,6 +3202,7 @@ class AgentResponse(SerializationMixin):
created_at: A timestamp for the chat response.
usage_details: The usage details for the chat response.
value: The structured output of the agent run response, if applicable.
response_format: Optional response format for the agent response.
additional_properties: Any additional properties associated with the chat response.
raw_representation: The raw representation of the chat response from an underlying implementation.
**kwargs: Additional properties to set on the response.
@@ -3180,7 +3230,9 @@ class AgentResponse(SerializationMixin):
self.response_id = response_id
self.created_at = created_at
self.usage_details = usage_details
self.value = value
self._value: Any | None = value
self._response_format: type[BaseModel] | None = response_format
self._value_parsed: bool = value is not None
self.additional_properties = additional_properties or {}
self.additional_properties.update(kwargs or {})
self.raw_representation = raw_representation
@@ -3190,6 +3242,27 @@ class AgentResponse(SerializationMixin):
"""Get the concatenated text of all messages."""
return "".join(msg.text for msg in self.messages) if self.messages else ""
@property
def value(self) -> Any | None:
"""Get the parsed structured output value.
If a response_format was provided and parsing hasn't been attempted yet,
this will attempt to parse the text into the specified type.
Raises:
ValidationError: If the response text doesn't match the expected schema.
"""
if self._value_parsed:
return self._value
if (
self._response_format is not None
and isinstance(self._response_format, type)
and issubclass(self._response_format, BaseModel)
):
self._value = self._response_format.model_validate_json(self.text)
self._value_parsed = True
return self._value
@property
def user_input_requests(self) -> list[UserInputRequestContents]:
"""Get all BaseUserInputRequest messages from the response."""
@@ -3215,7 +3288,7 @@ class AgentResponse(SerializationMixin):
Keyword Args:
output_format_type: Optional Pydantic model type to parse the response text into structured data.
"""
msg = cls(messages=[])
msg = cls(messages=[], response_format=output_format_type)
for update in updates:
_process_update(msg, update)
_finalize_response(msg)
@@ -3238,7 +3311,7 @@ class AgentResponse(SerializationMixin):
Keyword Args:
output_format_type: Optional Pydantic model type to parse the response text into structured data
"""
msg = cls(messages=[])
msg = cls(messages=[], response_format=output_format_type)
async for update in updates:
_process_update(msg, update)
_finalize_response(msg)
@@ -3249,13 +3322,40 @@ class AgentResponse(SerializationMixin):
def __str__(self) -> str:
return self.text
def try_parse_value(self, output_format_type: type[BaseModel]) -> None:
"""If there is a value, does nothing, otherwise tries to parse the text into the value."""
if self.value is None:
try:
self.value = output_format_type.model_validate_json(self.text) # type: ignore[reportUnknownMemberType]
except ValidationError as ex:
logger.debug("Failed to parse value from agent run response text: %s", ex)
def try_parse_value(self, output_format_type: type[_T] | None = None) -> _T | None:
"""Try to parse the text into a typed value.
This is the safe alternative when you need to parse the response text into a typed value.
Returns the parsed value on success, or None on failure.
Args:
output_format_type: The Pydantic model type to parse into.
If None, uses the response_format from initialization.
Returns:
The parsed value as the specified type, or None if parsing fails.
"""
format_type = output_format_type or self._response_format
if format_type is None or not (isinstance(format_type, type) and issubclass(format_type, BaseModel)):
return None
# Cache the result unless a different schema than the configured response_format is requested.
# This prevents calls with a different schema from polluting the cached value.
use_cache = (
self._response_format is None or output_format_type is None or output_format_type is self._response_format
)
if use_cache and self._value_parsed and self._value is not None:
return self._value # type: ignore[return-value, no-any-return]
try:
parsed_value = format_type.model_validate_json(self.text) # type: ignore[reportUnknownMemberType]
if use_cache:
self._value = parsed_value
self._value_parsed = True
return parsed_value # type: ignore[return-value]
except ValidationError as ex:
logger.warning("Failed to parse value from agent run response text: %s", ex)
return None
# region AgentResponseUpdate
python/packages/core/tests/core/test_types.py
+118
View File
@@ -705,6 +705,124 @@ def test_chat_response_with_format_init():
assert response.value.response == "Hello"
def test_chat_response_value_raises_on_invalid_schema():
"""Test that value property raises ValidationError with field constraint details."""
from typing import Literal
from pydantic import Field, ValidationError
class StrictSchema(BaseModel):
id: Literal[5]
name: str = Field(min_length=10)
score: int = Field(gt=0, le=100)
message = ChatMessage(role="assistant", text='{"id": 1, "name": "test", "score": -5}')
response = ChatResponse(messages=message, response_format=StrictSchema)
with raises(ValidationError) as exc_info:
_ = response.value
errors = exc_info.value.errors()
error_fields = {e["loc"][0] for e in errors}
assert "id" in error_fields, "Expected 'id' Literal constraint error"
assert "name" in error_fields, "Expected 'name' min_length constraint error"
assert "score" in error_fields, "Expected 'score' gt constraint error"
def test_chat_response_try_parse_value_returns_none_on_invalid():
"""Test that try_parse_value returns None on validation failure with Field constraints."""
from typing import Literal
from pydantic import Field
class StrictSchema(BaseModel):
id: Literal[5]
name: str = Field(min_length=10)
score: int = Field(gt=0, le=100)
message = ChatMessage(role="assistant", text='{"id": 1, "name": "test", "score": -5}')
response = ChatResponse(messages=message)
result = response.try_parse_value(StrictSchema)
assert result is None
def test_chat_response_try_parse_value_returns_value_on_success():
"""Test that try_parse_value returns parsed value when all constraints pass."""
from pydantic import Field
class MySchema(BaseModel):
name: str = Field(min_length=3)
score: int = Field(ge=0, le=100)
message = ChatMessage(role="assistant", text='{"name": "test", "score": 85}')
response = ChatResponse(messages=message)
result = response.try_parse_value(MySchema)
assert result is not None
assert result.name == "test"
assert result.score == 85
def test_agent_response_value_raises_on_invalid_schema():
"""Test that AgentResponse.value property raises ValidationError with field constraint details."""
from typing import Literal
from pydantic import Field, ValidationError
class StrictSchema(BaseModel):
id: Literal[5]
name: str = Field(min_length=10)
score: int = Field(gt=0, le=100)
message = ChatMessage(role="assistant", text='{"id": 1, "name": "test", "score": -5}')
response = AgentResponse(messages=message, response_format=StrictSchema)
with raises(ValidationError) as exc_info:
_ = response.value
errors = exc_info.value.errors()
error_fields = {e["loc"][0] for e in errors}
assert "id" in error_fields, "Expected 'id' Literal constraint error"
assert "name" in error_fields, "Expected 'name' min_length constraint error"
assert "score" in error_fields, "Expected 'score' gt constraint error"
def test_agent_response_try_parse_value_returns_none_on_invalid():
"""Test that AgentResponse.try_parse_value returns None on Field constraint failure."""
from typing import Literal
from pydantic import Field
class StrictSchema(BaseModel):
id: Literal[5]
name: str = Field(min_length=10)
score: int = Field(gt=0, le=100)
message = ChatMessage(role="assistant", text='{"id": 1, "name": "test", "score": -5}')
response = AgentResponse(messages=message)
result = response.try_parse_value(StrictSchema)
assert result is None
def test_agent_response_try_parse_value_returns_value_on_success():
"""Test that AgentResponse.try_parse_value returns parsed value when all constraints pass."""
from pydantic import Field
class MySchema(BaseModel):
name: str = Field(min_length=3)
score: int = Field(ge=0, le=100)
message = ChatMessage(role="assistant", text='{"name": "test", "score": 85}')
response = AgentResponse(messages=message)
result = response.try_parse_value(MySchema)
assert result is not None
assert result.name == "test"
assert result.score == 85
# region ChatResponseUpdate
python/samples/getting_started/agents/azure_ai/azure_ai_with_response_format.py
+3 -2
View File
@@ -41,12 +41,13 @@ async def main() -> None:
print(f"User: {query}")
result = await agent.run(query)
if isinstance(result.value, ReleaseBrief):
release_brief = result.value
if release_brief := result.try_parse_value(ReleaseBrief):
print("Agent:")
print(f"Feature: {release_brief.feature}")
print(f"Benefit: {release_brief.benefit}")
print(f"Launch date: {release_brief.launch_date}")
else:
print(f"Failed to parse response: {result.text}")
if __name__ == "__main__":
python/samples/getting_started/agents/azure_ai_agent/azure_ai_with_response_format.py
+6 -4
View File
@@ -56,13 +56,14 @@ async def main() -> None:
result1 = await agent.run(query1)
if isinstance(result1.value, WeatherInfo):
weather = result1.value
if weather := result1.try_parse_value(WeatherInfo):
print("Agent:")
print(f" Location: {weather.location}")
print(f" Temperature: {weather.temperature}")
print(f" Conditions: {weather.conditions}")
print(f" Recommendation: {weather.recommendation}")
else:
print(f"Failed to parse response: {result1.text}")
# Request 2: Override response_format at runtime with CityInfo
print("\n--- Request 2: Runtime override with CityInfo ---")
@@ -71,12 +72,13 @@ async def main() -> None:
result2 = await agent.run(query2, options={"response_format": CityInfo})
if isinstance(result2.value, CityInfo):
city = result2.value
if city := result2.try_parse_value(CityInfo):
print("Agent:")
print(f" City: {city.city_name}")
print(f" Population: {city.population}")
print(f" Country: {city.country}")
else:
print(f"Failed to parse response: {result2.text}")
if __name__ == "__main__":
python/samples/getting_started/agents/openai/openai_assistants_with_response_format.py
+6 -4
View File
@@ -59,13 +59,14 @@ async def main() -> None:
result1 = await agent.run(query1)
if isinstance(result1.value, WeatherInfo):
weather = result1.value
if weather := result1.try_parse_value(WeatherInfo):
print("Agent:")
print(f" Location: {weather.location}")
print(f" Temperature: {weather.temperature}")
print(f" Conditions: {weather.conditions}")
print(f" Recommendation: {weather.recommendation}")
else:
print(f"Failed to parse response: {result1.text}")
# Request 2: Override response_format at runtime with CityInfo
print("\n--- Request 2: Runtime override with CityInfo ---")
@@ -74,12 +75,13 @@ async def main() -> None:
result2 = await agent.run(query2, options={"response_format": CityInfo})
if isinstance(result2.value, CityInfo):
city = result2.value
if city := result2.try_parse_value(CityInfo):
print("Agent:")
print(f" City: {city.city_name}")
print(f" Population: {city.population}")
print(f" Country: {city.country}")
else:
print(f"Failed to parse response: {result2.text}")
finally:
await client.beta.assistants.delete(agent.id)
python/samples/getting_started/agents/openai/openai_responses_client_with_structured_output.py
+7 -9
View File
@@ -37,14 +37,13 @@ async def non_streaming_example() -> None:
# Get structured response from the agent using response_format parameter
result = await agent.run(query, options={"response_format": OutputStruct})
# Access the structured output directly from the response value
if result.value:
structured_data: OutputStruct = result.value # type: ignore
print("Structured Output Agent (from result.value):")
# Access the structured output using try_parse_value for safe parsing
if structured_data := result.try_parse_value(OutputStruct):
print("Structured Output Agent (from result.try_parse_value):")
print(f"City: {structured_data.city}")
print(f"Description: {structured_data.description}")
else:
print("Error: No structured data found in result.value")
print(f"Failed to parse response: {result.text}")
async def streaming_example() -> None:
@@ -67,14 +66,13 @@ async def streaming_example() -> None:
output_format_type=OutputStruct,
)
# Access the structured output directly from the response value
if result.value:
structured_data: OutputStruct = result.value # type: ignore
# Access the structured output using try_parse_value for safe parsing
if structured_data := result.try_parse_value(OutputStruct):
print("Structured Output (from streaming with AgentResponse.from_agent_response_generator):")
print(f"City: {structured_data.city}")
print(f"Description: {structured_data.description}")
else:
print("Error: No structured data found in result.value")
print(f"Failed to parse response: {result.text}")
async def main() -> None:
python/samples/getting_started/azure_functions/06_multi_agent_orchestration_conditionals/function_app.py
+7 -3
View File
@@ -12,7 +12,7 @@ Functions host."""
import json
import logging
from collections.abc import Mapping
from typing import Any, cast
from typing import Any
import azure.functions as func
from agent_framework.azure import AgentFunctionApp, AzureOpenAIChatClient
@@ -102,7 +102,9 @@ def spam_detection_orchestration(context: DurableOrchestrationContext):
options={"response_format": SpamDetectionResult},
)
spam_result = cast(SpamDetectionResult, spam_result_raw.value)
spam_result = spam_result_raw.try_parse_value(SpamDetectionResult)
if spam_result is None:
raise ValueError("Failed to parse spam detection result")
if spam_result.is_spam:
result = yield context.call_activity("handle_spam_email", spam_result.reason)
@@ -123,7 +125,9 @@ def spam_detection_orchestration(context: DurableOrchestrationContext):
options={"response_format": EmailResponse},
)
email_result = cast(EmailResponse, email_result_raw.value)
email_result = email_result_raw.try_parse_value(EmailResponse)
if email_result is None:
raise ValueError("Failed to parse email response")
result = yield context.call_activity("send_email", email_result.response)
return result
python/samples/getting_started/azure_functions/07_single_agent_orchestration_hitl/function_app.py
+4 -6
View File
@@ -101,10 +101,10 @@ def content_generation_hitl_orchestration(context: DurableOrchestrationContext):
options={"response_format": GeneratedContent},
)
content = initial_raw.value
content = initial_raw.try_parse_value(GeneratedContent)
logger.info("Type of content after extraction: %s", type(content))
if content is None or not isinstance(content, GeneratedContent):
if content is None:
raise ValueError("Agent returned no content after extraction.")
attempt = 0
@@ -146,11 +146,9 @@ def content_generation_hitl_orchestration(context: DurableOrchestrationContext):
options={"response_format": GeneratedContent},
)
rewritten_value = rewritten_raw.value
if rewritten_value is None or not isinstance(rewritten_value, GeneratedContent):
content = rewritten_raw.try_parse_value(GeneratedContent)
if content is None:
raise ValueError("Agent returned no content after rewrite.")
content = rewritten_value
else:
context.set_custom_status(
f"Human approval timed out after {payload.approval_timeout_hours} hour(s). Treating as rejection."
python/samples/getting_started/chat_client/azure_responses_client.py
+8 -3
View File
@@ -44,11 +44,16 @@ async def main() -> None:
client.get_streaming_response(message, tools=get_weather, options={"response_format": OutputStruct}),
output_format_type=OutputStruct,
)
print(f"Assistant: {response.value}")
if result := response.try_parse_value(OutputStruct):
print(f"Assistant: {result}")
else:
print(f"Assistant: {response.text}")
else:
response = await client.get_response(message, tools=get_weather, options={"response_format": OutputStruct})
print(f"Assistant: {response.value}")
if result := response.try_parse_value(OutputStruct):
print(f"Assistant: {result}")
else:
print(f"Assistant: {response.text}")
if __name__ == "__main__":
python/samples/getting_started/context_providers/simple_context_provider.py
+5 -5
View File
@@ -52,11 +52,11 @@ class UserInfoMemory(ContextProvider):
)
# Update user info with extracted data
if result.value and isinstance(result.value, UserInfo):
if self.user_info.name is None and result.value.name:
self.user_info.name = result.value.name
if self.user_info.age is None and result.value.age:
self.user_info.age = result.value.age
if extracted := result.try_parse_value(UserInfo):
if self.user_info.name is None and extracted.name:
self.user_info.name = extracted.name
if self.user_info.age is None and extracted.age:
self.user_info.age = extracted.age
except Exception:
pass # Failed to extract, continue without updating
python/samples/getting_started/declarative/azure_openai_responses_agent.py
+5 -1
View File
@@ -20,7 +20,11 @@ async def main():
agent = AgentFactory(client_kwargs={"credential": AzureCliCredential()}).create_agent_from_yaml(yaml_str)
# use the agent
response = await agent.run("Why is the sky blue, answer in Dutch?")
print("Agent response:", response.value.model_dump_json(indent=2))
# Use try_parse_value() for safe parsing - returns None if no response_format or parsing fails
if parsed := response.try_parse_value():
print("Agent response:", parsed.model_dump_json(indent=2))
else:
print("Agent response:", response.text)
if __name__ == "__main__":
python/samples/getting_started/declarative/openai_responses_agent.py
+5 -1
View File
@@ -19,7 +19,11 @@ async def main():
agent = AgentFactory().create_agent_from_yaml(yaml_str)
# use the agent
response = await agent.run("Why is the sky blue, answer in Dutch?")
print("Agent response:", response.value)
# Use try_parse_value() for safe parsing - returns None if no response_format or parsing fails
if parsed := response.try_parse_value():
print("Agent response:", parsed)
else:
print("Agent response:", response.text)
if __name__ == "__main__":