Name | pydantic-ai-helpers JSON |
Version |
0.0.2
JSON |
| download |
home_page | None |
Summary | Boring, opinionated helpers for PydanticAI that are so simple you didn't want to even vibe code them. (Unofficial) |
upload_time | 2025-08-17 20:35:13 |
maintainer | None |
docs_url | None |
author | Jan Siml |
requires_python | >=3.10 |
license | None |
keywords |
pydantic
ai
utilities
history
llm
|
VCS |
 |
bugtrack_url |
|
requirements |
No requirements were recorded.
|
Travis-CI |
No Travis.
|
coveralls test coverage |
No coveralls.
|
# pydantic-ai-helpers
> Boring, opinionated helpers for PydanticAI that are so dumb you didn't want to implement them. So I did.
**⚠️ This is NOT an official PydanticAI package** - just a simple personal helper library.
[](https://pypi.org/project/pydantic-ai-helpers/)
[](https://pypi.org/project/pydantic-ai-helpers/)
[](https://github.com/svilupp/pydantic-ai-helpers/actions)
[](https://codecov.io/gh/svilupp/pydantic-ai-helpers)
[](https://github.com/svilupp/pydantic-ai-helpers/blob/main/LICENSE)
## The Problem
[PydanticAI](https://github.com/pydantic/pydantic-ai) is amazing! But at some point you'll need to quickly and easily extract aspects of your conversations. It's not hard but it's a pain to do, because neither you nor the LLMS know how to do it, so you'll waste 10+ minutes to do:
```python
# Want the last tool call for your UI updates?
last_tool_call = None
for message in result.all_messages():
for part in message.parts:
if isinstance(part, ToolCallPart):
last_tool_call = part
# Need that metadata you passed for evaluations?
metadata_parts = []
for message in result.all_messages():
for part in message.parts:
if isinstance(part, ToolReturnPart) and part.metadata:
metadata_parts.append(part.metadata)
# How about just the user's question again?
user_question = None
for message in result.all_messages():
for part in message.parts:
if isinstance(part, UserPromptPart):
user_question = part.content
break
```
We've all been there. **We've got you!**
```python
from pydantic_ai_helpers import History
# or for convenience:
import pydantic_ai_helpers as ph
hist = History(result) # or ph.History(result)
last_tool_call = hist.tools.calls().last() # Done
metadata = hist.tools.returns().last().metadata # Easy
user_question = hist.user.last().content # Simple
system_prompt = hist.system_prompt() # Get system message
media_items = hist.media.images() # Extract media content
```
The best part? Your IDE will help you with the suggestions for the available methods so you don't have to remember anything!
## Installation
```bash
uv add pydantic-ai-helpers
# pip install pydantic-ai-helpers
# poetry add pydantic-ai-helpers
```
## Quick Start
```python
from pydantic_ai import Agent
from pydantic_ai_helpers import History
# or: import pydantic_ai_helpers as ph
agent = Agent("openai:gpt-4.1-mini")
result = agent.run_sync("Tell me a joke")
# Wrap once, access everything
hist = History(result) # or ph.History(result)
# Get the first and last user messages
print(hist.user.first().content) # First user message
print(hist.user.last().content) # Last user message
# Output: "Tell me a joke"
# Get all AI responses
for response in hist.ai.all():
print(response.content)
# Check token usage
print(f"Tokens used: {hist.usage().total_tokens}")
# Access system prompt (if any)
if system_prompt := hist.system_prompt():
print(f"System prompt: {system_prompt.content}")
# Access media content
images = hist.media.images()
if images:
print(f"Found {len(images)} images in conversation")
```
## Common Use Cases
### Extract What You Need for Your App
```python
hist = History(result)
# Update your UI with the latest tool status
if latest_call := hist.tools.calls().last():
update_ui_status(f"Called {latest_call.tool_name}...")
# Get conversation context for logging
user_query = hist.user.last().content
ai_response = hist.ai.last().content
log_conversation(user_query, ai_response)
# Check token costs for billing
total_cost = hist.usage().total_tokens * your_token_rate
```
### Debug Tool Workflows
```python
# See what tools were actually called
for call in hist.tools.calls().all():
print(f"Called {call.tool_name} with {call.args}")
# Check what came back
for ret in hist.tools.returns().all():
print(f"{ret.tool_name} returned: {ret.content}")
if ret.metadata: # Your evaluation metadata
print(f"Metadata: {ret.metadata}")
```
### Analyze Conversations
```python
# Count interactions
print(f"User asked {len(hist.user.all())} questions")
print(f"AI made {len(hist.tools.calls().all())} tool calls")
print(f"Total tokens: {hist.usage().total_tokens}")
# Get specific tool results for processing
weather_results = hist.tools.returns(name="get_weather").all()
for result in weather_results:
process_weather_data(result.content)
```
### Work with Media Content
```python
# Access all media content
all_media = hist.media.all()
print(f"Found {len(all_media)} media items")
# Get specific media types
images = hist.media.images() # All images (URLs + binary)
audio = hist.media.audio() # All audio files
documents = hist.media.documents() # All documents
videos = hist.media.videos() # All videos
# Filter by storage type
url_images = hist.media.images(url_only=True) # Only ImageUrl objects
binary_images = hist.media.images(binary_only=True) # Only binary images
# Get the most recent media
latest_media = hist.media.last()
if latest_media:
print(f"Latest media: {type(latest_media).__name__}")
# Filter by exact type
from pydantic_ai.messages import ImageUrl, BinaryContent
image_urls = hist.media.by_type(ImageUrl)
binary_content = hist.media.by_type(BinaryContent)
```
### Access System Prompts
```python
# Get the system prompt (if any)
system_prompt = hist.system_prompt()
if system_prompt:
print(f"System prompt: {system_prompt.content}")
else:
print("No system prompt found")
# Use in analysis
if system_prompt and "helpful" in system_prompt.content:
print("This agent was configured to be helpful")
```
## Examples
### Multi-turn Conversation Analysis
```python
messages = []
topics = [
"What's the weather in London?",
"How about Paris?",
"Which city is warmer?"
]
for topic in topics:
result = agent.run_sync(topic, message_history=messages)
messages = result.all_messages()
hist = History(result)
# Analyze the conversation flow
print(f"User asked {len(hist.user.all())} questions")
print(f"AI responded {len(hist.ai.all())} times")
print(f"Made {len(hist.tools.calls())} tool calls")
# Get specific information
london_weather = hist.tools.returns(name="get_weather").all()[0]
paris_weather = hist.tools.returns(name="get_weather").all()[1]
```
### Dice Game with Tools
```python
# From the PydanticAI tutorial
result = agent.run_sync("Roll a dice")
hist = History(result)
# Find what the dice rolled
dice_result = hist.tools.returns(name="roll_dice").last()
print(f"Dice rolled: {dice_result.content}")
# See how the AI responded
ai_message = hist.ai.last()
print(f"AI said: {ai_message.content}")
```
### Streaming Support
```python
async with agent.run_stream("Tell me a story") as result:
async for chunk in result.stream():
print(chunk, end="")
# After streaming completes
hist = History(result)
print(f"\nTotal tokens: {hist.tokens().total_tokens}")
```
### Loading from Serialized Conversations
```python
import json
from pydantic_core import to_jsonable_python
from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessagesTypeAdapter
# Save a conversation
agent = Agent('openai:gpt-4.1-mini')
result = agent.run_sync('Tell me a joke.')
messages = result.all_messages()
# Serialize to file
with open('conversation.json', 'w') as f:
json.dump(to_jsonable_python(messages), f)
# Later, load it back
hist = History('conversation.json')
print(hist) # History(1 turn, 50 tokens)
print(hist.user.last().content) # "Tell me a joke."
print(hist.ai.last().content) # The joke response
# Or use Path objects
from pathlib import Path
hist = History(Path('conversation.json'))
# Continue the conversation with loaded history
same_messages = ModelMessagesTypeAdapter.validate_python(
to_jsonable_python(hist.all_messages())
)
result2 = agent.run_sync(
'Tell me a different joke.',
message_history=same_messages
)
```
## Evals Helpers
You can compare values and collections with simple, reusable comparators, or use small evaluator classes to compare fields by dotted paths. **Now with fuzzy string matching support!**
### Quick Comparators
```python
from pydantic_ai_helpers.evals import ScalarCompare, ListCompare, InclusionCompare
# Scalars with coercion and tolerance
comp = ScalarCompare(coerce_to="float", abs_tol=0.01)
score, why = comp("3.14", 3.13) # -> (1.0, 'numbers match')
# Lists with recall/precision or equality
recall = ListCompare(mode="recall")
score, why = recall(["a", "b"], ["a", "b", "c"]) # -> ~0.667
equality = ListCompare(mode="equality", order_sensitive=False)
score, _ = equality(["a", "b"], ["b", "a"]) # -> 1.0
# Value in acceptable list with fuzzy matching (NEW!)
inc = InclusionCompare() # Uses defaults: normalization + fuzzy matching
score, _ = inc("aple", ["apple", "banana", "cherry"]) # -> ~0.9 (fuzzy match)
```
### Fuzzy String Matching (NEW!)
The library now includes powerful fuzzy string matching using rapidfuzz:
```python
from pydantic_ai_helpers.evals import ScalarCompare, CompareOptions, FuzzyOptions
# Default behavior: fuzzy matching enabled with 0.85 threshold
comp = ScalarCompare()
score, why = comp("colour", "color") # -> (0.91, 'fuzzy match (score=0.91)')
# Exact matching (disable fuzzy)
comp = ScalarCompare(fuzzy_enabled=False)
score, why = comp("colour", "color") # -> (0.0, 'values differ...')
# Custom fuzzy settings
comp = ScalarCompare(
fuzzy_threshold=0.9, # Stricter threshold
fuzzy_algorithm="ratio", # Different algorithm
normalize_lowercase=True # Case insensitive
)
# For lists with fuzzy matching
from pydantic_ai_helpers.evals import ListRecall
evaluator = ListRecall() # Fuzzy enabled by default
score, why = evaluator(
["Python", "AI", "Machine Learning"], # Output
["python", "ai", "data science", "ml"] # Expected
)
# Uses fuzzy scores: "Machine Learning" partially matches "ml"
```
### Field-to-Field Evaluators
Use evaluators when you want to compare fields inside nested objects using dotted paths:
```python
from pydantic_ai_helpers.evals import ScalarEquals, ListRecall, ListEquality, ValueInExpectedList
from pydantic_evals.evaluators import EvaluatorContext
# Basic usage (fuzzy enabled by default)
evaluator = ScalarEquals(
output_path="user.name",
expected_path="user.name",
evaluation_name="name_match",
)
# Custom fuzzy settings for stricter matching
evaluator = ScalarEquals(
output_path="predicted.category",
expected_path="actual.category",
fuzzy_threshold=0.95, # Very strict
normalize_alphanum=True, # Remove punctuation
evaluation_name="category_match",
)
# List evaluation with fuzzy matching
list_evaluator = ListRecall(
output_path="predicted_tags",
expected_path="required_tags",
fuzzy_enabled=True, # Default: True
fuzzy_threshold=0.8, # Lower threshold for more matches
normalize_lowercase=True, # Default: True
)
# Disable fuzzy for exact matching only
exact_evaluator = ScalarEquals(
output_path="user.id",
expected_path="user.id",
fuzzy_enabled=False, # Exact matching only
coerce_to="str",
)
# Given output/expected objects, use EvaluatorContext to evaluate
ctx = EvaluatorContext(
inputs=None,
output={"user": {"name": "Jon Smith"}},
expected_output={"user": {"name": "John Smith"}}
)
res = evaluator.evaluate(ctx)
print(res.value, res.reason) # 0.89, "[name_match] fuzzy match (score=0.89)"
```
### Advanced Fuzzy Options
```python
from pydantic_ai_helpers.evals import CompareOptions, FuzzyOptions, NormalizeOptions
# Structured options for complex cases
opts = CompareOptions(
normalize=NormalizeOptions(
lowercase=True, # Case insensitive
strip=True, # Remove whitespace
alphanum=True, # Keep only letters/numbers
),
fuzzy=FuzzyOptions(
enabled=True,
threshold=0.85, # 85% similarity required
algorithm="token_set_ratio" # Best for unordered word matching
)
)
evaluator = ScalarEquals(
output_path="description",
expected_path="description",
compare_options=opts
)
# Available fuzzy algorithms:
# - "ratio": Character-based similarity
# - "partial_ratio": Best substring match
# - "token_sort_ratio": Word-based with sorting
# - "token_set_ratio": Word-based with set logic (default)
```
### Practical Examples
```python
# Product name matching with typos
evaluator = ScalarEquals(
output_path="product_name",
expected_path="product_name",
fuzzy_threshold=0.8, # Allow some typos
normalize_lowercase=True
)
# Tag similarity for content classification
tag_recall = ListRecall(
output_path="predicted_tags",
expected_path="actual_tags",
fuzzy_enabled=True, # Handle variations like "AI" vs "artificial intelligence"
normalize_strip=True
)
# Category validation with fuzzy fallback
category_check = ValueInExpectedList(
output_path="predicted_category",
expected_path="valid_categories",
fuzzy_threshold=0.9, # High threshold for category validation
normalize_alphanum=True # Ignore punctuation differences
)
```
Notes:
- **Fuzzy matching is enabled by default** with 0.85 threshold and `token_set_ratio` algorithm
- Allowed `coerce_to` values: "str", "int", "float", "bool", "enum" (or pass an Enum class)
- `ListCompare.mode` values: "equality", "recall", "precision"
- Normalization defaults: `lowercase=True, strip=True, collapse_spaces=True, alphanum=False`
- Fuzzy algorithms: "ratio", "partial_ratio", "token_sort_ratio", "token_set_ratio"
- **Normalization always happens before fuzzy matching** for better results
## API Reference
### `History` Class
The main wrapper class that provides access to all functionality.
**Constructor:**
- `History(result_or_messages)` - Accepts a `RunResult`, `StreamedRunResult`, or `list[ModelMessage]`
**Attributes:**
- `user: RoleView` - Access user messages
- `ai: RoleView` - Access AI messages
- `system: RoleView` - Access system messages
- `tools: ToolsView` - Access tool calls and returns
- `media: MediaView` - Access media content in user messages
**Methods:**
- `all_messages() -> list[ModelMessage]` - Get raw message list
- `usage() -> Usage` - Aggregate token usage
- `tokens() -> Usage` - Alias for `usage()`
- `system_prompt() -> SystemPromptPart | None` - Get the first system prompt
### `RoleView` Class
Provides filtered access to messages by role.
**Methods:**
- `all() -> list[Part]` - Get all parts for this role
- `last() -> Part | None` - Get the most recent part
- `first() -> Part | None` - Get the first part
### `ToolsView` Class
Access tool-related messages.
**Methods:**
- `calls(*, name: str | None = None) -> ToolPartView` - Access tool calls
- `returns(*, name: str | None = None) -> ToolPartView` - Access tool returns
### `ToolPartView` Class
Filtered view of tool calls or returns.
**Methods:**
- `all() -> list[ToolCallPart | ToolReturnPart]` - Get all matching parts
- `last() -> ToolCallPart | ToolReturnPart | None` - Get the most recent part
- `first() -> ToolCallPart | ToolReturnPart | None` - Get the first part
**Args Conversion:**
When tool calls are accessed via `all()`, `last()`, or `first()`, the library automatically converts unparsed string args to dictionary args when possible. If a `ToolCallPart` has string-based args that contain valid JSON (non-empty after stripping), they will be converted to dictionary args using the `.args_as_dict()` method. This ensures consistent dictionary-based args for tool calls that contain valid JSON payloads, which is a minor deviation from the standard PydanticAI behavior that would leave them as strings.
### `MediaView` Class
Access media content from user messages (images, audio, documents, videos).
**Methods:**
- `all() -> list[MediaContent]` - Get all media content
- `last() -> MediaContent | None` - Get the most recent media item
- `first() -> MediaContent | None` - Get the first media item
- `images(*, url_only=False, binary_only=False)` - Get image content
- `audio(*, url_only=False, binary_only=False)` - Get audio content
- `documents(*, url_only=False, binary_only=False)` - Get document content
- `videos(*, url_only=False, binary_only=False)` - Get video content
- `by_type(media_type)` - Get content by specific type (e.g., `ImageUrl`, `BinaryContent`)
## Common Patterns
### Check if a Tool Was Used
```python
if hist.tools.calls(name="calculator").last():
result = hist.tools.returns(name="calculator").last()
print(f"Calculation result: {result.content}")
```
### Count Message Types
```python
print(f"User messages: {len(hist.user.all())}")
print(f"AI responses: {len(hist.ai.all())}")
print(f"Tool calls: {len(hist.tools.calls().all())}")
print(f"Tool returns: {len(hist.tools.returns().all())}")
```
### Extract Conversation Text
```python
# Get all user inputs
user_inputs = [msg.content for msg in hist.user.all()]
# Get all AI responses
ai_responses = [msg.content for msg in hist.ai.all()]
# Create a simple transcript
for user, ai in zip(user_inputs, ai_responses):
print(f"User: {user}")
print(f"AI: {ai}")
print()
```
### Work with Media Content
```python
# Check if conversation has images
if hist.media.images():
print("This conversation contains images")
for img in hist.media.images():
if hasattr(img, 'url'):
print(f"Image URL: {img.url}")
else:
print(f"Binary image: {img.media_type}, {len(img.data)} bytes")
# Process different media types
for media_item in hist.media.all():
if isinstance(media_item, ImageUrl):
download_image(media_item.url)
elif isinstance(media_item, BinaryContent):
save_binary_content(media_item.data, media_item.media_type)
```
### Extract System Configuration
```python
# Check system prompt for agent behavior
system_prompt = hist.system_prompt()
if system_prompt:
if "helpful" in system_prompt.content.lower():
agent_type = "helpful_assistant"
elif "creative" in system_prompt.content.lower():
agent_type = "creative_writer"
else:
agent_type = "general_purpose"
print(f"Agent type: {agent_type}")
```
## Design Philosophy
1. **Boring is Good** - No clever magic, just simple method calls
2. **Autocomplete-Friendly** - Your IDE knows exactly what's available
3. **Zero Config** - Works out of the box with any PydanticAI result
4. **Type Safe** - Full type hints for everything
5. **Immutable** - History objects don't modify your data
## Contributing
Found a bug? Want a feature? PRs welcome!
1. Fork the repo
2. Create your feature branch (`git checkout -b feature/amazing-feature`)
3. Write tests (we maintain 100% coverage)
4. Make your changes
5. Run `make lint test`
6. Commit your changes (`git commit -m 'Add amazing feature'`)
7. Push to the branch (`git push origin feature/amazing-feature`)
8. Open a Pull Request
## Development
```bash
# Clone the repo
git clone https://github.com/yourusername/pydantic-ai-helpers.git
cd pydantic-ai-helpers
# Install in development mode
make install
# Run tests
make test
# Run linting
make lint
# Format code
make format
```
## License
MIT - see [LICENSE](LICENSE) file.
---
Built with boredom-driven development. Because sometimes the most useful code is the code that does the obvious thing, obviously.
Raw data
{
"_id": null,
"home_page": null,
"name": "pydantic-ai-helpers",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "pydantic, ai, utilities, history, llm",
"author": "Jan Siml",
"author_email": "Jan Siml <49557684+svilupp@users.noreply.github.com>",
"download_url": "https://files.pythonhosted.org/packages/49/ae/4aa17b40eca4888979891cf23b765abd17630757280ccc7c859b96ecf775/pydantic_ai_helpers-0.0.2.tar.gz",
"platform": null,
"description": "# pydantic-ai-helpers\n\n> Boring, opinionated helpers for PydanticAI that are so dumb you didn't want to implement them. So I did.\n\n**\u26a0\ufe0f This is NOT an official PydanticAI package** - just a simple personal helper library.\n\n[](https://pypi.org/project/pydantic-ai-helpers/)\n[](https://pypi.org/project/pydantic-ai-helpers/)\n[](https://github.com/svilupp/pydantic-ai-helpers/actions)\n[](https://codecov.io/gh/svilupp/pydantic-ai-helpers)\n[](https://github.com/svilupp/pydantic-ai-helpers/blob/main/LICENSE)\n\n## The Problem\n\n[PydanticAI](https://github.com/pydantic/pydantic-ai) is amazing! But at some point you'll need to quickly and easily extract aspects of your conversations. It's not hard but it's a pain to do, because neither you nor the LLMS know how to do it, so you'll waste 10+ minutes to do:\n\n```python\n# Want the last tool call for your UI updates?\nlast_tool_call = None\nfor message in result.all_messages():\n for part in message.parts:\n if isinstance(part, ToolCallPart):\n last_tool_call = part\n\n# Need that metadata you passed for evaluations?\nmetadata_parts = []\nfor message in result.all_messages():\n for part in message.parts:\n if isinstance(part, ToolReturnPart) and part.metadata:\n metadata_parts.append(part.metadata)\n\n# How about just the user's question again?\nuser_question = None\nfor message in result.all_messages():\n for part in message.parts:\n if isinstance(part, UserPromptPart):\n user_question = part.content\n break\n```\n\nWe've all been there. **We've got you!**\n\n```python\nfrom pydantic_ai_helpers import History\n# or for convenience:\nimport pydantic_ai_helpers as ph\n\nhist = History(result) # or ph.History(result)\nlast_tool_call = hist.tools.calls().last() # Done\nmetadata = hist.tools.returns().last().metadata # Easy\nuser_question = hist.user.last().content # Simple\nsystem_prompt = hist.system_prompt() # Get system message\nmedia_items = hist.media.images() # Extract media content\n```\n\nThe best part? Your IDE will help you with the suggestions for the available methods so you don't have to remember anything!\n\n## Installation\n\n```bash\nuv add pydantic-ai-helpers\n#\u00a0pip install pydantic-ai-helpers\n# poetry add pydantic-ai-helpers\n```\n\n## Quick Start\n\n```python\nfrom pydantic_ai import Agent\nfrom pydantic_ai_helpers import History\n# or: import pydantic_ai_helpers as ph\n\nagent = Agent(\"openai:gpt-4.1-mini\")\nresult = agent.run_sync(\"Tell me a joke\")\n\n# Wrap once, access everything\nhist = History(result) # or ph.History(result)\n\n# Get the first and last user messages\nprint(hist.user.first().content) # First user message\nprint(hist.user.last().content) # Last user message\n# Output: \"Tell me a joke\"\n\n# Get all AI responses\nfor response in hist.ai.all():\n print(response.content)\n\n# Check token usage\nprint(f\"Tokens used: {hist.usage().total_tokens}\")\n\n# Access system prompt (if any)\nif system_prompt := hist.system_prompt():\n print(f\"System prompt: {system_prompt.content}\")\n\n# Access media content\nimages = hist.media.images()\nif images:\n print(f\"Found {len(images)} images in conversation\")\n```\n\n## Common Use Cases\n\n### Extract What You Need for Your App\n\n```python\nhist = History(result)\n\n# Update your UI with the latest tool status\nif latest_call := hist.tools.calls().last():\n update_ui_status(f\"Called {latest_call.tool_name}...\")\n\n# Get conversation context for logging\nuser_query = hist.user.last().content\nai_response = hist.ai.last().content\nlog_conversation(user_query, ai_response)\n\n# Check token costs for billing\ntotal_cost = hist.usage().total_tokens * your_token_rate\n```\n\n### Debug Tool Workflows\n\n```python\n# See what tools were actually called\nfor call in hist.tools.calls().all():\n print(f\"Called {call.tool_name} with {call.args}\")\n\n# Check what came back\nfor ret in hist.tools.returns().all():\n print(f\"{ret.tool_name} returned: {ret.content}\")\n if ret.metadata: # Your evaluation metadata\n print(f\"Metadata: {ret.metadata}\")\n```\n\n### Analyze Conversations\n\n```python\n# Count interactions\nprint(f\"User asked {len(hist.user.all())} questions\")\nprint(f\"AI made {len(hist.tools.calls().all())} tool calls\")\nprint(f\"Total tokens: {hist.usage().total_tokens}\")\n\n# Get specific tool results for processing\nweather_results = hist.tools.returns(name=\"get_weather\").all()\nfor result in weather_results:\n process_weather_data(result.content)\n```\n\n### Work with Media Content\n\n```python\n# Access all media content\nall_media = hist.media.all()\nprint(f\"Found {len(all_media)} media items\")\n\n# Get specific media types\nimages = hist.media.images() # All images (URLs + binary)\naudio = hist.media.audio() # All audio files\ndocuments = hist.media.documents() # All documents\nvideos = hist.media.videos() # All videos\n\n# Filter by storage type\nurl_images = hist.media.images(url_only=True) # Only ImageUrl objects\nbinary_images = hist.media.images(binary_only=True) # Only binary images\n\n# Get the most recent media\nlatest_media = hist.media.last()\nif latest_media:\n print(f\"Latest media: {type(latest_media).__name__}\")\n\n# Filter by exact type\nfrom pydantic_ai.messages import ImageUrl, BinaryContent\nimage_urls = hist.media.by_type(ImageUrl)\nbinary_content = hist.media.by_type(BinaryContent)\n```\n\n### Access System Prompts\n\n```python\n# Get the system prompt (if any)\nsystem_prompt = hist.system_prompt()\nif system_prompt:\n print(f\"System prompt: {system_prompt.content}\")\nelse:\n print(\"No system prompt found\")\n\n# Use in analysis\nif system_prompt and \"helpful\" in system_prompt.content:\n print(\"This agent was configured to be helpful\")\n```\n\n## Examples\n\n### Multi-turn Conversation Analysis\n\n```python\nmessages = []\ntopics = [\n \"What's the weather in London?\",\n \"How about Paris?\",\n \"Which city is warmer?\"\n]\n\nfor topic in topics:\n result = agent.run_sync(topic, message_history=messages)\n messages = result.all_messages()\n\nhist = History(result)\n\n# Analyze the conversation flow\nprint(f\"User asked {len(hist.user.all())} questions\")\nprint(f\"AI responded {len(hist.ai.all())} times\")\nprint(f\"Made {len(hist.tools.calls())} tool calls\")\n\n# Get specific information\nlondon_weather = hist.tools.returns(name=\"get_weather\").all()[0]\nparis_weather = hist.tools.returns(name=\"get_weather\").all()[1]\n```\n\n### Dice Game with Tools\n\n```python\n# From the PydanticAI tutorial\nresult = agent.run_sync(\"Roll a dice\")\n\nhist = History(result)\n\n# Find what the dice rolled\ndice_result = hist.tools.returns(name=\"roll_dice\").last()\nprint(f\"Dice rolled: {dice_result.content}\")\n\n# See how the AI responded\nai_message = hist.ai.last()\nprint(f\"AI said: {ai_message.content}\")\n```\n\n### Streaming Support\n\n```python\nasync with agent.run_stream(\"Tell me a story\") as result:\n async for chunk in result.stream():\n print(chunk, end=\"\")\n\n # After streaming completes\n hist = History(result)\n print(f\"\\nTotal tokens: {hist.tokens().total_tokens}\")\n```\n\n### Loading from Serialized Conversations\n\n```python\nimport json\nfrom pydantic_core import to_jsonable_python\nfrom pydantic_ai import Agent\nfrom pydantic_ai.messages import ModelMessagesTypeAdapter\n\n# Save a conversation\nagent = Agent('openai:gpt-4.1-mini')\nresult = agent.run_sync('Tell me a joke.')\nmessages = result.all_messages()\n\n# Serialize to file\nwith open('conversation.json', 'w') as f:\n json.dump(to_jsonable_python(messages), f)\n\n# Later, load it back\nhist = History('conversation.json')\nprint(hist) # History(1 turn, 50 tokens)\nprint(hist.user.last().content) # \"Tell me a joke.\"\nprint(hist.ai.last().content) # The joke response\n\n# Or use Path objects\nfrom pathlib import Path\nhist = History(Path('conversation.json'))\n\n# Continue the conversation with loaded history\nsame_messages = ModelMessagesTypeAdapter.validate_python(\n to_jsonable_python(hist.all_messages())\n)\nresult2 = agent.run_sync(\n 'Tell me a different joke.',\n message_history=same_messages\n)\n```\n\n## Evals Helpers\n\nYou can compare values and collections with simple, reusable comparators, or use small evaluator classes to compare fields by dotted paths. **Now with fuzzy string matching support!**\n\n### Quick Comparators\n\n```python\nfrom pydantic_ai_helpers.evals import ScalarCompare, ListCompare, InclusionCompare\n\n# Scalars with coercion and tolerance\ncomp = ScalarCompare(coerce_to=\"float\", abs_tol=0.01)\nscore, why = comp(\"3.14\", 3.13) # -> (1.0, 'numbers match')\n\n# Lists with recall/precision or equality\nrecall = ListCompare(mode=\"recall\")\nscore, why = recall([\"a\", \"b\"], [\"a\", \"b\", \"c\"]) # -> ~0.667\n\nequality = ListCompare(mode=\"equality\", order_sensitive=False)\nscore, _ = equality([\"a\", \"b\"], [\"b\", \"a\"]) # -> 1.0\n\n# Value in acceptable list with fuzzy matching (NEW!)\ninc = InclusionCompare() # Uses defaults: normalization + fuzzy matching\nscore, _ = inc(\"aple\", [\"apple\", \"banana\", \"cherry\"]) # -> ~0.9 (fuzzy match)\n```\n\n### Fuzzy String Matching (NEW!)\n\nThe library now includes powerful fuzzy string matching using rapidfuzz:\n\n```python\nfrom pydantic_ai_helpers.evals import ScalarCompare, CompareOptions, FuzzyOptions\n\n# Default behavior: fuzzy matching enabled with 0.85 threshold\ncomp = ScalarCompare()\nscore, why = comp(\"colour\", \"color\") # -> (0.91, 'fuzzy match (score=0.91)')\n\n# Exact matching (disable fuzzy)\ncomp = ScalarCompare(fuzzy_enabled=False)\nscore, why = comp(\"colour\", \"color\") # -> (0.0, 'values differ...')\n\n# Custom fuzzy settings\ncomp = ScalarCompare(\n fuzzy_threshold=0.9, # Stricter threshold\n fuzzy_algorithm=\"ratio\", # Different algorithm\n normalize_lowercase=True # Case insensitive\n)\n\n# For lists with fuzzy matching\nfrom pydantic_ai_helpers.evals import ListRecall\nevaluator = ListRecall() # Fuzzy enabled by default\nscore, why = evaluator(\n [\"Python\", \"AI\", \"Machine Learning\"], # Output\n [\"python\", \"ai\", \"data science\", \"ml\"] # Expected\n)\n# Uses fuzzy scores: \"Machine Learning\" partially matches \"ml\"\n```\n\n### Field-to-Field Evaluators\n\nUse evaluators when you want to compare fields inside nested objects using dotted paths:\n\n```python\nfrom pydantic_ai_helpers.evals import ScalarEquals, ListRecall, ListEquality, ValueInExpectedList\nfrom pydantic_evals.evaluators import EvaluatorContext\n\n# Basic usage (fuzzy enabled by default)\nevaluator = ScalarEquals(\n output_path=\"user.name\",\n expected_path=\"user.name\",\n evaluation_name=\"name_match\",\n)\n\n# Custom fuzzy settings for stricter matching\nevaluator = ScalarEquals(\n output_path=\"predicted.category\",\n expected_path=\"actual.category\",\n fuzzy_threshold=0.95, # Very strict\n normalize_alphanum=True, # Remove punctuation\n evaluation_name=\"category_match\",\n)\n\n# List evaluation with fuzzy matching\nlist_evaluator = ListRecall(\n output_path=\"predicted_tags\",\n expected_path=\"required_tags\",\n fuzzy_enabled=True, # Default: True\n fuzzy_threshold=0.8, # Lower threshold for more matches\n normalize_lowercase=True, # Default: True\n)\n\n# Disable fuzzy for exact matching only\nexact_evaluator = ScalarEquals(\n output_path=\"user.id\",\n expected_path=\"user.id\",\n fuzzy_enabled=False, # Exact matching only\n coerce_to=\"str\",\n)\n\n# Given output/expected objects, use EvaluatorContext to evaluate\nctx = EvaluatorContext(\n inputs=None,\n output={\"user\": {\"name\": \"Jon Smith\"}},\n expected_output={\"user\": {\"name\": \"John Smith\"}}\n)\nres = evaluator.evaluate(ctx)\nprint(res.value, res.reason) # 0.89, \"[name_match] fuzzy match (score=0.89)\"\n```\n\n### Advanced Fuzzy Options\n\n```python\nfrom pydantic_ai_helpers.evals import CompareOptions, FuzzyOptions, NormalizeOptions\n\n# Structured options for complex cases\nopts = CompareOptions(\n normalize=NormalizeOptions(\n lowercase=True, # Case insensitive\n strip=True, # Remove whitespace\n alphanum=True, # Keep only letters/numbers\n ),\n fuzzy=FuzzyOptions(\n enabled=True,\n threshold=0.85, # 85% similarity required\n algorithm=\"token_set_ratio\" # Best for unordered word matching\n )\n)\n\nevaluator = ScalarEquals(\n output_path=\"description\",\n expected_path=\"description\",\n compare_options=opts\n)\n\n# Available fuzzy algorithms:\n# - \"ratio\": Character-based similarity\n# - \"partial_ratio\": Best substring match\n# - \"token_sort_ratio\": Word-based with sorting\n# - \"token_set_ratio\": Word-based with set logic (default)\n```\n\n### Practical Examples\n\n```python\n# Product name matching with typos\nevaluator = ScalarEquals(\n output_path=\"product_name\",\n expected_path=\"product_name\",\n fuzzy_threshold=0.8, # Allow some typos\n normalize_lowercase=True\n)\n\n# Tag similarity for content classification\ntag_recall = ListRecall(\n output_path=\"predicted_tags\",\n expected_path=\"actual_tags\",\n fuzzy_enabled=True, # Handle variations like \"AI\" vs \"artificial intelligence\"\n normalize_strip=True\n)\n\n# Category validation with fuzzy fallback\ncategory_check = ValueInExpectedList(\n output_path=\"predicted_category\",\n expected_path=\"valid_categories\",\n fuzzy_threshold=0.9, # High threshold for category validation\n normalize_alphanum=True # Ignore punctuation differences\n)\n```\n\nNotes:\n- **Fuzzy matching is enabled by default** with 0.85 threshold and `token_set_ratio` algorithm\n- Allowed `coerce_to` values: \"str\", \"int\", \"float\", \"bool\", \"enum\" (or pass an Enum class)\n- `ListCompare.mode` values: \"equality\", \"recall\", \"precision\"\n- Normalization defaults: `lowercase=True, strip=True, collapse_spaces=True, alphanum=False`\n- Fuzzy algorithms: \"ratio\", \"partial_ratio\", \"token_sort_ratio\", \"token_set_ratio\"\n- **Normalization always happens before fuzzy matching** for better results\n\n## API Reference\n\n### `History` Class\n\nThe main wrapper class that provides access to all functionality.\n\n**Constructor:**\n- `History(result_or_messages)` - Accepts a `RunResult`, `StreamedRunResult`, or `list[ModelMessage]`\n\n**Attributes:**\n- `user: RoleView` - Access user messages\n- `ai: RoleView` - Access AI messages\n- `system: RoleView` - Access system messages\n- `tools: ToolsView` - Access tool calls and returns\n- `media: MediaView` - Access media content in user messages\n\n**Methods:**\n- `all_messages() -> list[ModelMessage]` - Get raw message list\n- `usage() -> Usage` - Aggregate token usage\n- `tokens() -> Usage` - Alias for `usage()`\n- `system_prompt() -> SystemPromptPart | None` - Get the first system prompt\n\n### `RoleView` Class\n\nProvides filtered access to messages by role.\n\n**Methods:**\n- `all() -> list[Part]` - Get all parts for this role\n- `last() -> Part | None` - Get the most recent part\n- `first() -> Part | None` - Get the first part\n\n### `ToolsView` Class\n\nAccess tool-related messages.\n\n**Methods:**\n- `calls(*, name: str | None = None) -> ToolPartView` - Access tool calls\n- `returns(*, name: str | None = None) -> ToolPartView` - Access tool returns\n\n### `ToolPartView` Class\n\nFiltered view of tool calls or returns.\n\n**Methods:**\n- `all() -> list[ToolCallPart | ToolReturnPart]` - Get all matching parts\n- `last() -> ToolCallPart | ToolReturnPart | None` - Get the most recent part\n- `first() -> ToolCallPart | ToolReturnPart | None` - Get the first part\n\n**Args Conversion:**\nWhen tool calls are accessed via `all()`, `last()`, or `first()`, the library automatically converts unparsed string args to dictionary args when possible. If a `ToolCallPart` has string-based args that contain valid JSON (non-empty after stripping), they will be converted to dictionary args using the `.args_as_dict()` method. This ensures consistent dictionary-based args for tool calls that contain valid JSON payloads, which is a minor deviation from the standard PydanticAI behavior that would leave them as strings.\n\n### `MediaView` Class\n\nAccess media content from user messages (images, audio, documents, videos).\n\n**Methods:**\n- `all() -> list[MediaContent]` - Get all media content\n- `last() -> MediaContent | None` - Get the most recent media item\n- `first() -> MediaContent | None` - Get the first media item\n- `images(*, url_only=False, binary_only=False)` - Get image content\n- `audio(*, url_only=False, binary_only=False)` - Get audio content\n- `documents(*, url_only=False, binary_only=False)` - Get document content\n- `videos(*, url_only=False, binary_only=False)` - Get video content\n- `by_type(media_type)` - Get content by specific type (e.g., `ImageUrl`, `BinaryContent`)\n\n## Common Patterns\n\n### Check if a Tool Was Used\n\n```python\nif hist.tools.calls(name=\"calculator\").last():\n result = hist.tools.returns(name=\"calculator\").last()\n print(f\"Calculation result: {result.content}\")\n```\n\n### Count Message Types\n\n```python\nprint(f\"User messages: {len(hist.user.all())}\")\nprint(f\"AI responses: {len(hist.ai.all())}\")\nprint(f\"Tool calls: {len(hist.tools.calls().all())}\")\nprint(f\"Tool returns: {len(hist.tools.returns().all())}\")\n```\n\n### Extract Conversation Text\n\n```python\n# Get all user inputs\nuser_inputs = [msg.content for msg in hist.user.all()]\n\n# Get all AI responses\nai_responses = [msg.content for msg in hist.ai.all()]\n\n# Create a simple transcript\nfor user, ai in zip(user_inputs, ai_responses):\n print(f\"User: {user}\")\n print(f\"AI: {ai}\")\n print()\n```\n\n### Work with Media Content\n\n```python\n# Check if conversation has images\nif hist.media.images():\n print(\"This conversation contains images\")\n for img in hist.media.images():\n if hasattr(img, 'url'):\n print(f\"Image URL: {img.url}\")\n else:\n print(f\"Binary image: {img.media_type}, {len(img.data)} bytes\")\n\n# Process different media types\nfor media_item in hist.media.all():\n if isinstance(media_item, ImageUrl):\n download_image(media_item.url)\n elif isinstance(media_item, BinaryContent):\n save_binary_content(media_item.data, media_item.media_type)\n```\n\n### Extract System Configuration\n\n```python\n# Check system prompt for agent behavior\nsystem_prompt = hist.system_prompt()\nif system_prompt:\n if \"helpful\" in system_prompt.content.lower():\n agent_type = \"helpful_assistant\"\n elif \"creative\" in system_prompt.content.lower():\n agent_type = \"creative_writer\"\n else:\n agent_type = \"general_purpose\"\n\n print(f\"Agent type: {agent_type}\")\n```\n\n## Design Philosophy\n\n1. **Boring is Good** - No clever magic, just simple method calls\n2. **Autocomplete-Friendly** - Your IDE knows exactly what's available\n3. **Zero Config** - Works out of the box with any PydanticAI result\n4. **Type Safe** - Full type hints for everything\n5. **Immutable** - History objects don't modify your data\n\n## Contributing\n\nFound a bug? Want a feature? PRs welcome!\n\n1. Fork the repo\n2. Create your feature branch (`git checkout -b feature/amazing-feature`)\n3. Write tests (we maintain 100% coverage)\n4. Make your changes\n5. Run `make lint test`\n6. Commit your changes (`git commit -m 'Add amazing feature'`)\n7. Push to the branch (`git push origin feature/amazing-feature`)\n8. Open a Pull Request\n\n## Development\n\n```bash\n# Clone the repo\ngit clone https://github.com/yourusername/pydantic-ai-helpers.git\ncd pydantic-ai-helpers\n\n# Install in development mode\nmake install\n\n# Run tests\nmake test\n\n# Run linting\nmake lint\n\n# Format code\nmake format\n```\n\n## License\n\nMIT - see [LICENSE](LICENSE) file.\n\n---\n\nBuilt with boredom-driven development. Because sometimes the most useful code is the code that does the obvious thing, obviously.\n",
"bugtrack_url": null,
"license": null,
"summary": "Boring, opinionated helpers for PydanticAI that are so simple you didn't want to even vibe code them. (Unofficial)",
"version": "0.0.2",
"project_urls": {
"Changelog": "https://github.com/yourusername/pydantic-ai-helpers/blob/main/CHANGELOG.md",
"Documentation": "https://github.com/yourusername/pydantic-ai-helpers#readme",
"Homepage": "https://github.com/yourusername/pydantic-ai-helpers",
"Issues": "https://github.com/yourusername/pydantic-ai-helpers/issues",
"Repository": "https://github.com/yourusername/pydantic-ai-helpers"
},
"split_keywords": [
"pydantic",
" ai",
" utilities",
" history",
" llm"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "b8ced462d5606111438b8661e40c1b017f94be08bab6828978afba9a384e3b52",
"md5": "b0d3ad1cedbc914b610cb6693811ad08",
"sha256": "53b1807dfda90c195526f403236aca84ddc783fdd6b57ceb7516b49e77c935ee"
},
"downloads": -1,
"filename": "pydantic_ai_helpers-0.0.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "b0d3ad1cedbc914b610cb6693811ad08",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 35201,
"upload_time": "2025-08-17T20:35:11",
"upload_time_iso_8601": "2025-08-17T20:35:11.801503Z",
"url": "https://files.pythonhosted.org/packages/b8/ce/d462d5606111438b8661e40c1b017f94be08bab6828978afba9a384e3b52/pydantic_ai_helpers-0.0.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "49ae4aa17b40eca4888979891cf23b765abd17630757280ccc7c859b96ecf775",
"md5": "757263dc7c39c3c899cd24adaf1ea977",
"sha256": "eb977c29eb23decec78fb31f07c7bc06fdb0bd53f551d7fb8b77e65bc69e1f82"
},
"downloads": -1,
"filename": "pydantic_ai_helpers-0.0.2.tar.gz",
"has_sig": false,
"md5_digest": "757263dc7c39c3c899cd24adaf1ea977",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 31256,
"upload_time": "2025-08-17T20:35:13",
"upload_time_iso_8601": "2025-08-17T20:35:13.041862Z",
"url": "https://files.pythonhosted.org/packages/49/ae/4aa17b40eca4888979891cf23b765abd17630757280ccc7c859b96ecf775/pydantic_ai_helpers-0.0.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-17 20:35:13",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "yourusername",
"github_project": "pydantic-ai-helpers",
"github_not_found": true,
"lcname": "pydantic-ai-helpers"
}