from pydantic import BaseModel, Field
from typing import Callable, Awaitable, Any, Optional
import re
from apps.webui.models.users import Users
from apps.webui.models.tools import Tools
from utils.misc import get_last_user_message
class Filter:
class Valves(BaseModel):
status: bool = Field(default=False)
match_threshold: int = Field(
default=1
) # Default threshold for number of matching words
symbols: list = Field(default=["?"]) # Default list of symbols
def __init__(self):
self.valves = self.Valves()
async def inlet(
self,
body: dict,
__event_emitter__: Callable[[Any], Awaitable[None]],
__user__: Optional[dict] = None,
__model__: Optional[dict] = None,
) -> dict:
messages = body["messages"]
user_message = get_last_user_message(messages)
if self.valves.status:
await __event_emitter__(
{
"type": "status",
"data": {
"description": "Finding the right tools...",
"done": False,
},
}
)
all_tools = [
{"id": tool.id, "description": tool.meta.description}
for tool in Tools.get_tools()
]
available_tool_ids = (
__model__.get("info", {}).get("meta", {}).get("toolIds", [])
)
available_tools = [
tool for tool in all_tools if tool["id"] in available_tool_ids
]
# Check each word and symbol in user_message against tool descriptions
user_words = re.findall(r"\w+", user_message.lower())
user_symbols = [char for char in user_message if char in self.valves.symbols]
matched_tool_ids = []
for tool in available_tools:
description_words = re.findall(r"\w+", tool["description"].lower())
description_symbols = [
char for char in tool["description"] if char in self.valves.symbols
]
match_count = sum(
1 for word in user_words if word in description_words
) + sum(1 for symbol in user_symbols if symbol in description_symbols)
if match_count >= self.valves.match_threshold:
matched_tool_ids.append(tool["id"])
if matched_tool_ids:
body["tool_ids"] = matched_tool_ids
if self.valves.status:
await __event_emitter__(
{
"type": "status",
"data": {
"description": f"Found matching tools: {', '.join(matched_tool_ids)}",
"done": True,
},
}
)
if __event_emitter__:
await __event_emitter__(
{
"type": "status",
"data": {
"description": f"List of tools chosen: {', '.join(matched_tool_ids)}",
"done": True,
},
}
)
else:
if self.valves.status:
await __event_emitter__(
{
"type": "status",
"data": {
"description": "No matching tools found.",
"done": True,
},
}
)
return body