Tool calling w produkcji — niezawodne agenty AI z funkcjami zewnętrznymi
Tool calling to mechanizm, który zmienia LLM z chatbota w agenta zdolnego do działania — wywoływania API, query baz danych, uruchamiania kodu. Jak definiować narzędzia, obsługiwać błędy, wywoływać równolegle i monitorować w produkcji — z pełnym kodem.
Tool calling transformuje LLM z chatbota w agenta zdolnego do działania. Zamiast tylko odpowiadać na pytania, model może wywołać API, wykonać query do bazy danych, uruchomić kalkulację lub sprawdzić stan zewnętrznego systemu — i dopiero na podstawie tych danych sformułować odpowiedź. To mechanizm stojący za "inteligencją" nowoczesnych asystentów AI: zamiast halucynować brakujące fakty, model prosi o dane, które faktycznie istnieją.
Jak dokładnie działa? Model generuje JSON z nazwą funkcji i argumentami — ale nie wykonuje jej. Ty otrzymujesz ten JSON, wykonujesz funkcję własnym kodem (z odpowiednimi uprawnieniami, walidacją, logowaniem) i zwracasz wynik do modelu. Model widzi wynik i formułuje finalną odpowiedź. Dlatego tool calling jest bezpieczny — LLM nigdy nie ma bezpośredniego dostępu do Twoich systemów.
/// MECHANIZM TOOL CALLING
Cykl: User → LLM → Tool → LLM → Response
Kluczowe zrozumienie: model nie "decyduje co wykonać" tylko raz. W złożonych agentach pętla (LLM → tool call → result → LLM) może się powtórzyć kilka razy. Ustaw limit iteracji (zwykle 10), żeby uniknąć nieskończonych pętli gdy model się "zapętla" między narzędziami.
Definicja narzędzi — JSON Schema i walidacja Pydantic
Jakość opisów narzędzi ma bezpośredni wpływ na to, czy model poprawnie je wywoła. Zbyt lakoniczny opis → model nie wie kiedy użyć narzędzia lub podaje złe argumenty. Zbyt ogólny → model używa narzędzia wszędzie, nawet gdy nie powinien. Pydantic eliminuje boilerplate przy definiowaniu JSON Schema i waliduje argumenty przed przekazaniem do handlera.
from pydantic import BaseModel, Fieldfrom openai import OpenAIimport jsonclient = OpenAI()# ─── Schematy wejścia (Pydantic → JSON Schema automatycznie) ─────────────────class SearchProductsInput(BaseModel): query: str = Field(description="Fraza wyszukiwania produktów w katalogu") max_results: int = Field(default=5, ge=1, le=20, description="Maks. liczba wyników") category: str | None = Field(default=None, description="Filtr kategorii (opcjonalny)")class GetOrderStatusInput(BaseModel): order_id: str = Field(description="ID zamówienia w formacie ORD-XXXXXX")# ─── Rejestr narzędzi (schemat + opis dla modelu) ────────────────────────────TOOLS = [ { "type": "function", "function": { "name": "search_products", "description": "Wyszukuje produkty w katalogu. Używaj gdy użytkownik pyta o dostępność lub porównanie produktów. NIE używaj do sprawdzania zamówień.", "parameters": SearchProductsInput.model_json_schema(), }, }, { "type": "function", "function": { "name": "get_order_status", "description": "Pobiera status zamówienia po ID. Używaj tylko gdy użytkownik podał numer zamówienia (ORD-XXXXXX).", "parameters": GetOrderStatusInput.model_json_schema(), }, },]# ─── Implementacje narzędzi (Twój kod backendowy) ────────────────────────────def search_products(query: str, max_results: int = 5, category: str | None = None) -> list[dict]: # Tu: Elasticsearch / Algolia / SQL query return [{"id": "P001", "name": f"Produkt: {query}", "price": 99.99, "stock": 12}]def get_order_status(order_id: str) -> dict: # Tu: query do bazy zamówień return {"order_id": order_id, "status": "shipped", "delivery": "2026-06-07"}TOOL_HANDLERS: dict = {"search_products": search_products, "get_order_status": get_order_status}# ─── Pętla agenta (max 10 iteracji) ──────────────────────────────────────────def run_agent(user_message: str) -> str: messages = [{"role": "user", "content": user_message}] for _ in range(10): resp = client.chat.completions.create( model="gpt-4o", messages=messages, tools=TOOLS, tool_choice="auto" ) msg = resp.choices[0].message if not msg.tool_calls: return msg.content messages.append(msg) for tc in msg.tool_calls: result = TOOL_HANDLERS[tc.function.name](**json.loads(tc.function.arguments)) messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result, ensure_ascii=False)}) return "Przekroczono limit iteracji agenta."
Trzy reguły dobrego opisu narzędzia: (1) powiedz kiedy używać i kiedy NIE — model potrzebuje kontrastu żeby dobrze decydować; (2) opisz format argumentów (np. "ORD-XXXXXX") — redukuje błędy parsowania; (3) jeden tool = jedna odpowiedzialność — nie łącz "szukaj i zamów" w jedną funkcję.
Wywołania równoległe — gdy model potrzebuje kilku narzędzi naraz
GPT-4o i Claude mogą w jednej odpowiedzi zwrócić listę wielu tool_calls. Jeśli wykonujesz je sekwencyjnie (jeden po drugim), tracisz czas bez potrzeby — niezależne wywołania można wykonać równolegle z asyncio.gather. Różnica przy 3 narzędziach o latency 500ms każde: sequential = 1.5s, parallel = ~500ms.
/// WZORCE WYWOŁAŃ NARZĘDZI
3 główne wzorce tool calling
import asyncioimport jsonfrom openai import AsyncOpenAIclient = AsyncOpenAI()async def execute_tool_async(tool_call, handlers: dict) -> dict: handler = handlers[tool_call.function.name] args = json.loads(tool_call.function.arguments) # Jeśli handler async (httpx, asyncpg) — await; inaczej run_in_executor if asyncio.iscoroutinefunction(handler): result = await handler(**args) else: loop = asyncio.get_running_loop() result = await loop.run_in_executor(None, lambda: handler(**args)) return {"role": "tool", "tool_call_id": tool_call.id, "content": json.dumps(result, ensure_ascii=False)}async def run_agent_parallel(user_message: str, tools: list, handlers: dict) -> str: messages = [{"role": "user", "content": user_message}] for _ in range(10): resp = await client.chat.completions.create( model="gpt-4o", messages=messages, tools=tools, tool_choice="auto" ) msg = resp.choices[0].message if not msg.tool_calls: return msg.content messages.append(msg) tool_results = await asyncio.gather( *[execute_tool_async(tc, handlers) for tc in msg.tool_calls] ) messages.extend(tool_results) return "Przekroczono limit iteracji."
Kiedy model decyduje o równoległości: model sam wykrywa niezależne zapytania. Możesz wymusić użycie narzędzi przez `tool_choice="required"`, ale nie możesz zmusić modelu do wywołania konkretnej pary — tylko kontrolujesz czy narzędzia są dostępne w ogóle.
Obsługa błędów — co gdy narzędzie zawodzi
Błędy w narzędziach dzielą się na trzy typy: retryable (chwilowy problem z siecią, timeout — warto ponowić z backoffem), permanent (złe argumenty, brak uprawnień — nie ponawiaj, zwróć czytelny komunikat dla modelu) i unexpected (bug w kodzie — loguj, zwróć ogólny błąd). Kluczowe: zawsze zwracaj coś do modelu przez pole `tool` — nigdy nie przerywaj pętli bez wyniku.
import jsonimport timeimport logginglogger = logging.getLogger(__name__)class ToolError(Exception): def __init__(self, message: str, retryable: bool = False): self.retryable = retryable super().__init__(message)def execute_with_retry(tool_call, handlers: dict, max_retries: int = 2) -> str: name = tool_call.function.name args = json.loads(tool_call.function.arguments) handler = handlers.get(name) if handler is None: return json.dumps({"error": f"Narzędzie '{name}' nie istnieje.", "code": "UNKNOWN_TOOL"}) for attempt in range(max_retries + 1): try: result = handler(**args) return json.dumps(result, ensure_ascii=False) except ToolError as e: if not e.retryable or attempt == max_retries: logger.error("Tool %s failed: %s", name, e) return json.dumps({"error": str(e), "code": "TOOL_ERROR"}) time.sleep(2 ** attempt) except Exception as e: logger.exception("Unexpected error in tool %s", name) return json.dumps({"error": "Wewnętrzny błąd narzędzia.", "code": "INTERNAL_ERROR"}) return json.dumps({"error": "Przekroczono limit prób.", "code": "MAX_RETRIES"})def run_agent_safe(messages: list, tools: list, handlers: dict, llm) -> str: for _ in range(10): resp = llm.chat.completions.create(model="gpt-4o", messages=messages, tools=tools, tool_choice="auto") msg = resp.choices[0].message if not msg.tool_calls: return msg.content messages.append(msg) for tc in msg.tool_calls: result = execute_with_retry(tc, handlers) messages.append({"role": "tool", "tool_call_id": tc.id, "content": result}) return "Przekroczono limit iteracji agenta."
Ważne: gdy narzędzie zwróci błąd jako JSON z polem "error" — model zazwyczaj sam zdecyduje co zrobić: spróbuje z innymi argumentami, poprosi o wyjaśnienie lub przyzna że nie może wykonać zadania. Zaufaj mu, daj mu czytelny komunikat i nie wymuszaj zachowania w kodzie pętli.
Monitoring wywołań narzędzi w produkcji
Bez monitoringu nie wiesz które narzędzia są najwolniejsze, które najczęściej zawodzą i ile kosztuje sesja agenta w tokenach. Loguj każde wywołanie: nazwa narzędzia, latency, rozmiar argumentów i wyników, sukces/błąd. To wystarczy do identyfikacji bottlenecków i alertów.
import timeimport jsonimport loggingfrom dataclasses import dataclasslogger = logging.getLogger(__name__)@dataclassclass ToolMetric: tool_name: str success: bool latency_ms: float args_bytes: int result_bytes: int error: str | None = Nonedef monitored_call(tool_call, handlers: dict) -> tuple[str, ToolMetric]: name = tool_call.function.name args_json = tool_call.function.arguments t0 = time.perf_counter() try: result = handlers[name](**json.loads(args_json)) result_json = json.dumps(result, ensure_ascii=False) ms = (time.perf_counter() - t0) * 1000 metric = ToolMetric(name, True, round(ms, 1), len(args_json.encode()), len(result_json.encode())) logger.info("tool=%s latency_ms=%.1f ok", name, ms) return result_json, metric except Exception as e: err_json = json.dumps({"error": str(e)}) ms = (time.perf_counter() - t0) * 1000 metric = ToolMetric(name, False, round(ms, 1), len(args_json.encode()), len(err_json.encode()), str(e)) logger.error("tool=%s latency_ms=%.1f error=%s", name, ms, e) return err_json, metric
Kluczowe metryki: latency p50/p95 per narzędzie (outliers wskazują na problemy z zależnościami), error rate per narzędzie (>5% = problem z argumentami lub zależnością), średnia liczba tool calls per sesja (zbyt wiele = model się "zapętla"), koszt tokenów per sesja.
Bezpieczeństwo — prompt injection i walidacja wejść
Użytkownicy mogą próbować manipulować modelem żeby wywołał narzędzie z niebezpiecznymi argumentami. Dwie warstwy obrony: walidacja argumentów (Pydantic sprawdzi typy i zakresy — model nie może podać ujemnego `max_results` jeśli schemat ma `ge=1`) i autoryzacja w handlerze (sprawdź czy zalogowany użytkownik ma uprawnienia — model nie wie kto jest zalogowany, Ty wiesz). Nigdy nie ufaj argumentom z tool call bez weryfikacji uprawnień.
| Wzorzec | Round trips | Kiedy używać | Uwaga |
|---|---|---|---|
| Single tool | +1 | Proste zadania: pobierz info, sprawdź status | Najprostszy do debugowania |
| Parallel tools | +1 | Niezależne operacje równocześnie | asyncio.gather — oszczędza latency |
| Sequential chain | +N | Wynik A potrzebny do wywołania B | Każde łącze = dodatkowy LLM call |
| Forced choice | +1 | Gwarancja użycia konkretnego narzędzia | tool_choice z nazwą funkcji |
---
Buduję agenty AI z tool calling dla firm — od prostych asystentów z dostępem do bazy produktów po złożone pipeline z wieloma narzędziami, logiką autoryzacji i monitoringiem produkcyjnym. Napisz do mnie — zaczynam od analizy Twoich przypadków użycia i projektu architektury narzędzi.
/// RELATED_RECORDS
Jak AI czyta faktury z maila i wprowadza je do ERP
AI odczytuje fakturę z załącznika e-mail — PDF, skan lub zdjęcie z telefonu — i wprowadza dane bezpośrednio do ERP bez ręcznego przepisywania. Pełna automatyzacja obiegu faktur kosztowych: od skrzynki mailowej do zaksięgowania dokumentu.
Od czego zacząć wdrażanie AI w firmie?
Wdrażanie AI w firmie zaczyna się nie od wyboru narzędzia, lecz od jednego powtarzalnego procesu, który dziś zabiera najwięcej czasu. Dowiedz się jak krok po kroku wybrać, opisać i zautomatyzować ten proces.
Jak zbudować wewnętrzną bazę wiedzy firmy z AI (RAG w praktyce)
Wewnętrzna baza wiedzy oparta na RAG pozwala stworzyć własnego chatbota firmowego, który odpowiada wyłącznie na podstawie dokumentów Twojej firmy — nie domysłów modelu. Bezpieczne, aktualne, precyzyjne AI z pełną kontrolą nad danymi.
Signal received?
Przerwij
Ciszę
Zainicjuj protokół. Nawiąż połączenie. Zbudujmy coś głośnego.