Your production AI agent starts with three simple tools: search_docs, get_user_profile, and create_support_ticket. At three tools, function calling feels like black magic. Accuracy is practically 100%. The model picks the exact tool you want, formats the arguments flawlessly, and streams a clean response back to the user.
Then your product manager asks for feature parity with your web dashboard.
Six months later, your agent's tool declaration array contains 22 separate tools. You have search_docs, search_knowledge_base, query_faq, get_user_profile, get_account_details, fetch_billing_history, cancel_subscription, pause_subscription, issue_refund, and 13 other API integrations.
Suddenly, your agent starts making erratic choices:
- A user asks, "Can I pause my billing for two months?" and the agent calls
cancel_subscription. - A user asks, "Where is my invoice?" and the agent passes a query string into
search_docsinstead of passing the user ID intofetch_billing_history. - Input token usage doubles on every user turn because you are sending 4,500 tokens of raw JSON Schema on every single message.
- Time-To-First-Token (TTFT) degrades by 600 milliseconds because the model spends precious attention heads sorting through overlapping tool descriptions.
This phenomenon is known as tool saturation decay. According to evaluations on benchmarks like the Berkeley Function Calling Leaderboard (BFCL) and real-world evaluation sets, function calling accuracy drops non-linearly once an LLM context contains more than 8–10 tool schemas. When tools share overlapping semantic domain boundaries, the failure rate spikes even faster.
Exposing 20+ tools directly to a single LLM prompt is an architectural anti-pattern. Here is how modern production AI engineering teams keep tool selection accurate, fast, and deterministic as agent capabilities scale.
📋 Table of Contents
- The Mechanics of Tool Saturation Decay
- Pattern 1: Tool-RAG (Dynamic Vector-Based Tool Retrieval)
- Pattern 2: The Router-Worker (Hierarchical Sub-Agent) Architecture
- Pattern 3: Two-Phase Deliberative Selection
- Pattern 4: Deterministic State-Machine Tool Masking
- Pattern 5: Schema Disambiguation & Negative Descriptions
- Pattern 6: Dynamic In-Context Few-Shot Example Injection
- Pattern 7: Model Context Protocol (MCP) Domain Scoping
- Pattern Comparison & Trade-Off Matrix
- Building a Production Tool Selection Pipeline
The Mechanics of Tool Saturation Decay
Why do LLMs struggle when faced with 20 or more tool choices?
Function calling isn't a native hardware instruction inside the model. When you pass a tools list to Anthropic, OpenAI, or an open-weights model, the framework converts those JSON Schemas into a system prompt section formatted in a specific syntax (such as XML tags or JSON blocks).
System Prompt
├── System Instructions
├── Conversation Context
└── Tool Definitions (20 Tools = ~4,000 Tokens)
├── Tool 1: Schema + Parameters + Descriptions
├── Tool 2: Schema + Parameters + Descriptions
└── ...
When an LLM processes this context, its self-attention mechanism must calculate query-key-value projections across every token in every tool definition. As the tool payload grows:
- Semantic Collision: Tools with overlapping descriptions (e.g.,
get_user_accountvsget_customer_profile) create closely aligned vector representations in key-value space. The model's soft-max probability distribution across potential function tokens flattens out, leading to unpredictable sampling errors. - Attention Dilution: The "Lost in the Middle" phenomenon applies to tool declarations. Tool definitions placed in the middle of a massive array of schemas receive less attention weight than tools at the beginning or end of the system block.
- Payload Bloat: Parameters with nested objects, enums, and deep property trees consume context context windows at an alarming rate, ballooning cloud API costs without offering any real benefit for simple user requests.
To fix this, you must shift from a static tool injection model (sending all tools every turn) to a dynamic context strategy (providing only the precise tools required for the current execution step).
Pattern 1: Tool-RAG (Dynamic Vector-Based Tool Retrieval)
Scenario
Your agent platform supports 60 enterprise integrations across HR, accounting, CRM, and developer tools. Sending all schemas simultaneously breaks model limits and costs thousands of dollars per day in raw input tokens.
Why it matters
The user query "Where is my W2 tax form?" requires HR tools (fetch_tax_documents), but has zero need for developer tools (create_jira_issue) or CRM tools (update_salesforce_lead).
Solution
Treat your tool schemas as documents in a vector database. When a user sends a prompt, query the vector index using semantic search to retrieve only the top $K$ relevant tool definitions (typically $K=3$ to $5$). Inject only those $K$ tools into the current API request payload.
User Query: "Find my W2 form"
│
▼
┌──────────────┐
│ Vector Search│ ──► Index of 60 Tool Embeddings
└──────┬───────┘
│ Top-3 relevant tools
▼
┌──────────────┐
│ LLM Engine │ ──► Receives ONLY: [fetch_tax_docs, search_hr_portal, get_employee_file]
└──────────────┘
Here is a production-ready Python implementation using Pydantic, OpenAI embeddings, and standard cosine similarity to construct a dynamic ToolRegistry:
import json
import numpy as np
from typing import List, Dict, Any
from pydantic import BaseModel, Field
class ToolDefinition(BaseModel):
name: str
description: str
parameters: Dict[str, Any]
embedding: List[float] = Field(default_factory=list)
def to_openai_schema(self) -> Dict[str, Any]:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
}
}
class DynamicToolRegistry:
def __init__(self, embedding_client):
self.client = embedding_client
self.tools: Dict[str, ToolDefinition] = {}
def register_tool(self, name: str, description: str, parameters: Dict[str, Any]):
# Combine name and description to create a rich semantic vector representation
semantic_text = f"Tool Name: {name}. Purpose: {description}"
response = self.client.embeddings.create(
input=semantic_text,
model="text-embedding-3-small"
)
embedding = response.data[0].embedding
tool_def = ToolDefinition(
name=name,
description=description,
parameters=parameters,
embedding=embedding
)
self.tools[name] = tool_def
def retrieve_tools(self, query: str, top_k: int = 3) -> List[Dict[str, Any]]:
if not self.tools:
return []
# Embed the incoming user intent
query_response = self.client.embeddings.create(
input=query,
model="text-embedding-3-small"
)
query_vec = np.array(query_response.data[0].embedding)
# Compute cosine similarities across all registered tools
tool_names = list(self.tools.keys())
matrix = np.array([self.tools[name].embedding for name in tool_names])
# Cosine similarity calculation
scores = np.dot(matrix, query_vec) / (
np.linalg.norm(matrix, axis=1) * np.linalg.norm(query_vec)
)
# Grab top-k indices
top_indices = np.argsort(scores)[::-1][:top_k]
return [self.tools[tool_names[idx]].to_openai_schema() for idx in top_indices]
Why this works
By filtering the array down from 60 tools to the top 3 semantically aligned schemas, you eliminate 95% of the noise in the LLM's system block. Attention heads focus entirely on distinguishing between the 3 relevant choices rather than swimming in dozens of irrelevant options.
⚠️ Gotcha: Plain vector embeddings struggle with short function names or industry acronyms. Always generate embeddings using a structured string format that combines the tool name, its primary use case, key parameter descriptions, and explicit alias keywords (e.g.,
"W2, tax form, income statement, earnings").
Pattern 2: The Router-Worker (Hierarchical Sub-Agent) Architecture
Scenario
Your AI assistant handles enterprise IT operations. It manages database administrative tasks, cloud deployments, user access management, and network monitoring.
Why it matters
A single monolithic LLM prompt containing DB tools, Kubernetes tools, Okta tools, and Datadog tools will confuse parameters across operational domains. It also risks destructive outcomes—like triggering an AWS infrastructure teardown when the user meant to reset a local cluster pod.
Solution
Decouple the architecture. Use a top-level Router Agent whose only job is to classify the request and hand off control to a specialized Domain Worker Agent. The top-level agent has zero system execution tools—it only has access to "handoff" tools.
┌──────────────────┐
│ Router Agent │
│ (No Exec Tools) │
└────────┬─────────┘
│
┌────────────────────┼────────────────────┐
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ DB Worker Agent │ │ K8s Worker Agent│ │ Auth Worker │
│ (3 DB Tools) │ │ (4 Pod Tools) │ │ (3 Okta Tools) │
└─────────────────┘ └─────────────────┘ └─────────────────┘
Here is how you express this cleanly in Python using structured outputs:
from typing import Literal, Optional
from pydantic import BaseModel, Field
class DomainHandoff(BaseModel):
"""Routes the incoming user issue to a specialized domain worker agent."""
target_domain: Literal["database", "kubernetes", "identity_management", "general_inquiry"] = Field(
description="The administrative domain best suited to handle the request."
)
reasoning: str = Field(
description="Short architectural justification for choosing this domain worker."
)
extracted_context: str = Field(
description="Key entities extracted from the conversation required by the worker."
)
class KubernetesWorker:
def __init__(self):
# Worker contains strictly relevant k8s tools
self.tools = [
{"type": "function", "function": {"name": "get_pod_logs", ...}},
{"type": "function", "function": {"name": "restart_deployment", ...}},
{"type": "function", "function": {"name": "scale_replicas", ...}},
]
def execute(self, user_prompt: str, context: str):
# Invokes underlying model with only its 3 dedicated tools exposed
pass
def route_and_dispatch(user_query: str, client):
# System prompt instructs router to evaluate macro intent
completion = client.beta.chat.completions.parse(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are a master router. Analyze user requests and assign them to specialized agents."},
{"role": "user", "content": user_query}
],
response_format=DomainHandoff,
)
route = completion.choices[0].message.parsed
if route.target_domain == "kubernetes":
worker = KubernetesWorker()
return worker.execute(user_query, route.extracted_context)
# Handle other domain routes...
Why this works
The router agent works with a minimal cognitive payload: a single enum choice. Once delegated, the specialized domain worker receives a clean context window equipped with only the 3-4 operational tools specific to its scope. Tool selection accuracy within the worker reaches near perfection because the domain context is narrowly scoped.
💡 Practical note: Router models don't require high-end, high-parameter LLMs. You can use lighter, faster, cheaper models (like
gpt-4o-mini,claude-3-5-haiku, or fine-tunedLlama-3-8B) for the routing step, reserving your largest flagship models for complex parameter extraction inside the worker agents.
Pattern 3: Two-Phase Deliberative Selection
Scenario
You have 18 financial analysis tools, including calculate_compound_interest, generate_amortization_schedule, and project_portfolio_yield. The JSON Schemas for these mathematical tools are massive—filled with deeply nested arrays, formatting enums, and required calculations.
Why it matters
When an LLM must simultaneously identify the right tool and construct complex JSON arguments, it often experiences reasoning strain. It misformats arguments or chooses the wrong tool because its generation context is overwhelmed by syntax construction.
Solution
Split execution into two dedicated LLM calls:
- Phase 1 (Intent Selection): The model gets a stripped-down string list of tool names and micro-descriptions. It outputs only the names of the tools it needs.
- Phase 2 (Argument Formatting): You hydrate only the chosen tools into their full, detailed JSON Schema specifications and pass them back to the model to extract and validate the arguments.
import json
from typing import List
from pydantic import BaseModel, Field
# Light representation for Phase 1
LIGHTWEIGHT_TOOL_INDEX = """
1. calculate_compound_interest: Calculates investment growth over time given principal and rates.
2. generate_amortization_schedule: Generates monthly payment breakdowns for fixed loans.
3. project_portfolio_yield: Computes weighted return metrics for multi-asset portfolios.
"""
class SelectedTools(BaseModel):
selected_tool_names: List[str] = Field(
description="Array of tool names required to answer the prompt. Max 2."
)
def two_phase_tool_execution(user_query: str, full_tool_registry: dict, client):
# --- PHASE 1: Lightweight Tool Selection ---
phase1_prompt = f"""
You are a tool router. Select the minimal set of tool names needed to fulfill the request.
Available Tools:
{LIGHTWEIGHT_TOOL_INDEX}
Request: {user_query}
"""
selection_completion = client.beta.chat.completions.parse(
model="gpt-4o-mini",
messages=[{"role": "user", "content": phase1_prompt}],
response_format=SelectedTools,
)
chosen_names = selection_completion.choices[0].message.parsed.selected_tool_names
# --- PHASE 2: Hydrate Full Schemas for Execution ---
hydrated_schemas = [
full_tool_registry[name] for name in chosen_names if name in full_tool_registry
]
# Run the main generation call with ONLY the selected full schemas
phase2_response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "You are a financial assistant. Call the provided tools with exact parameters."},
{"role": "user", "content": user_query}
],
tools=hydrated_schemas,
tool_choice="auto"
)
return phase2_response.choices[0].message
Why this works
Phase 1 drops token overhead by up to 90% while isolating the selection problem. Phase 2 then operates with complete structural clarity: the model no longer needs to decide which tool to pick among dozens; it simply focuses its attention on mapping user context accurately into the parameters of the designated schema.
Pattern 4: Deterministic State-Machine Tool Masking
Scenario
An e-commerce order assistant can execute actions like search_catalog, add_to_cart, apply_discount_code, enter_shipping_address, and process_payment.
Why it matters
Calling process_payment when the cart is completely empty is a logic error. Allowing the LLM to choose apply_discount_code before the user has authenticated produces useless runtime exceptions. Relying purely on system prompt warnings like "Do not call process_payment unless the cart contains items" fails regularly under adversarial or edge-case user prompts.
Solution
Do not trust the LLM to enforce business process constraints. Implement a State-Driven Tool Masking Engine. Filter the tools list programmatically using deterministic code before sending the request to the LLM API.
State: ANONYMOUS_BROWSER
└── Allowed: [search_catalog, view_product_details, login]
└── Masked out: [add_to_cart, process_payment, view_order_history]
State: CHECKOUT_PENDING_PAYMENT
└── Allowed: [apply_discount_code, enter_shipping_address, process_payment]
└── Masked out: [search_catalog, login]
Here is a middleware pattern for dynamic tool filtering based on session state:
from enum import Enum
from typing import List, Dict, Any
class SessionState(Enum):
UNAUTHENTICATED = "UNAUTHENTICATED"
BROWSING = "BROWSING"
CHECKOUT = "CHECKOUT"
# Define explicit deterministic state transitions
STATE_TOOL_PERMISSIONS: Dict[SessionState, List[str]] = {
SessionState.UNAUTHENTICATED: [
"search_catalog",
"get_product_details",
"authenticate_user"
],
SessionState.BROWSING: [
"search_catalog",
"get_product_details",
"add_item_to_cart",
"view_cart",
"initiate_checkout"
],
SessionState.CHECKOUT: [
"view_cart",
"apply_promo_code",
"set_shipping_address",
"execute_payment"
]
}
class ContextAwareToolFilter:
def __init__(self, full_catalog: Dict[str, Dict[str, Any]]):
self.catalog = full_catalog
def get_allowed_tools(self, current_state: SessionState) -> List[Dict[str, Any]]:
allowed_names = STATE_TOOL_PERMISSIONS.get(current_state, [])
# Build the dynamic payload using strict hard constraints
filtered_tools = []
for name in allowed_names:
if name in self.catalog:
filtered_tools.append(self.catalog[name])
return filtered_tools
Why this works
This design removes structural error entirely. The LLM cannot accidentally select execute_payment during an unauthenticated browsing session because the tool schema does not exist in the API payload for that turn.
🧠 The important part: Deterministic software code should handle state validation, permissions checks, and business rules. LLMs should handle language translation, semantic comprehension, and parameter extraction. Do not force an LLM to do a runtime rule engine's job.
Pattern 5: Schema Disambiguation & Negative Descriptions
Scenario
Your enterprise application contains two tools:
cancel_subscription: Immediately halts recurring billing and revokes system privileges.pause_subscription: Temporarily disables billing while retaining user data for up to 90 days.
Users frequently write ambiguously: "I want to stop my subscription for a bit."
Why it matters
Standard docstrings often use positive statements (e.g., "Cancels a subscription" vs "Pauses a subscription"). When user intent is ambiguous, the model defaults to whichever tool representation is nearest in its pre-trained semantic memory—often leading to destructive mistakes like full subscription cancellations.
Solution
Engineered schemas must feature Disambiguation Sections and explicit Negative Definitions ("When NOT to use this tool"). Additionally, enforce parameter verification through typed sub-fields.
Below is an improved schema definition using Pydantic V2 docstrings:
from pydantic import BaseModel, Field
from typing import Literal
class CancelSubscriptionSchema(BaseModel):
"""
PERMANENTLY terminates user billing and immediately revokes platform access.
WHEN TO USE:
- User explicitly demands permanent account termination, deletion, or full cancellation.
WHEN NOT TO USE:
- DO NOT USE if the user mentions 'taking a break', 'holding', 'pausing', or 'temporary' stops.
In those cases, you MUST call pause_subscription instead.
- DO NOT USE for refund requests. Use issue_refund.
"""
user_id: str = Field(description="The unique system identifier for the customer.")
cancellation_reason: str = Field(description="Detailed reason given by user for leaving.")
confirm_immediate_data_deletion: bool = Field(
description="Must be True ONLY if user explicitly requested full data erasure."
)
class PauseSubscriptionSchema(BaseModel):
"""
TEMPORARILY halts monthly billing cycles while safely maintaining account data.
WHEN TO USE:
- User asks to stop billing temporarily, hold their plan, or take a break.
WHEN NOT TO USE:
- DO NOT USE if user requests full account deletion or permanent termination.
"""
user_id: str = Field(description="The unique system identifier for the customer.")
pause_duration_months: int = Field(
default=1,
description="Number of months to hold billing (Allowed range: 1 to 3)."
)
Why this works
By highlighting negative constraints ("WHEN NOT TO USE"), you target the negative space in the LLM's attention mechanism. When the model processes words like "temporary" in the user prompt, the explicit negative directive inside CancelSubscriptionSchema acts as an attention barrier, pushing the probability weight directly toward PauseSubscriptionSchema.
Pattern 6: Dynamic In-Context Few-Shot Example Injection
Scenario
You have tools with subtle input formatting expectations. For example, query_database requires a custom DSL query object, while search_vector_store takes a unstructured natural language string.
Why it matters
The LLM frequently picks the wrong tool because it fails to grasp how user queries map to function signatures based on docstrings alone.
Solution
Pair your tools with dynamic Few-Shot Execution Traces. When a tool is included in the active prompt payload, automatically inject 1–2 real-world user intent mapping examples directly into system messages or tool documentation notes.
Here is an architectural pattern for pairing tool schemas with executable demonstration traces:
from typing import List, Dict, Any
TOOL_DEMONSTRATION_LIBRARY = {
"query_relational_db": [
{
"user_query": "Show me total enterprise revenue from Q3 for accounts over 500 seats.",
"tool_call": {
"name": "query_relational_db",
"arguments": {
"query_type": "AGGREGATE_SUM",
"target_metric": "revenue",
"filters": [
{"field": "quarter", "op": "EQ", "value": "Q3"},
{"field": "seat_count", "op": "GT", "value": 500}
]
}
}
}
],
"search_unstructured_docs": [
{
"user_query": "What are our official security policies around third-party vendors?",
"tool_call": {
"name": "search_unstructured_docs",
"arguments": {
"search_phrase": "third party vendor security compliance rules",
"department": "INFOSEC"
}
}
}
]
}
def build_few_shot_system_context(active_tool_names: List[str]) -> str:
"""Generates dynamic example pairs based exclusively on currently selected tools."""
few_shot_block = "### TOOL SELECTION AND EXAMPLES\n"
for tool_name in active_tool_names:
if tool_name in TOOL_DEMONSTRATION_LIBRARY:
examples = TOOL_DEMONSTRATION_LIBRARY[tool_name]
for ex in examples:
few_shot_block += f"\nUser Goal: \"{ex['user_query']}\"\n"
few_shot_block += f"Correct Tool Selection: {ex['tool_call']['name']}\n"
few_shot_block += f"Arguments Payload: {json.dumps(ex['tool_call']['arguments'])}\n"
return few_shot_block
Why this works
LLMs demonstrate significant performance gains under few-shot prompting compared to zero-shot instructions. Providing explicit user-intent-to-tool-payload examples grounds the model's token predictions in concrete execution patterns rather than relying on abstract docstrings.
🚨 Production warning: Keep your few-shot examples minimal. Injecting 10 examples across 5 tools will quickly consume thousands of tokens, undermining the context optimizations you set out to achieve. Limit your injection to 1 perfect example per active tool.
Pattern 7: Model Context Protocol (MCP) Domain Scoping
Scenario
You are building an AI agent system using the Model Context Protocol (MCP). Your architecture connects to multiple independent MCP tool servers: a GitHub server, a PostgreSQL server, a Slack server, and a Customer CRM server. Combined, these servers expose over 40 tools.
Why it matters
MCP makes connecting external tools effortless, but attaching multiple servers simultaneously floods the agent with vast tool payloads, causing performance degradation and high latency.
Solution
Implement MCP Client-Side Tool Scoping. Instead of exposing every discovered tool from an MCP server connection, pass tool definitions through a context-aware server filter middleware.
┌─────────────────────────────────────────────────────────┐
│ MCP Agent Core │
└───────────────────────────┬─────────────────────────────┘
│
┌─────────────┴─────────────┐
▼ ▼
┌───────────────────────────┐ ┌───────────────────────────┐
│ GitHub MCP Client │ │ PostgreSQL MCP Client │
│ (Filter: Read Only) │ │ (Filter: Customer DB) │
└─────────────┬─────────────┘ └─────────────┬─────────────┘
│ 3 Tools Exposed │ 2 Tools Exposed
▼ ▼
┌───────────────────────────┐ ┌───────────────────────────┐
│ [search_issues, │ │ [query_customer_by_id, │
│ get_pull_request, │ │ get_active_subscriptions]│
│ list_repo_commits] │ │ │
└───────────────────────────┘ └───────────────────────────┘
Here is how to wrap an MCP tool discovery loop with selective dynamic filters using the official Python MCP SDK paradigms:
from mcp import ClientSession
from typing import List, Dict, Any
class ScopedMCPManager:
def __init__(self):
self.sessions: Dict[str, ClientSession] = {}
async def get_scoped_tools_for_intent(
self,
active_domains: List[str]
) -> List[Dict[str, Any]]:
"""
Polls connected MCP servers, but ONLY imports schemas
matching allowed active domains for the session.
"""
active_mcp_tools = []
for domain, session in self.sessions.items():
# Skip entire MCP server connections if domain is not active
if domain not in active_domains:
continue
# Fetch remote tool list from active server
result = await session.list_tools()
for tool in result.tools:
# Transform native MCP tool schema to LLM payload schema
active_mcp_tools.append({
"type": "function",
"function": {
"name": f"{domain}__{tool.name}", # Namespaced execution key
"description": tool.description,
"parameters": tool.inputSchema,
}
})
return active_mcp_tools
Why this works
Namespacing (github__get_pull_request, postgres__query_customer) prevents schema collisions across independent systems. Filtering at the MCP client layer stops irrelevant server capabilities from reaching the prompt context entirely.
Pattern Comparison & Trade-Off Matrix
Choosing the right approach depends on your scale, latency budget, and system complexity. The following breakdown shows how these key architectural patterns compare:
| Pattern | Accuracy Impact | Implementation Complexity | Latency Impact | Input Token Overhead | Ideal Tool Count |
|---|---|---|---|---|---|
| Tool-RAG (Vector Search) | High | Medium | +80ms (Embedding call) | Very Low (-80%) | 30 to 100+ Tools |
| Router-Worker Agents | Extremely High | High | +400ms (Secondary LLM turn) | Very Low (-75%) | 15 to 50+ Tools |
| Two-Phase Deliberative | High | Low | +250ms (Extra JSON parse step) | Low (-60%) | 10 to 30 Tools |
| State-Machine Masking | Maximum (100% Deterministic) | Low | 0ms (Local code execution) | Variable | 5 to 50 Tools |
| Schema Disambiguation | Medium | Very Low | 0ms | Minimal Increase (+5%) | 1 to 15 Tools |
| Few-Shot Example Injection | High | Low | 0ms | Moderate Increase (+20%) | 5 to 20 Tools |
| Scoped MCP Discovery | High | Medium | +20ms (In-memory filtering) | Low (-70%) | 20 to 100+ Tools |
Building a Production Tool Selection Pipeline
In high-reliability enterprise systems, you rarely rely on a single pattern. Production platforms chain these approaches into a unified pipeline:
User Request
│
▼
┌────────────────────────────────────┐
│ STEP 1: Deterministic State Masking│
│ (Drops unauthorized/invalid tools) │
└──────────────────┬─────────────────┘
│
▼
┌────────────────────────────────────┐
│ STEP 2: Intent Domain Routing │
│ (Selects designated Worker Agent) │
└──────────────────┬─────────────────┘
│
▼
┌────────────────────────────────────┐
│ STEP 3: Vector Tool-RAG Retrieval │
│ (Retrieves Top 3-5 Schema Matches) │
└──────────────────┬─────────────────┘
│
▼
┌────────────────────────────────────┐
│ STEP 4: Injection & Disambiguation │
│ (Adds Negative Rules & Few-Shots) │
└──────────────────┬─────────────────┘
│
▼
LLM Function Execution
End-to-End Implementation Example
Here is a complete production pipeline incorporating dynamic state masking, vector retrieval, and schema safety:
import os
import json
import numpy as np
from typing import List, Dict, Any
from pydantic import BaseModel, Field
from openai import OpenAI
# Initialize client
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
# 1. Define Master Catalog with Explicit Docstrings & Metadata
TOOL_CATALOG = {
"get_customer_balance": {
"domain": "billing",
"description": "Fetches current account balance. DO NOT use for invoice history download.",
"parameters": {
"type": "object",
"properties": {"account_id": {"type": "string"}},
"required": ["account_id"]
}
},
"issue_store_credit": {
"domain": "billing",
"description": "Issues promotional account credit. DO NOT use for credit card refunds.",
"parameters": {
"type": "object",
"properties": {
"account_id": {"type": "string"},
"amount_usd": {"type": "number"}
},
"required": ["account_id", "amount_usd"]
}
},
"restart_server_instance": {
"domain": "infrastructure",
"description": "Reboots a virtual machine. Use ONLY when server is unresponsive.",
"parameters": {
"type": "object",
"properties": {"instance_id": {"type": "string"}},
"required": ["instance_id"]
}
}
}
class PipelineAgent:
def __init__(self, catalog: Dict[str, Any]):
self.catalog = catalog
self.tool_embeddings: Dict[str, List[float]] = {}
self._precompute_catalog_embeddings()
def _precompute_catalog_embeddings(self):
"""Precompute embeddings for all tools during initialization."""
for name, spec in self.catalog.items():
text = f"Tool: {name}. Description: {spec['description']}"
res = client.embeddings.create(input=text, model="text-embedding-3-small")
self.tool_embeddings[name] = res.data[0].embedding
def _mask_by_state(self, user_role: str) -> List[str]:
"""STEP 1: Hard deterministic masking based on user permissions."""
if user_role != "admin":
# Non-admins cannot see infrastructure tools
return [k for k, v in self.catalog.items() if v["domain"] != "infrastructure"]
return list(self.catalog.keys())
def _retrieve_top_k(self, query: str, candidate_names: List[str], k: int = 2) -> List[Dict[str, Any]]:
"""STEP 2: Vector search over state-filtered candidates."""
query_res = client.embeddings.create(input=query, model="text-embedding-3-small")
q_vec = np.array(query_res.data[0].embedding)
scores = {}
for name in candidate_names:
t_vec = np.array(self.tool_embeddings[name])
scores[name] = np.dot(q_vec, t_vec) / (np.linalg.norm(q_vec) * np.linalg.norm(t_vec))
# Sort and take top k
top_names = sorted(scores, key=scores.get, reverse=True)[:k]
# Format for OpenAI Tools API Payload
formatted_tools = []
for name in top_names:
formatted_tools.append({
"type": "function",
"function": {
"name": name,
"description": self.catalog[name]["description"],
"parameters": self.catalog[name]["parameters"]
}
})
return formatted_tools
def run(self, user_prompt: str, user_role: str):
# Step 1: Programmatically filter out prohibited tools
allowed_tool_names = self._mask_by_state(user_role)
# Step 2: Use vector similarity to retrieve relevant tools
selected_tools = self._retrieve_top_k(user_prompt, allowed_tool_names, k=2)
# Step 3: Run LLM execution turn with optimized context window
messages = [
{"role": "system", "content": "You are a specialized support engineer. Use provided tools when necessary."},
{"role": "user", "content": user_prompt}
]
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
tools=selected_tools,
tool_choice="auto"
)
return response.choices[0].message
# --- Execution Demo ---
agent = PipelineAgent(TOOL_CATALOG)
# User asks for balance; system restricts tools to billing domain, selecting only relevant tools
result = agent.run(
user_prompt="I need to check how much money is left in account_9921",
user_role="standard_user"
)
if result.tool_calls:
for call in result.tool_calls:
print(f"Selected Tool: {call.function.name}")
print(f"Arguments: {call.function.arguments}")
else:
print(f"Response: {result.content}")
Summary & Strategic Choice Framework
Keeping tool selection accurate boils down to a single engineering principle: Never force an LLM to make a decision it shouldn't have to make.
- If a tool is invalid for a given user role or workflow step, drop it programmatically using State Masking.
- If your tool collection spans multiple domains, divide responsibility using a Router-Worker architecture.
- If you have dozens of administrative APIs, narrow them down on every turn using Vector Search Tool Retrieval.
- If two tools look similar, rewrite their docstrings using explicit Negative Constraints ("WHEN NOT TO USE").
By shifting tool integration from static prompt declarations into a dynamic context management pipeline, you protect accuracy, minimize latency, and keep costs under control—even as your agent scales to dozens of tools.