Spaces:
Sleeping
Sleeping
File size: 3,565 Bytes
6bd3e57 e252f82 6bd3e57 e252f82 6bd3e57 e252f82 6bd3e57 e252f82 6bd3e57 c7abf8d 6bd3e57 1b8e582 6bd3e57 e252f82 c7abf8d e252f82 6bd3e57 e252f82 6bd3e57 e252f82 6bd3e57 c7abf8d e252f82 c7abf8d 6bd3e57 e252f82 6bd3e57 c7abf8d e252f82 c7abf8d 1b8e582 6bd3e57 c7abf8d e252f82 c7abf8d e252f82 6bd3e57 e252f82 c7abf8d 6bd3e57 e252f82 | 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 | """
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")
@validator('model_name')
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
@validator('tools')
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() |