"""
title: User Info Retrieval Filter
description: Retrieve and inject user information (ID, name, email, role) into the model's context.
author: Wes Caldwell
email:
[email protected]
version: 0.3.0
license: MIT
"""
import asyncio
from typing import Callable, Awaitable, Any
from pydantic import BaseModel, Field
class Filter:
class Valves(BaseModel):
ALLOW_ID_RETRIEVAL: bool = Field(
default=True, description="Whether to allow retrieval of user IDs."
)
ALLOW_NAME_RETRIEVAL: bool = Field(
default=True, description="Whether to allow retrieval of user names."
)
ALLOW_EMAIL_RETRIEVAL: bool = Field(
default=True, description="Whether to allow retrieval of user emails."
)
ALLOW_ROLE_RETRIEVAL: bool = Field(
default=True, description="Whether to allow retrieval of user roles."
)
DISPLAY_EVENT_EMITTERS: bool = Field(
default=True,
description="Whether to display event emitters during user information retrieval.",
)
class UserValves(BaseModel):
EXCLUDE_USER_ID: bool = Field(
default=False,
description="Whether to exclude the user's ID from the retrieved information.",
)
EXCLUDE_USER_NAME: bool = Field(
default=False,
description="Whether to exclude the user's name from the retrieved information.",
)
EXCLUDE_USER_EMAIL: bool = Field(
default=False,
description="Whether to exclude the user's email from the retrieved information.",
)
EXCLUDE_USER_ROLE: bool = Field(
default=False,
description="Whether to exclude the user's role from the retrieved information.",
)
DISPLAY_EVENT_EMITTERS: bool = Field(
default=True,
description="Whether to display event emitters during user information retrieval (user-specific setting).",
)
def __init__(self):
self.valves = self.Valves()
async def inlet(
self,
body: dict,
__event_emitter__: Callable[[dict], Awaitable[None]],
__user__: dict = {},
) -> dict:
"""
Retrieve user information and inject it into the model's context, respecting both Valves and UserValves settings.
"""
result = []
retrieved_items = []
# Get user_valves directly from __user__ dict
user_valves = __user__.get("valves")
if not user_valves:
user_valves = self.UserValves() # Use default UserValves if not provided
# Determine whether to display event emitters based on both global and user-specific settings
display_event_emitters = (
self.valves.DISPLAY_EVENT_EMITTERS and user_valves.DISPLAY_EVENT_EMITTERS
)
if display_event_emitters:
await __event_emitter__(
{
"type": "status",
"data": {
"status": "in_progress",
"description": "Starting user information retrieval",
"done": False,
},
}
)
if (
self.valves.ALLOW_ID_RETRIEVAL
and not user_valves.EXCLUDE_USER_ID
and "id" in __user__
):
result.append(f"ID: {__user__['id']}")
retrieved_items.append("ID")
if (
self.valves.ALLOW_NAME_RETRIEVAL
and not user_valves.EXCLUDE_USER_NAME
and "name" in __user__
):
result.append(f"User: {__user__['name']}")
retrieved_items.append("name")
if (
self.valves.ALLOW_EMAIL_RETRIEVAL
and not user_valves.EXCLUDE_USER_EMAIL
and "email" in __user__
):
result.append(f"Email: {__user__['email']}")
retrieved_items.append("email")
if (
self.valves.ALLOW_ROLE_RETRIEVAL
and not user_valves.EXCLUDE_USER_ROLE
and "role" in __user__
):
result.append(f"Role: {__user__['role']}")
retrieved_items.append("role")
if not result:
result = ["User: Unknown"]
retrieved_items = ["no data"]
final_result = " | ".join(result)
if display_event_emitters:
await __event_emitter__(
{
"type": "status",
"data": {
"status": "complete",
"description": f"User information retrieval completed. Retrieved: {', '.join(retrieved_items)}",
"done": True,
},
}
)
# Inject retrieved user information into system context
system_message = {
"role": "system",
"content": f"User information: {final_result}",
}
body.setdefault("messages", []).insert(0, system_message)
return body
