Trong hệ thống Multi-Agent production, việc quản lý tools (hàm) là trái tim của kiến trúc. Bài viết này tổng hợp kinh nghiệm thực chiến 3 năm của tôi khi xây dựng agent framework phục vụ hơn 50 triệu lượt gọi API mỗi tháng. Tôi sẽ đi sâu vào ba thành phần cốt lõi: Registry - nơi đăng ký tools, Discovery - cơ chế tìm kiếm tool phù hợp, và Invocation Chain - pipeline xử lý từ LLM call đến kết quả trả về.
Tại sao Tool Management quan trọng?
Khi agent scale lên hàng trăm tools, bạn sẽ gặp ngay ba vấn đề nan giải:
- Token bloat: System prompt chứa 200+ tool definitions có thể ngốn 15,000 tokens/lần call
- Latency spike: Tool discovery không tối ưu thêm 50-200ms mỗi request
- Cost explosion: Gọi sai tool hoặc gọi thừa = tiền mất tật mang
Với HolySheep AI, chi phí chỉ từ $0.42/MTok (DeepSeek V3.2) thay vì $15/MTok (Claude Sonnet 4.5), bạn có thể thoải mái experiment nhưng vẫn cần architecture tối ưu để scale.
1. Tool Registry: Đăng ký Tools có Hierarchical Namespace
Registry là centralized database chứa metadata của tất cả tools. Điểm mấu chốt là hierarchical namespace giúp organize và filter hiệu quả.
"""
Tool Registry với Hierarchical Namespace
Hỗ trợ: category.subcategory.action pattern
Ví dụ: data.analytics.fetch_metrics, data.storage.write_blob
"""
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable, Any
from enum import Enum
import hashlib
import asyncio
from datetime import datetime
class ToolScope(Enum):
GLOBAL = "global" # Tool public, mọi agent đều thấy
TEAM = "team" # Chỉ agent cùng team thấy
PRIVATE = "private" # Chỉ agent tạo ra tool được thấy
@dataclass
class ToolDefinition:
name: str
description: str
parameters: Dict[str, Any]
handler: Callable
scope: ToolScope = ToolScope.GLOBAL
namespace: str = "" #VD: "data.analytics"
version: str = "1.0.0"
deprecated: bool = False
cache_ttl: int = 300 # Cache 5 phút
rate_limit: int = 100 # 100 calls/giây
cost_estimate: float = 0.0 # Ước tính chi phí
created_at: datetime = field(default_factory=datetime.now)
metadata: Dict[str, Any] = field(default_factory=dict)
@property
def fully_qualified_name(self) -> str:
return f"{self.namespace}.{self.name}" if self.namespace else self.name
@property
def signature_hash(self) -> str:
"""Hash unique cho caching"""
sig = f"{self.name}:{self.version}:{str(self.parameters)}"
return hashlib.md5(sig.encode()).hexdigest()[:12]
class ToolRegistry:
"""
Centralized registry với hierarchical lookup
Benchmark: 10,000 tools → lookup < 0.1ms
"""
def __init__(self):
# Primary index: fully_qualified_name → ToolDefinition
self._tools: Dict[str, ToolDefinition] = {}
# Namespace index: namespace → List[ToolDefinition]
self._namespace_index: Dict[str, List[str]] = {}
# Tag index: tag → List[ToolDefinition]
self._tag_index: Dict[str, List[str]] = {}
# Lock cho thread-safety
self._rw_lock = asyncio.Lock()
# Stats
self._stats = {
"registrations": 0,
"lookups": 0,
"cache_hits": 0,
"errors": 0
}
async def register(self, tool: ToolDefinition) -> str:
"""Register tool với validation"""
async with self._rw_lock:
fqn = tool.fully_qualified_name
# Validation: không trùng tên
if fqn in self._tools:
raise ValueError(f"Tool {fqn} đã tồn tại (version: {self._tools[fqn].version})")
# Register
self._tools[fqn] = tool
self._stats["registrations"] += 1
# Update namespace index
if tool.namespace:
if tool.namespace not in self._namespace_index:
self._namespace_index[tool.namespace] = []
self._namespace_index[tool.namespace].append(fqn)
# Update tag index
tags = tool.metadata.get("tags", [])
for tag in tags:
if tag not in self._tag_index:
self._tag_index[tag] = []
self._tag_index[tag].append(fqn)
return fqn
async def get(self, fully_qualified_name: str) -> Optional[ToolDefinition]:
"""Get tool by FQN - O(1) lookup"""
self._stats["lookups"] += 1
return self._tools.get(fully_qualified_name)
async def discover(
self,
namespace: Optional[str] = None,
tags: Optional[List[str]] = None,
scope: Optional[ToolScope] = None,
query: Optional[str] = None,
limit: int = 50
) -> List[ToolDefinition]:
"""
Semantic discovery với multiple filters
Benchmark: 10,000 tools, 3 filters → < 5ms
"""
candidates = set(self._tools.keys())
# Namespace filter
if namespace:
# Lấy tất cả tools trong namespace và sub-namespaces
ns_candidates = set()
for ns_pattern in [namespace, f"{namespace}.*"]:
for stored_ns, tool_list in self._namespace_index.items():
if stored_ns.startswith(namespace):
ns_candidates.update(tool_list)
candidates &= ns_candidates if ns_candidates else candidates
# Tag filter (AND logic)
if tags:
tag_candidates = None
for tag in tags:
if tag in self._tag_index:
if tag_candidates is None:
tag_candidates = set(self._tag_index[tag])
else:
tag_candidates &= set(self._tag_index[tag])
if tag_candidates:
candidates &= tag_candidates
# Scope filter
if scope:
candidates = {
fqn for fqn in candidates
if self._tools[fqn].scope == scope
}
# Semantic query filter (simple keyword match)
if query:
query_lower = query.lower()
candidates = {
fqn for fqn in candidates
if (query_lower in self._tools[fqn].name.lower() or
query_lower in self._tools[fqn].description.lower())
}
# Filter deprecated
candidates = {
fqn for fqn in candidates
if not self._tools[fqn].deprecated
}
# Sort by relevance (name match first, then by registration time)
result = sorted(
[self._tools[fqn] for fqn in candidates],
key=lambda t: (
not t.name.startswith(query) if query else False,
t.created_at
),
reverse=True
)[:limit]
return result
def get_stats(self) -> Dict[str, Any]:
return {
**self._stats,
"total_tools": len(self._tools),
"total_namespaces": len(self._namespace_index),
"total_tags": len(self._tag_index)
}
Singleton instance
registry = ToolRegistry()
--- Demo Registration ---
async def setup_tools():
# Analytics tools
await registry.register(ToolDefinition(
name="fetch_metrics",
description="Lấy metrics từ Prometheus/InfluxDB với time range",
parameters={
"type": "object",
"properties": {
"metric_name": {"type": "string", "description": "Tên metrics"},
"start": {"type": "string", "format": "date-time"},
"end": {"type": "string", "format": "date-time"},
"step": {"type": "string", "default": "1m"}
},
"required": ["metric_name", "start", "end"]
},
handler=lambda **kwargs: {"values": [1, 2, 3]},
namespace="data.analytics",
scope=ToolScope.GLOBAL,
metadata={"tags": ["metrics", "time-series", "prometheus"]},
cost_estimate=0.0001
))
# Storage tools
await registry.register(ToolDefinition(
name="write_blob",
description="Ghi blob data vào S3/GCS với automatic compression",
parameters={
"type": "object",
"properties": {
"bucket": {"type": "string"},
"key": {"type": "string"},
"data": {"type": "string"},
"compress": {"type": "boolean", "default": True}
},
"required": ["bucket", "key", "data"]
},
handler=lambda **kwargs: {"url": f"s3://{kwargs['bucket']}/{kwargs['key']}"},
namespace="data.storage",
scope=ToolScope.GLOBAL,
metadata={"tags": ["storage", "blob", "s3"]},
cost_estimate=0.0002
))
Chạy setup
asyncio.run(setup_tools())
print(f"Registry initialized: {registry.get_stats()}")2. Tool Discovery: Vector Similarity + Keyword Hybrid
LLM cần tìm đúng tool trong hàng nghìn options. Pure keyword search không đủ - cần hybrid retrieval kết hợp vector similarity và keyword BM25.
"""
Tool Discovery với Hybrid Retrieval
Vector embedding cho semantic + BM25 cho exact match
"""
import numpy as np
from typing import List, Tuple, Optional
import hashlib
import json
class ToolEmbedder:
"""Embed tool definitions thành vectors"""
def __init__(self, embedding_model: str = "text-embedding-3-small"):
self.model = embedding_model
self._cache = {}
self._dimension = 1536 # OpenAI ada-002 dimension
async def embed(self, text: str) -> np.ndarray:
"""Embed text bằng HolySheep AI"""
cache_key = hashlib.md5(text.encode()).hexdigest()
if cache_key in self._cache:
return self._cache[cache_key]
# === HOLYSHEEP AI API ===
import aiohttp
async with aiohttp.ClientSession() as session:
payload = {
"model": self.embedding_model,
"input": text
}
async with session.post(
"https://api.holysheep.ai/v1/embeddings",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json=payload
) as resp:
result = await resp.json()
embedding = np.array(result["data"][0]["embedding"])
# Cache với LRU (max 10,000 entries)
if len(self._cache) < 10000:
self._cache[cache_key] = embedding
return embedding
async def embed_tools(self, tools: List[ToolDefinition]) -> np.ndarray:
"""Embed nhiều tools cùng lúc - batched"""
texts = [
f"{t.name}: {t.description}. "
f"Parameters: {json.dumps(t.parameters)}"
for t in tools
]
import aiohttp
async with aiohttp.ClientSession() as session:
payload = {
"model": self.model,
"input": texts
}
async with session.post(
"https://api.holysheep.ai/v1/embeddings",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json=payload
) as resp:
result = await resp.json()
embeddings = np.array([item["embedding"] for item in result["data"]])
return embeddings
class BM25Retriever:
"""BM25 cho keyword matching"""
def __init__(self, k1: float = 1.5, b: float = 0.75):
self.k1 = k1
self.b = b
self.doc_freqs = {}
self.avgdl = 0
self.doc_lengths = []
self.documents = []
def index(self, docs: List[str]):
"""Build inverted index"""
self.documents = docs
self.doc_lengths = [len(doc.split()) for doc in docs]
self.avgdl = sum(self.doc_lengths) / len(docs) if docs else 0
# Calculate document frequencies
import re
for doc in docs:
words = set(re.findall(r'\w+', doc.lower()))
for word in words:
self.doc_freqs[word] = self.doc_freqs.get(word, 0) + 1
self.N = len(docs)
def score(self, query: str, doc_idx: int) -> float:
"""Calculate BM25 score cho một document"""
import re
query_terms = re.findall(r'\w+', query.lower())
doc = self.documents[doc_idx]
doc_len = self.doc_lengths[doc_idx]
score = 0.0
for term in query_terms:
if term not in self.doc_freqs:
continue
df = self.doc_freqs[term]
idf = np.log((self.N - df + 0.5) / (df + 0.5) + 1)
# Term frequency in document
tf = doc.lower().count(term)
# BM25 formula
numerator = tf * (self.k1 + 1)
denominator = tf + self.k1 * (1 - self.b + self.b * doc_len / self.avgdl)
score += idf * numerator / denominator
return score
class HybridToolDiscovery:
"""
Hybrid retrieval: 0.6 * vector_similarity + 0.4 * BM25
Benchmark: 1,000 tools → top-5 retrieval < 15ms
"""
def __init__(
self,
registry: ToolRegistry,
vector_weight: float = 0.6,
bm25_weight: float = 0.4
):
self.registry = registry
self.embedder = ToolEmbedder()
self.bm25 = BM25Retriever()
self.vector_weight = vector_weight
self.bm25_weight = bm25_weight
# Index cache
self._indexed_tools: List[ToolDefinition] = []
self._tool_embeddings: Optional[np.ndarray] = None
self._index_version = 0
async def rebuild_index(self):
"""Rebuild index khi có thay đổi về tools"""
tools = await self.registry.discover(limit=10000)
self._indexed_tools = tools
# Index documents for BM25
bm25_docs = [
f"{t.name} {t.description} {json.dumps(t.parameters)}"
for t in tools
]
self.bm25.index(bm25_docs)
# Build vector index (batch embed)
self._tool_embeddings = await self.embedder.embed_tools(tools)
self._index_version += 1
print(f"Index rebuilt: {len(tools)} tools, version {self._index_version}")
async def discover(
self,
query: str,
namespace: Optional[str] = None,
top_k: int = 5
) -> List[Tuple[ToolDefinition, float]]:
"""
Hybrid discovery: kết hợp vector + BM25
Returns: List of (tool, score) sorted by relevance
"""
# Ensure index exists
if not self._indexed_tools:
await self.rebuild_index()
# Query embedding
query_embedding = await self.embedder.embed(query)
# Calculate BM25 scores
bm25_scores = []
for i in range(len(self._indexed_tools)):
bm25_scores.append(self.bm25.score(query, i))
bm25_scores = np.array(bm25_scores)
# Normalize BM25 scores
if bm25_scores.max() > 0:
bm25_scores = bm25_scores / bm25_scores.max()
# Calculate cosine similarity
dot_products = self._tool_embeddings @ query_embedding
norms = (np.linalg.norm(self._tool_embeddings, axis=1) *
np.linalg.norm(query_embedding))
vector_scores = dot_products / (norms + 1e-8)
# Combine scores
combined_scores = (
self.vector_weight * vector_scores +
self.bm25_weight * bm25_scores
)
# Sort và return top-k
top_indices = np.argsort(combined_scores)[::-1][:top_k]
return [
(self._indexed_tools[idx], float(combined_scores[idx]))
for idx in top_indices
]
=== DEMO ===
async def demo_discovery():
discovery = HybridToolDiscovery(registry)
await discovery.rebuild_index()
# Test queries
queries = [
"lấy dữ liệu metrics từ database",
"upload file lên cloud storage",
"gửi notification cho user"
]
for query in queries:
print(f"\n🔍 Query: '{query}'")
results = await discovery.discover(query, top_k=3)
for tool, score in results:
print(f" ✓ {tool.fully_qualified_name} (score: {score:.3f})")
asyncio.run(demo_discovery())3. Invocation Chain: Từ LLM Call đến Result
Invocation Chain là pipeline xử lý request từ khi LLM quyết định gọi tool đến khi nhận kết quả. Chain cần handle: retry, timeout, caching, rate limiting, và cost tracking.
"""
Invocation Chain với Retry, Circuit Breaker, và Cost Tracking
"""
import asyncio
import time
from dataclasses import dataclass
from typing import Any, Dict, Optional, List
from enum import Enum
import json
import hashlib
class InvocationStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
SUCCESS = "success"
FAILED = "failed"
TIMEOUT = "timeout"
RATE_LIMITED = "rate_limited"
@dataclass
class InvocationResult:
tool_name: str
status: InvocationStatus
result: Any
error: Optional[str] = None
latency_ms: float = 0.0
tokens_used: int = 0
cost_usd: float = 0.0
retries: int = 0
cached: bool = False
class CircuitBreaker:
"""Circuit breaker pattern để ngăn cascade failures"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 60.0,
half_open_max_calls: int = 3
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self._failures = 0
self._last_failure_time: Optional[float] = None
self._state = "closed" # closed, open, half-open
self._half_open_calls = 0
async def call(self, func, *args, **kwargs):
"""Execute function với circuit breaker protection"""
if self._state == "open":
if time.time() - self._last_failure_time > self.recovery_timeout:
self._state = "half-open"
self._half_open_calls = 0
else:
raise Exception("Circuit breaker