Spaces:
Sleeping
Sleeping
| """ | |
| Tools node using state-of-the-art patterns with Pydantic validation. | |
| """ | |
| import os | |
| from typing import List, Optional | |
| from langchain_core.tools import BaseTool | |
| from langgraph.prebuilt import ToolNode | |
| from pydantic import BaseModel, Field, validator | |
| from dotenv import load_dotenv | |
| # Load environment variables | |
| load_dotenv() | |
| # Import tools with proper schemas | |
| from ..agent_tools.execute_sql_query import execute_sql_query | |
| from ..agent_tools.get_exchange_rates import exchange_converter | |
| from ..agent_tools.create_quote import create_quote | |
| from ..agent_tools.tavily_search_tool import tavily_search_product_specs | |
| from ..agent_tools.tavily_web_extract_tool import tavily_extract_product_content | |
| class ToolConfig(BaseModel): | |
| """Configuration for tools with validation.""" | |
| enable_advanced_tools: bool = Field(default=False, description="Enable advanced database tools") | |
| max_tools: int = Field(default=10, ge=1, le=50, description="Maximum number of tools to load") | |
| model_name: str = Field(default_factory=lambda: os.getenv("MODEL_NAME", "gpt-4o-mini"), description="Model optimized for these tools") | |
| def validate_model_name(cls, v): | |
| allowed_models = ["gpt-5-mini", "gpt-4o-mini", "gpt-4", "gpt-3.5-turbo"] | |
| if v not in allowed_models: | |
| raise ValueError(f"Model must be one of {allowed_models}") | |
| return v | |
| class ToolRegistry(BaseModel): | |
| """Registry for validated tools.""" | |
| tools: List[BaseTool] = Field(description="List of validated tools") | |
| config: ToolConfig = Field(description="Tool configuration") | |
| class Config: | |
| arbitrary_types_allowed = True | |
| def validate_tools(cls, v): | |
| """Validate that all tools have proper schemas.""" | |
| for tool in v: | |
| if not hasattr(tool, 'args_schema'): | |
| raise ValueError(f"Tool {tool.name} missing args_schema for validation") | |
| if not hasattr(tool, 'name') or not tool.name: | |
| raise ValueError("Tool missing required name attribute") | |
| return v | |
| def get_all_tools(config: Optional[ToolConfig] = None) -> List[BaseTool]: | |
| """ | |
| Get gpt-5-mini optimized tools with proper validation. | |
| Args: | |
| config: Optional tool configuration with validation | |
| Returns: | |
| List of validated tools | |
| """ | |
| if config is None: | |
| config = ToolConfig() | |
| # Core tools with Pydantic schemas for LangGraph | |
| core_tools = [ | |
| execute_sql_query, | |
| exchange_converter, | |
| create_quote, | |
| tavily_search_product_specs, | |
| tavily_extract_product_content | |
| ] | |
| # Create registry with validation | |
| registry = ToolRegistry(tools=core_tools, config=config) | |
| return registry.tools[:config.max_tools] | |
| def create_tool_node(config: Optional[ToolConfig] = None) -> ToolNode: | |
| """Create a validated tool node with LangGraph built-in error handling.""" | |
| tools = get_all_tools(config) | |
| # LangGraph ToolNode handles validation, execution, and error handling automatically | |
| return ToolNode(tools) | |
| # Factory function with LangGraph best practices | |
| def create_optimized_tool_node() -> ToolNode: | |
| """Create tool node optimized for the configured model from environment.""" | |
| config = ToolConfig( | |
| enable_advanced_tools=True, | |
| max_tools=20, | |
| model_name=os.getenv("MODEL_NAME", "gpt-5-mini") | |
| ) | |
| return create_tool_node(config) | |
| # Use the optimized factory for production | |
| tool_node = create_optimized_tool_node() |