From a16c0ed4c4edb37895a59a23790500bf9262dd09 Mon Sep 17 00:00:00 2001 From: quanglm Date: Tue, 25 Nov 2025 19:52:30 +0700 Subject: [PATCH] feat: add all duckduckgo search type tools --- docs/common-tools.md | 36 ++- .../pydantic_ai/common_tools/duckduckgo.py | 288 +++++++++++++++++- 2 files changed, 313 insertions(+), 11 deletions(-) diff --git a/docs/common-tools.md b/docs/common-tools.md index 78baaaad02..dbb6f7193f 100644 --- a/docs/common-tools.md +++ b/docs/common-tools.md @@ -2,23 +2,25 @@ Pydantic AI ships with native tools that can be used to enhance your agent's capabilities. -## DuckDuckGo Search Tool +## DuckDuckGo Tools -The DuckDuckGo search tool allows you to search the web for information. It is built on top of the +The DuckDuckGo tools allow you to search the web for text, images, videos, and news. They are built on top of the [DuckDuckGo API](https://github.com/deedy5/ddgs). ### Installation -To use [`duckduckgo_search_tool`][pydantic_ai.common_tools.duckduckgo.duckduckgo_search_tool], you need to install +To use the DuckDuckGo tools, you need to install [`pydantic-ai-slim`](install.md#slim-install) with the `duckduckgo` optional group: ```bash pip/uv-add "pydantic-ai-slim[duckduckgo]" ``` -### Usage +### Text Search + +The [`duckduckgo_search_tool`][pydantic_ai.common_tools.duckduckgo.duckduckgo_search_tool] searches for text results. -Here's an example of how you can use the DuckDuckGo search tool with an agent: +Here's an example of how you can use the DuckDuckGo text search tool with an agent: ```py {title="duckduckgo_search.py" test="skip"} from pydantic_ai import Agent @@ -81,6 +83,30 @@ Would you like help finding a current source or additional details on where to l """ ``` +### Images Search + +The [`duckduckgo_images_search_tool`][pydantic_ai.common_tools.duckduckgo.duckduckgo_images_search_tool] searches for images with support for filtering by size, color, type, layout, and license. + +```py {title="duckduckgo_images.py"} +--8<-- "examples/pydantic_ai_examples/duckduckgo_images.py" +``` + +### Videos Search + +The [`duckduckgo_videos_search_tool`][pydantic_ai.common_tools.duckduckgo.duckduckgo_videos_search_tool] searches for videos with support for filtering by resolution, duration, and license. + +```py {title="duckduckgo_videos.py"} +--8<-- "examples/pydantic_ai_examples/duckduckgo_videos.py" +``` + +### News Search + +The [`duckduckgo_news_search_tool`][pydantic_ai.common_tools.duckduckgo.duckduckgo_news_search_tool] searches for recent news articles. + +```py {title="duckduckgo_news.py"} +--8<-- "examples/pydantic_ai_examples/duckduckgo_news.py" +``` + ## Tavily Search Tool !!! info diff --git a/pydantic_ai_slim/pydantic_ai/common_tools/duckduckgo.py b/pydantic_ai_slim/pydantic_ai/common_tools/duckduckgo.py index c0a32f68b7..93a20dce42 100644 --- a/pydantic_ai_slim/pydantic_ai/common_tools/duckduckgo.py +++ b/pydantic_ai_slim/pydantic_ai/common_tools/duckduckgo.py @@ -4,7 +4,7 @@ import anyio import anyio.to_thread from pydantic import TypeAdapter -from typing_extensions import Any, TypedDict +from typing_extensions import Any, Literal, TypedDict from pydantic_ai.tools import Tool @@ -19,7 +19,7 @@ 'you can use the `duckduckgo` optional group — `pip install "pydantic-ai-slim[duckduckgo]"`' ) from _import_error -__all__ = ('duckduckgo_search_tool',) +__all__ = ('duckduckgo_search_tool', 'duckduckgo_images_search_tool', 'duckduckgo_videos_search_tool', 'duckduckgo_news_search_tool') class DuckDuckGoResult(TypedDict): @@ -33,7 +33,78 @@ class DuckDuckGoResult(TypedDict): """The body of the search result.""" +class DuckDuckGoImageResult(TypedDict, total=False): + """A DuckDuckGo image search result.""" + + title: str + """The title of the image.""" + image: str + """The image URL.""" + thumbnail: str + """The thumbnail URL.""" + url: str + """The source page URL.""" + height: int + """The image height.""" + width: int + """The image width.""" + source: str + """The image source.""" + + +class DuckDuckGoVideoResult(TypedDict, total=False): + """A DuckDuckGo video search result.""" + + content: str + """The video content URL.""" + description: str + """The video description.""" + duration: str + """The video duration.""" + embed_html: str + """The embed HTML.""" + embed_url: str + """The embed URL.""" + image_token: str + """The image token.""" + images: dict[str, str] + """The video images (large, medium, motion, small).""" + provider: str + """The video provider.""" + published: str + """The publication date.""" + publisher: str + """The video publisher.""" + statistics: dict[str, int] + """The video statistics (e.g., viewCount).""" + title: str + """The video title.""" + uploader: str + """The video uploader.""" + + +class DuckDuckGoNewsResult(TypedDict, total=False): + """A DuckDuckGo news search result.""" + + date: str + """The publication date.""" + title: str + """The news title.""" + body: str + """The news body.""" + url: str + """The news URL.""" + image: str + """The news image URL.""" + source: str + """The news source.""" + + +# TypeAdapters for validation duckduckgo_ta = TypeAdapter(list[DuckDuckGoResult]) +images_ta = TypeAdapter(list[DuckDuckGoImageResult]) +videos_ta = TypeAdapter(list[DuckDuckGoVideoResult]) +news_ta = TypeAdapter(list[DuckDuckGoNewsResult]) @dataclass @@ -48,22 +119,31 @@ class DuckDuckGoSearchTool: max_results: int | None """The maximum number of results. If None, returns results only from the first response.""" - async def __call__(self, query: str) -> list[DuckDuckGoResult]: + async def __call__( + self, + query: str, + region: str = 'us-en', + safesearch: Literal['on', 'moderate', 'off'] = 'moderate', + ) -> list[DuckDuckGoResult]: """Searches DuckDuckGo for the given query and returns the results. Args: query: The query to search for. + region: The region to search in (e.g., 'us-en', 'uk-en'). + safesearch: The safe search setting ('on', 'moderate', or 'off'). Returns: The search results. """ - search = functools.partial(self.client.text, max_results=self.max_results) + search = functools.partial( + self.client.text, region=region, safesearch=safesearch, max_results=self.max_results + ) results = await anyio.to_thread.run_sync(search, query) return duckduckgo_ta.validate_python(results) def duckduckgo_search_tool(duckduckgo_client: DDGS | None = None, max_results: int | None = None): - """Creates a DuckDuckGo search tool. + """Creates a DuckDuckGo text search tool. Args: duckduckgo_client: The DuckDuckGo search client. @@ -72,5 +152,201 @@ def duckduckgo_search_tool(duckduckgo_client: DDGS | None = None, max_results: i return Tool[Any]( DuckDuckGoSearchTool(client=duckduckgo_client or DDGS(), max_results=max_results).__call__, name='duckduckgo_search', - description='Searches DuckDuckGo for the given query and returns the results.', + description='Searches DuckDuckGo for the given query and returns text results.', + ) + + +@dataclass +class DuckDuckGoImagesSearchTool: + """The DuckDuckGo images search tool.""" + + client: DDGS + """The DuckDuckGo search client.""" + + _: KW_ONLY + + max_results: int | None + """The maximum number of results. If None, returns results only from the first response.""" + + async def __call__( + self, + query: str, + region: str = 'us-en', + safesearch: Literal['on', 'moderate', 'off'] = 'moderate', + size: Literal['Small', 'Medium', 'Large', 'Wallpaper'] | None = None, + color: Literal[ + 'color', + 'Monochrome', + 'Red', + 'Orange', + 'Yellow', + 'Green', + 'Blue', + 'Purple', + 'Pink', + 'Brown', + 'Black', + 'Gray', + 'Teal', + 'White', + ] + | None = None, + type_image: Literal['photo', 'clipart', 'gif', 'transparent', 'line'] | None = None, + layout: Literal['Square', 'Tall', 'Wide'] | None = None, + license_image: Literal['any', 'Public', 'Share', 'ShareCommercially', 'Modify', 'ModifyCommercially'] + | None = None, + ) -> list[DuckDuckGoImageResult]: + """Searches DuckDuckGo for images matching the given query. + + Args: + query: The query to search for. + region: The region to search in (e.g., 'us-en', 'uk-en'). + safesearch: The safe search setting ('on', 'moderate', or 'off'). + size: Filter by image size. + color: Filter by image color. + type_image: Filter by image type. + layout: Filter by image layout. + license_image: Filter by image license. + + Returns: + The image search results. + """ + kwargs = {'region': region, 'safesearch': safesearch, 'max_results': self.max_results} + if size is not None: + kwargs['size'] = size + if color is not None: + kwargs['color'] = color + if type_image is not None: + kwargs['type_image'] = type_image + if layout is not None: + kwargs['layout'] = layout + if license_image is not None: + kwargs['license_image'] = license_image + + search = functools.partial(self.client.images, **kwargs) + results = await anyio.to_thread.run_sync(search, query) + return images_ta.validate_python(results) + + +def duckduckgo_images_search_tool(duckduckgo_client: DDGS | None = None, max_results: int | None = None): + """Creates a DuckDuckGo images search tool. + + Args: + duckduckgo_client: The DuckDuckGo search client. + max_results: The maximum number of results. If None, returns results only from the first response. + """ + return Tool[Any]( + DuckDuckGoImagesSearchTool(client=duckduckgo_client or DDGS(), max_results=max_results).__call__, + name='duckduckgo_images', + description='Searches DuckDuckGo for images matching the given query.', + ) + + +@dataclass +class DuckDuckGoVideosSearchTool: + """The DuckDuckGo videos search tool.""" + + client: DDGS + """The DuckDuckGo search client.""" + + _: KW_ONLY + + max_results: int | None + """The maximum number of results. If None, returns results only from the first response.""" + + async def __call__( + self, + query: str, + region: str = 'us-en', + safesearch: Literal['on', 'moderate', 'off'] = 'moderate', + resolution: str | None = None, + duration: Literal['short', 'medium', 'long'] | None = None, + license_videos: Literal['creativeCommon', 'youtube'] | None = None, + ) -> list[DuckDuckGoVideoResult]: + """Searches DuckDuckGo for videos matching the given query. + + Args: + query: The query to search for. + region: The region to search in (e.g., 'us-en', 'uk-en'). + safesearch: The safe search setting ('on', 'moderate', or 'off'). + resolution: Filter by video resolution. + duration: Filter by video duration ('short', 'medium', or 'long'). + license_videos: Filter by video license. + + Returns: + The video search results. + """ + kwargs = {'region': region, 'safesearch': safesearch, 'max_results': self.max_results} + if resolution is not None: + kwargs['resolution'] = resolution + if duration is not None: + kwargs['duration'] = duration + if license_videos is not None: + kwargs['license_videos'] = license_videos + + search = functools.partial(self.client.videos, **kwargs) + results = await anyio.to_thread.run_sync(search, query) + return videos_ta.validate_python(results) + + +def duckduckgo_videos_search_tool(duckduckgo_client: DDGS | None = None, max_results: int | None = None): + """Creates a DuckDuckGo videos search tool. + + Args: + duckduckgo_client: The DuckDuckGo search client. + max_results: The maximum number of results. If None, returns results only from the first response. + """ + return Tool[Any]( + DuckDuckGoVideosSearchTool(client=duckduckgo_client or DDGS(), max_results=max_results).__call__, + name='duckduckgo_videos', + description='Searches DuckDuckGo for videos matching the given query.', + ) + + +@dataclass +class DuckDuckGoNewsSearchTool: + """The DuckDuckGo news search tool.""" + + client: DDGS + """The DuckDuckGo search client.""" + + _: KW_ONLY + + max_results: int | None + """The maximum number of results. If None, returns results only from the first response.""" + + async def __call__( + self, + query: str, + region: str = 'us-en', + safesearch: Literal['on', 'moderate', 'off'] = 'moderate', + ) -> list[DuckDuckGoNewsResult]: + """Searches DuckDuckGo for news articles matching the given query. + + Args: + query: The query to search for. + region: The region to search in (e.g., 'us-en', 'uk-en'). + safesearch: The safe search setting ('on', 'moderate', or 'off'). + + Returns: + The news search results. + """ + kwargs = {'region': region, 'safesearch': safesearch, 'max_results': self.max_results} + + search = functools.partial(self.client.news, **kwargs) + results = await anyio.to_thread.run_sync(search, query) + return news_ta.validate_python(results) + + +def duckduckgo_news_search_tool(duckduckgo_client: DDGS | None = None, max_results: int | None = None): + """Creates a DuckDuckGo news search tool. + + Args: + duckduckgo_client: The DuckDuckGo search client. + max_results: The maximum number of results. If None, returns results only from the first response. + """ + return Tool[Any]( + DuckDuckGoNewsSearchTool(client=duckduckgo_client or DDGS(), max_results=max_results).__call__, + name='duckduckgo_news', + description='Searches DuckDuckGo for news articles matching the given query.', )