1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
| """
title: Tavily Search Tool
author: victor1203
maintainer: sievelau
description: This tool performs internet searches using the Tavily API to get real-time information with advanced context and Q&A capabilities
required_open_webui_version: 0.4.0
requirements: tavily-python
version: 1.1.0
licence: MIT
"""
from pydantic import BaseModel, Field
from typing import Optional, Literal
from tavily import TavilyClient
class Tools:
def __init__(self):
"""Initialize the Tool with valves."""
self.valves = self.Valves()
self._client = None
class Valves(BaseModel):
tavily_api_key: str = Field(
"", description="Your Tavily API key (starts with 'tvly-')"
)
search_depth: str = Field(
"basic", description="Search depth - basic or advanced"
)
include_answer: bool = Field(
True, description="Include an AI-generated answer in the response"
)
max_results: int = Field(
5, description="Maximum number of search results to return (1-10)"
)
def _get_client(self) -> TavilyClient:
"""Get or create Tavily client instance."""
if self._client is None:
self._client = TavilyClient(api_key=self.valves.tavily_api_key)
return self._client
async def search(self, query: str, __event_emitter__=None) -> str:
"""
Perform a Tavily search and return the results.
Args:
query: The search query string
search_type: Type of search to perform:
- regular: Standard search with full results
- context: Optimized for RAG applications
- qa: Quick answer to a specific question
Returns:
A formatted string containing the search results
"""
try:
if not self.valves.tavily_api_key:
return "Error: Tavily API key not configured. Please set up the API key in the tool settings."
if __event_emitter__:
await __event_emitter__(
{
"type": "status",
"data": {
"description": f"Initiating Tavily search...",
"done": False,
},
}
)
client = self._get_client()
if __event_emitter__:
await __event_emitter__(
{
"type": "status",
"data": {
"description": "Fetching search results...",
"done": False,
},
}
)
result = client.search(
query=query,
search_depth=self.valves.search_depth,
include_answer=self.valves.include_answer,
max_results=self.valves.max_results,
)
formatted_results = ""
if self.valves.include_answer and "answer" in result:
formatted_results += f"AI Answer:\n{result['answer']}\n\n"
formatted_results += "Tavily Search Results:\n\n"
for i, item in enumerate(
result.get("results", [])[: self.valves.max_results], 1
):
formatted_results += f"{i}. {item.get('title', 'No title')}\n"
formatted_results += f"URL: {item.get('url', 'No URL')}\n"
formatted_results += f"Content: {item.get('content', 'No content')}\n"
formatted_results += f"Relevance Score: {item.get('score', '0.0')}\n"
formatted_results += "\n"
if __event_emitter__:
await __event_emitter__(
{
"type": "status",
"data": {
"description": "Search completed successfully",
"done": True,
},
}
)
return formatted_results
except Exception as e:
error_message = f"An error occurred while performing the search: {str(e)}"
if __event_emitter__:
await __event_emitter__(
{
"type": "status",
"data": {"description": error_message, "done": True},
}
)
return error_message
|
