-
Notifications
You must be signed in to change notification settings - Fork 1.4k
Closed
Labels
Description
Initial Checks
- I confirm that I'm using the latest version of Pydantic AI
- I confirm that I searched for my issue in https://github.com/pydantic/pydantic-ai/issues before opening this issue
Description
Bug: TypeError when importing duckduckgo_search_tool
Description
Importing pydantic_ai.common_tools.duckduckgo.duckduckgo_search_tool raises a TypeError due to missing from __future__ import annotations.
Error
from pydantic_ai.common_tools.duckduckgo import duckduckgo_search_toolTypeError: unsupported operand type(s) for |: '_DDGSLazyLoader' and 'NoneType'
File "pydantic_ai/common_tools/duckduckgo.py", line 65, in <module>
def duckduckgo_search_tool(duckduckgo_client: DDGS | None = None, max_results: int | None = None):
Root Cause
ddgspackage exportsDDGSas a_DDGSLazyLoaderinstance (not a type)- Type annotation
DDGS | Noneevaluates at import time - The lazy loader instance doesn't support the
|operator
Fix
Add from __future__ import annotations to the top of pydantic_ai/common_tools/duckduckgo.py:
from __future__ import annotations
import functools
from dataclasses import KW_ONLY, dataclass
# ... rest of fileImpact
- Particularly problematic in Temporal workflows where imports occur during activity execution
- Causes immediate import failures in any code using the DuckDuckGo tool
Will open a PR with the fix and link it here
Example Code
import asyncio
from datetime import timedelta
from temporalio import activity, workflow
from temporalio.client import Client
from temporalio.worker import Worker
with workflow.unsafe.imports_passed_through():
from pydantic_ai.common_tools.duckduckgo import duckduckgo_search_tool
@activity.defn
async def web_search_activity(query: str) -> list[dict]:
"""Same error get's throw in a DurableAgent as we try to make the same import"""
tool = duckduckgo_search_tool(max_results=3)
return await tool.function(query)
@workflow.defn
class SearchWorkflow:
@workflow.run
async def run(self, query: str) -> list[dict]:
return await workflow.execute_activity(
web_search_activity,
query,
start_to_close_timeout=timedelta(seconds=30),
)
async def main():
client = await Client.connect("localhost:7233")
async with Worker(
client,
task_queue="test-duckduckgo-queue",
workflows=[SearchWorkflow],
activities=[web_search_activity],
):
result = await client.execute_workflow(
SearchWorkflow.run,
"Python programming",
id=f"test-search-{asyncio.get_event_loop().time()}",
task_queue="test-duckduckgo-queue",
)
print(f"Success! Found {len(result)} results")
if __name__ == "__main__":
asyncio.run(main())Python, Pydantic AI & LLM client version
- Python 3.12
- pydantic-ai-slim[duckduckgo] >= 1.10.0
- ddgs >= 9.7.0