Tworzenie agenta AI od zera w Pythonie: pętla GAME krok po kroku
Stwórz działającego agenta do wywoływania narzędzi w prostym Pythonie z elementami takimi jak cele, działania, pamięć i środowisko, a następnie dodaj rejestr narzędzi, ukrytą iniekcję zależności oraz typizowany wynik.
Ramki agentów takie jak Pydantic AI, AutoGen i Microsoft Agent Framework sprawdzają się dobrze do szybkiego stworzenia prototypu. Jednak po wypróbowaniu kilku z nich zwykle pojawiają się pewne trudności.
- Ekosystem wciąż się rozwija. Każda z tych ramek obejmuje niektóre z potrzebnych funkcji, a pomija inne, natomiast inna rama posiada te elementy, których brakuje pierwszej. To normalne w młodym obszarze, a z upływem czasu większość z nich będzie radzić sobie z typowymi przypadkami użycia.
- Narasta abstrakcja. Ramki te mają na celu obsługę bardzo różnorodnych projektów agentów, dlatego dodają warstwy, które są trudne do zrozumienia i jeszcze trudniejsze do debugowania. LangChain jest typowym przykładem: gdy system zachowuje się niewłaściwie, te warstwy utrudniają zdiagnozowanie przyczyny.
Pozytywną stroną jest to, że agent nie musi być skomplikowany. Do wielu zadań wystarczy prosta implementacja. Stworzenie go od zera daje ci pełną kontrolę nad tym, co trafia do modelu, jak podejmowane są decyzje i jak zarządzane są koszty, a znajomość podstaw sprawia, że późniejsze frameworki są łatwiejsze do zaufania. Poniżej przedstawiono krok po kroku proces tworzenia działającego agenta w Pythonie, wychodząc od podstaw.
Główna idea: pętla agenta
Najpierw należy uprościć tę koncepcję. Agent AI to duży model językowy, który może podejmować działania i wykorzystywać ich wyniki do decydowania, co robić dalej. W tradycyjnej wymianie z modelem językowym wysyłasz prompt i otrzymujesz jedną odpowiedź. Agent natomiast działa w pętli: analizuje obecną sytuację, wybiera działanie, je wykonuje, otrzymuje informacje zwrotne i powtarza ten proces, aż zadanie zostanie zakończone.
GAME: czteroczęściowy model projektowania
Aby stworzyć taką pętlę, użyjemy struktury GAME, która dzieli agenta na Cele, Działania, Pamięć oraz Środowisko. Dzięki temu powstaje zwięzły model mentalny, a kod pozostaje modułowy i łatwy do rozbudowy.
Cel
Cel określa, czego agent próbuje osiągnąć. Zawiera opis zadania oraz wszelkie instrukcje kierujące jego zachowaniem i staje się promptem systemu.
class Goal(BaseModel):
name: str
description: str
Działanie (narzędzie)
Baza kodu nazywa to Tool, ale oba terminy oznaczają to samo. Działanie to coś, co agent może wykonać: wywołać API, przeszukać dokumenty, zapytać bazę danych lub przeprowadzić obliczenie.
class Tool(BaseModel):
name: str
description: str
parameters: Dict[str, Any]
function: Callable
tags: Optional[List[str]] = None
terminal: bool = False
Obecne API modeli językowych, w tym te należące do OpenAI, przyjmują definicje narzędzi wraz z poleceniem. Definicje te muszą być napisane w formacie JSON Schema, podając nazwę każdego narzędzia, jego parametry oraz oczekiwany wprowadzony danych.
Jako pierwsze narzędzie weźmy funkcję, która dodaje dwa liczby:
def add(a: float, b: float) -> float:
return a + b
Następnie opisz to w formie, którą model może odczytać:
number_pair = {
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"},
},
"required": ["a", "b"],
}
add_tool = Tool(
name="add",
description="Adds two numbers.",
parameters=number_pair,
function=add,
)
Gdy budowana jest prośba, te obiekty działań są przekształcane w definicje narzędzi i wysyłane do modelu. Model nigdy nie uruchamia samej funkcji – decyduje jedynie, które narzędzie wywołać, oraz generuje dla niego argumenty.
Załóżmy, że użytkownik pyta: „Ile to 10 + 20?”. Model może odpowiedzieć coś w tym rodzaju:
{
"name": "add",
"arguments": {
"a": 10,
"b": 20
}
}
Wybór akcji i podanie jej argumentów to cała praca modelu.
Wysyłanie akcji do modelu
Ten pomocnik przekazuje historię rozmowy oraz dostępne akcje do LLM:
def generate_response(memory: Memory, tools: List[Dict[str, Any]]) -> Any:
from litellm import completion
messages = [m.model_dump(exclude_none=True) for m in memory.get()]
return completion(
model="openai/gpt-4o",
messages=messages,
max_tokens=1024,
tools=tools,
tool_choice="required",
parallel_tool_calls=False,
)
tools = [
{
"type": "function",
"function": {
"name": add_tool.name,
"description": add_tool.description,
"parameters": add_tool.parameters,
}
}
]
result = generate_response(
memory=Memory(messages=[
Message(role=Role.SYSTEM, content="Perform arithmetic using the add and multiply tools. "),
Message(role=Role.USER, content="Add 200 and 10, then multiply the result by 2.")]),
tools=tools
)
print(json.dumps(result.choices[0].message.model_dump(exclude_none=True), indent=2))
# Result:
{
"role": "assistant",
"tool_calls": [
{
"function": {
"arguments": "{\"a\":200,\"b\":10}",
"name": "add"
},
"id": "call_ZDlLQxVB9WLY7fPHwRipeeM1",
"type": "function"
}
],
"provider_specific_fields": {
"refusal": null
},
"annotations": []
}
Następnie agent uruchamia odpowiednią funkcję w Pythonie i wprowadza wynik do kolejnego kroku pętli.
Pamięć
Pamięć przechowuje dane ze wcześniejszych iteracji, dzięki czemu agent może brać pod uwagę poprzednie obserwacje przy wyborze następnego kroku.
Na razie jest ona minimalna: nowe wiadomości są po prostu dodawane do listy. Okna kontekstowe modeli LLM są skończone, więc w rzeczywistym systemie potrzebne jest coś bardziej zaawansowanego do ich redagowania lub streszczania, czego ten przykład nie uwzględnia.
class Memory(BaseModel):
"""M in GAME: the agent's conversational record - the running list of
Messages the LLM sees each turn. A thin, swappable store so strategies
(windowing, summarization, persistence) can vary later without touching
Agent.run; for now it is just an append-and-read list."""
messages: List[Message] = []
def add(self, message: Message) -> None:
self.messages.append(message)
def get(self) -> List[Message]:
return self.messages
Model Message reprezentuje każdą interakcję w pętli. Obserwacje, wyniki narzędzi, dane wprowadzone przez użytkownika oraz odpowiedzi asystenta przekształcają się wszystkie w wiadomości. Jego struktura jest bliska formatowi wiadomości obecnych API LLM, dzięki czemu tworzenie promptów na podstawie historii jest proste, a pamięć łatwa do rozszerzania.
from enum import Enum
class Role(str, Enum):
SYSTEM = "system"
USER = "user"
ASSISTANT = "assistant"
TOOL= "tool"
class Message(BaseModel):
role: str
content: Optional[str] = None
Środowisko
Możesz się zastanawiać, dlaczego w ogóle istnieje środowisko, skoro operacja może zostać wykonywana bezpośrednio wewnątrz jego własnej klasy.
class Action:
...
def execute_action(self, **args):
return function(**args)
To w porządku dla przykładów edukacyjnych. Staje się to ograniczeniem, gdy potrzebujesz iniekcji zależności, ukrytego kontekstu, połączeń z bazą danych, sesji użytkowników lub innych zasobów z zewnątrz, a późniejsze rozdziały pokazują, w jaki sposób to działa.
class Environment:
def execute_action(
self,
action: Tool,
args: dict,
) -> Any:
"""Run `action` with the LLM-provided `args`, injecting the hidden context
into whichever param the tool declares for it (by name or subclass type).
Returns the tool's native result, or {"error": ...} if it raised."""
try:
call_args = dict(args)
return action.function(**call_args)
except Exception as e:
return {"error": str(e)}
Budowa pętli agenta
Gdy cztery komponenty GAME zostały zdefiniowane, można napisać podstawową pętlę. Składa ona prompt, prosi LLM o odpowiedź, wykonuje wybrane działania, zbiera obserwacje i powtarza ten proces aż zostanie spełniona warunek zatrzymania.
class Agent:
def __init__(self, goal: Goal, actions: List[Action], environment: Optional[Environment] = None):
self.goal = goal
self.actions = {a.name: a for a in actions}
self.environment = environment or Environment()
self.memory = Memory()
self.memory.add(Message(role=Role.SYSTEM, content=goal.description))
def tools_schema(self) -> List[Dict[str, Any]]:
"""Describe every action in the function-calling format the LLM expects."""
return [
{
"type": "function",
"function": {
"name": a.name,
"description": a.description,
"parameters": a.parameters,
},
}
for a in self.actions.values()
]
def run(self, user_input: str, max_iterations: int = 15) -> Any:
self.memory.add(Message(role=Role.USER, content=user_input))
for _ in range(max_iterations):
# 1. Ask the LLM what to do next, given the conversation + tools.
response = generate_response(self.memory, tools=self.tools_schema())
assistant_msg = response.choices[0].message
# 2. Record the assistant's turn (text and/or tool calls).
self.memory.add(
Message(
role=Role.ASSISTANT,
content=assistant_msg.content,
tool_calls=[tc.model_dump() for tc in assistant_msg.tool_calls]
if assistant_msg.tool_calls
else None,
)
)
# 3. If it asked for tools, run them and feed the results back.
if assistant_msg.tool_calls:
for call in assistant_msg.tool_calls:
action = self.actions[call.function.name]
args = json.loads(call.function.arguments)
result = self.environment.execute_action(action, args)
self.memory.add(
Message(
role=Role.TOOL,
tool_call_id=call.id,
name=action.name,
content=str(result),
)
)
# 4. A terminal action ends the loop and carries the result.
if action.terminal:
return result
return None
if __name__ == "__main__":
number_pair = {
"type": "object",
"properties": {"a": {"type": "number"}, "b": {"type": "number"}},
"required": ["a", "b"],
}
actions = [
Action(
name="add",
description="Adds two numbers.",
parameters=number_pair,
function=add,
),
Action(
name="multiply",
description="Multiplies two numbers.",
parameters=number_pair,
function=multiply,
),
Action(
name="finish",
description="Provide the final answer to the user and end the task.",
parameters={
"type": "object",
"properties": {"answer": {"type": "string"}},
"required": ["answer"],
},
function=finish,
terminal=True, # this one ends the loop
),
]
agent = Agent(
goal=Goal(
name="calculator",
description=(
"Perform arithmetic using the add and multiply tools. "
"When you have the final answer, call finish with it."
),
),
actions=actions,
)
result = agent.run("Add 200 and 10, then multiply the result by 2.")
print("\nresult:", result)
print(json.dumps([m.model_dump(exclude_none=True) for m in agent.memory.get()], indent=2))
## Result
result: 420
[
{
"role": "system",
"content": "Perform arithmetic using the add and multiply tools. When you have the final answer, call finish with it."
},
{
"role": "user",
"content": "Add 200 and 10, then multiply the result by 2."
},
{
"role": "assistant",
"tool_calls": [
{
"function": {
"arguments": "{\"a\":200,\"b\":10}",
"name": "add"
},
"id": "call_wYPzGCwLuc71cK7q2Ok81OzF",
"type": "function"
}
]
},
{
"role": "tool",
"content": "210",
"tool_call_id": "call_wYPzGCwLuc71cK7q2Ok81OzF",
"name": "add"
},
{
"role": "assistant",
"tool_calls": [
{
"function": {
"arguments": "{\"a\":210,\"b\":2}",
"name": "multiply"
},
"id": "call_5JD4McfNm0m68Poedsm8uODO",
"type": "function"
}
]
},
{
"role": "tool",
"content": "420",
"tool_call_id": "call_5JD4McfNm0m68Poedsm8uODO",
"name": "multiply"
},
{
"role": "assistant",
"tool_calls": [
{
"function": {
"arguments": "{\"answer\":\"420\"}",
"name": "finish"
},
"id": "call_nbIGeKvgWtXPh4YlioZxfiyj",
"type": "function"
}
]
},
{
"role": "tool",
"content": "420",
"tool_call_id": "call_nbIGeKvgWtXPh4YlioZxfiyj",
"name": "finish"
}
]
Kilka uwag dotyczących szczegółów. Format wiadomości mógłby być ukształtowany inaczej niż rola "tool" użyta tutaj, ale struktura wymagałaby niewielkich modyfikacji. Większość interfejsów API LLM stosuje konwencję, zgodnie z którą wynik narzędzia musi być powiązany z wcześniejszym wywołaniem tego narzędzia, zazwyczaj za pomocą tool_id; dlatego ten sam identyfikator pojawia się dwa razy w historii: raz w żądaniu, a raz w odpowiadającym mu wyniku.
Współczesne modele są również intensywnie szkolenia na przykładach wywołań narzędzi. Ponowne użycie ról oraz wzorów wiadomości, które widziały podczas szkolenia, odpowiada ich oczekiwaniom i zazwyczaj zapewnia bardziej stabilne wywołania narzędzi oraz lepszą ogólną wydajność.
Specialny narzędzie finish również zasługuje na wzmiankę. Daje modelowi wyraźny sposób na poinformowanie, że zadanie zostało ukończone, dzięki czemu nie może być w nieskończoność zablokowany przy wywoływaniu narzędzi. Jest rejestrowane jak każde inne narzędzie w każdym agencie, co jest aspektem projektowym, który zostanie wkrótce udoskonalony.
Gdzie pierwsza wersja ma ograniczenia
Ten agent funkcjonuje i udostępnia cały cykl bez żadnych ukrytych elementów za frameworkiem, co jest idealne do nauki. Po przeanalizowaniu prostych przykładów szybko ujawniają się trzy ograniczenia.
- Definicje narzędzi nie są skalowalne. Każda akcja wymaga ręcznie napisanego JSON Schema, które musi być zsynchronizowane z odpowiadającą mu funkcją w Pythonie. W miarę wzrostu liczby narzędzi staje się to pracochłonne i podatne na błędy, a nawet niewielka zmiana w ich strukturze może sprawić, że schema stanie się przestarzałe, co skutkuje nieprawidłowymi wywołaniami przez model.
finish zatrzymuje pętlę i zwraca wynik, ale każdy agent musi to zarejestrować, a ono może zwrócić jedynie zwykłą wartość. Bardziej zaawansowani agenci będą chcieli bardziej złożonych reguł zatrzymywania oraz ustrukturyzowanego wyniku.Każdy z tych aspektów jest omawiany po kolei, począwszy od sposobu zarządzania i rejestracji narzędzi.
Problem 1: rejestr narzędzi
Zacznijmy od najbardziej irytującego aspektu – ręcznego dopasowywania schematu i funkcji. API służące do uzupełniania tekstu wymagają listy schematów JSON opisujących dostępne narzędzia.
W pierwszej wersji każde narzędzie składało się z dwóch oddzielnie napisanych elementów: funkcji w Pythonie, która wykonywała zadanie, oraz schematu JSON, który wskazywał modelowi językowemu, jak ją wywołać:
def add(a: float, b: float) -> float:
return a + b
add_tool = {
"type": "function",
"function": {
"name": "add",
"description": "Adds two floats.",
"parameters": {
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"},
},
"required": ["a", "b"],
},
},
}
tools = [add_tool]
...
completion(
model="openai/gpt-4o",
messages=messages,
tools=tools,
)
Rozwiązaniem jest wygenerowanie schematu na podstawie funkcji. Funkcja staje się jedynym źródłem prawdy, a schemat jest automatycznie wytwarzany podczas rejestracji narzędzia.
JSON_TYPE_MAP = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
}
def get_json_type(python_type: Any) -> str:
origin = get_origin(python_type)
if python_type in (list, List) or origin in (list, List):
return "array"
if python_type in (dict, Dict) or origin in (dict, Dict):
return "object"
if origin in (Union, types.UnionType):
non_none = [a for a in get_args(python_type) if a is not type(None)]
return get_json_type(non_none[0]) if non_none else "string"
return JSON_TYPE_MAP.get(python_type, "string")
def build_tool(func: Callable, ...) -> Tool:
name = name or func.__name__
description = description or (func.__doc__.strip() if func.__doc__ else "")
signature = inspect.signature(func)
type_hints = get_type_hints(func)
args_schema = {"type": "object", "properties": {}, "required": []}
for param_name, param in signature.parameters.items():
param_type = type_hints.get(param_name, str)
args_schema["properties"][param_name] = {"type": get_json_type(param_type)}
if param.default == inspect.Parameter.empty: # no default ⇒ required
args_schema["required"].append(param_name)
return Tool(name=name, description=description,
parameters=args_schema, function=func)
Następnie ten budownik powinien uruchamiać się automatycznie za każdym razem, gdy definiowana jest funkcja narzędzia, więc zadanie to wykonuje dekorator:
@tool()
def add(a: float, b: float) -> float:
"""Adds two floats."""
return a + b
Dekorator jest celowo skromny. Tworzy Tool, przypina go do obiektu funkcji i zwraca funkcję bez zmian:
def tool(name=None, description=None, parameters_override=None, tags=None):
def decorator(func):
func._tool_metadata = build_tool(func, name, description,
parameters_override, tags)
return func
return decorator
# The decorator automatically add a new attribute _tool_metadata contains the
# information of the tool (Tool)
Próba użycia:
serializable@tool()
def add(a: float, b: float) -> float:
"""Adds two floats."""
return a + b
if __name__ == "__main__":
# _tool_metadata (Tool) will be added as an attribute of the function after the decorator call
tool_metadata = add._tool_metadata
# exclude function as it can not be serializable
print(json.dumps(tool_metadata.model_dump(exclude={"function"}), indent=2)) #
# Result
{
"name": "add",
"description": "Adds two floats.",
"parameters": {
"type": "object",
"properties": {
"a": {
"type": "number"
},
"b": {
"type": "number"
}
},
"required": [
"a",
"b"
]
},
"tags": null,
"terminal": false
}
Gdy istnieje wiele narzędzi i kilku agentów, a agenci dzielą się narzędziami, pomocne jest centralne miejsce do ich zarządzania. Rejestr działań pełni tę rolę, odnotowując, jakie narzędzia może używać dany agent.
class ActionRegistry:
def __init__(self):
self.tools: Dict[str, Tool] = {}
def register(self, action: Callable | Tool, terminal: bool = False):
# accept a raw @tool-decorated function or a pre-built Tool
if isinstance(action, Tool):
tool = action
else:
tool = getattr(action, "_tool_metadata", None) or build_tool(action)
if terminal != tool.terminal:
tool = tool.model_copy(update={"terminal": terminal})
self.tools[tool.name] = tool
def get_tool(self, name: str) -> Optional[Tool]:
return self.tools.get(name)
def get_tools_schema(self) -> List[Dict[str, Any]]:
"""Tool schemas in the OpenAI / litellm function-calling format."""
return [
{"type": "function",
"function": {"name": t.name,
"description": t.description,
"parameters": t.parameters}}
for t in self.tools.values()
]
Zarejestrowanie funkcji polega wtedy na jednym wywołaniu:
registry.register(add)
rejestr odczytuje metadane dołączone przez @tool i przekształca je w instancję Tool. W przypadku funkcji, która nie została ozdobiona tym tagiem, system automatycznie ucieka się do wywołania build_tool(). To wygodny standard, ale osoby chcące mieć pełną kontrolę mogą nadal ręcznie tworzyć obiekty Tool.
Należy również zauważyć, że terminal jest ustawiany podczas rejestracji, a nie w samym narzędziu:
registry.register(tool, terminal=True)
To, czy narzędzie zakończa pętlę, jest właściwością agenta, a nie funkcji. To samo narzędzie może być terminalne dla jednego agenta, a zwykłe dla innego. Gdy wymagana jest inna wartość terminal, rejestr kopiuje Tool zamiast modyfikować oryginał, dzięki czemu konfiguracja nie ucieka między rejestrami, a jedna definicja może być bezpiecznie wykorzystywana w różnych agentach.
Problem 2: wprowadzanie zależności
Narzędzia obecnie działają, przy jednym ważnym ograniczeniu: widzą jedynie argumenty wygenerowane przez LLM.
Praktyczne narzędzia często potrzebują zasobów, które nie powinny być pokazywane modelowi, w tym połączeń z bazą danych, klientów API, obiektów konfiguracyjnych, sesji użytkowników oraz stanu aplikacji. Niektóre muszą również mieć dostęp do pamięci agenta. Wyobraźmy sobie narzędzie, które zapisuje artykuły naukowe do bazy danych:
@tool()
def save_paper(
title: str,
content: str,
database_url: str,
api_key: str,
):
...
Aby to lepiej ująć, klasa przechowuje kontekst działania:
class ActionContext(BaseModel):
database_url: str,
api_key: str
# Then the tool becomes:
@tool()
def save_paper(
title: str,
content: str,
action_context: ActionContext
):
...
Gdyby kontekst był po prostu kolejnym argumentem narzędzia, model LLM musiałby sam wymyślić te wartości, co nie ma sensu, ponieważ powinny być podawane z zewnątrz. W tym przypadku pomocne jest Environment. Parametry kontekstu są ukryte przed modelem i uzupełniane ręcznie w momencie wykonywania.
Ponieważ niektóre narzędzia wymagają kontekstu, a inne nie, pierwszym krokiem jest funkcja, która na podstawie adnotacji typów sprawdza, czy funkcja przyjmuje ActionContext lub jego podklasę:
def context_param(func: Callable) -> Optional[tuple[str, type]]:
try:
hints = get_type_hints(func)
except Exception:
hints = {}
for name in inspect.signature(func).parameters:
ann = hints.get(name)
if isinstance(ann, type) and issubclass(ann, ActionContext):
return name, ann
return None
# Now we modify the `build_tool` function
def build_tool(
func: Callable,
name: Optional[str] = None,
description: Optional[str] = None,
parameters_override: Optional[Dict[str, Any]] = None,
tags: Optional[List[str]] = None,
) -> Tool:
name = name or func.__name__
description = description or (func.__doc__.strip() if func.__doc__ else "")
if parameters_override is None:
signature = inspect.signature(func)
type_hints = get_type_hints(func)
args_schema = {
"type": "object",
"properties": {},
"required": [],
}
# The injected-context param (by name OR ActionContext-subclass annotation)
# is filled by the Environment from the hidden ActionContext. Never expose it
# to the LLM, or the LLM would try to generate the secrets that live inside it.
ctx = context_param(func)
ctx_param_name = ctx[0] if ctx else None
for param_name, param in signature.parameters.items():
if param_name == ctx_param_name:
continue
param_type = type_hints.get(param_name, str)
param_schema = {
"type": get_json_type(param_type)
}
args_schema["properties"][param_name] = param_schema
if param.default == inspect.Parameter.empty:
args_schema["required"].append(param_name)
else:
args_schema = parameters_override
return Tool(
name=name,
description=description,
parameters=args_schema,
function=func,
tags=tags or []
)
Narzędzie, które używa parametru ActionContext, pomija ten parametr w swoim schemacie JSON, więc model językowy nie wie o jego istnieniu i nie może spróbować go wygenerować. Environment następnie przywraca ten kontekst w momencie wykonywania:
class Environment:
def execute_action(
self,
action: Tool,
args: dict,
action_context: Optional[ActionContext] = None,
) -> Any:
try:
call_args = dict(args)
ctx = context_param(action.function)
if ctx:
ctx_param_name, _ = ctx
call_args[ctx_param_name] = action_context or ActionContext()
return action.function(**call_args)
except Exception as e:
return {"error": str(e)}
To właśnie jest sztuczka. Środowisko pełni rolę granicy pomiędzy modelem a światem zewnętrznym. Model dostarcza argumenty, o których wie, natomiast Środowisko dodaje zasoby, o których nie powinien wiedzieć. To prosta koncepcja, która sprawia, że agenci stają się znacznie bardziej praktyczne.
Problem 3: ustrukturyzowany wynik i wbudowane narzędzie zakończenia
Pozostają dwa problemy. Po pierwsze, agent może obecnie zwracać tylko łańcuchy znaków, podczas gdy rzeczywiste procesy często wymagają danych ustrukturyzowanych. Po drugie, narzędzie służące do zakończenia powinno być wbudowane w agenta, a nie czymś, co trzeba za każdym razem rejestrować zewnętrznie.
Oba problemy rozwiązuje się jednocześnie. Dla ustrukturyzowanego wyniku wiele API do uzupełniania oferuje parametr response_format, który zmusza model językowy do generowania JSON zgodnego ze schematem. To koliduje z pętlą wywoływania narzędzi. Gdy aktywowany zostanie response_format, model zazwyczaj od razu zapisuje swoją ostateczną odpowiedź zamiast wywoływać narzędzia, a agent przestaje po pierwszym uzupełnieniu. Mieszanie tych dwóch podejść okazało się również niezręczne i niespójne w praktyce.
Aльтernatywą jest ponowne wykorzystanie wywoływania narzędzi. Schemat wyniku staje się parametrami specjalnego narzędzia, a po zakończeniu pracy model językowy wywołuje to narzędzie z argumentami odpowiadającymi pożądanemu schematowi. Agent pobiera te argumenty i tworzy ostateczny obiekt wyniku.
Pomocnik w klasie Agent automatycznie rejestruje wbudowane narzędzie terminate:
OutputSchema = Union[type[str], type[BaseModel]]
class Agent:
...
@classmethod
def _build_terminate_tool(cls, output_schema: OutputSchema) -> Tool:
"""Build the terminate tool that ends the agent loop and carries the final result.
str is normalized to a one-field model {result: str} and unwrapped on return,
so a single code path handles both: the LLM-facing schema always comes from
pydantic's model_json_schema(), and result is the string (for str) or
the schema instance (for a BaseModel).
"""
unwrap_field = None
if output_schema is str:
output_schema = create_model("StrResult", **{cls._TERMINATE_STR_FIELD: (str, ...)})
unwrap_field = cls._TERMINATE_STR_FIELD
def terminate(**kwargs: Any) -> Any:
"""Provide the final result and end the task."""
obj = output_schema(**kwargs)
return getattr(obj, unwrap_field) if unwrap_field else obj
terminate_tool = build_tool(
func=terminate,
name=cls.TERMINATE_TOOL_NAME,
description=f"Provide the final result ({output_schema.__name__}) and end the task.",
parameters_override=output_schema.model_json_schema(),
)
terminate_tool.terminal = True
return terminate_tool
Oto ukończony pętla agenta:
class Agent:
TERMINATE_TOOL_NAME = "terminate"
_TERMINATE_STR_FIELD = "result"
def __init__(
self,
goal: str,
action_registry: ActionRegistry,
output_schema: Optional[OutputSchema] = None,
environment: Optional["Environment"] = None,
action_context: Optional[ActionContext] = None,
):
self.goal = goal
self.action_registry = action_registry
self.output_schema = output_schema
self.result: Any = None
self.environment = environment or Environment()
self.action_context = action_context or ActionContext()
self._validate_context()
# Tools the registry marked terminal (excluding the framework's terminate tool).
terminal_tools = [
t.name for t in action_registry.tools.values()
if t.terminal and t.name != self.TERMINATE_TOOL_NAME
]
# No exit declared at all -> default to a text answer (output_schema=str).
if output_schema is None and not terminal_tools:
output_schema = self.output_schema = str
# Because in my design, Agent can stop the loop after a specific tool call. (to save tokens, LLm do not need to generate terminal tool with the same parameters again)
# This choice make it a little bit complicated when output schema defined in Agent different from the tool's output schema.
if output_schema is not None:
if terminal_tools:
raise ValueError(
f"output_schema={output_schema.__name__} conflicts with terminal "
f"tool(s) {terminal_tools}: their output does not match the schema. "
f"Drop output_schema or unmark those tools as terminal."
)
action_registry.register_tool(self._build_terminate_tool(output_schema))
elif len(terminal_tools) > 1:
raise ValueError(
f"Multiple terminal tools {terminal_tools}; an agent must have exactly "
f"one exit so result has a well-defined type."
)
self.memory = Memory()
self.memory.add(Message(role=Role.SYSTEM, content=self.goal))
def _validate_context(self) -> None:
"""Verify that the agent's action_context satisfies every tool's declared"""
for tool in self.action_registry.tools.values():
ctx = context_param(tool.function)
if ctx is None:
continue # tool declares no context — nothing to satisfy
_, required = ctx
if not isinstance(self.action_context, required):
raise TypeError(
f"Tool {tool.name!r} requires action_context: {required.__name__}, "
f"but this agent was given {type(self.action_context).__name__}. "
f"Pass a context that is-a {required.__name__} (combine schemas via "
f"subclassing if multiple tools need different types), or split the "
f"tools across agents."
)
@classmethod
def _build_terminate_tool(cls, output_schema: OutputSchema) -> Tool:
"""Build the terminate tool that ends the agent loop and carries the final result.
str is normalized to a one-field model {result: str} and unwrapped on return,
so a single code path handles both: the LLM-facing schema always comes from
pydantic's model_json_schema(), and result is the string (for str) or
the schema instance (for a BaseModel).
"""
unwrap_field = None
if output_schema is str:
output_schema = create_model("StrResult", **{cls._TERMINATE_STR_FIELD: (str, ...)})
unwrap_field = cls._TERMINATE_STR_FIELD
def terminate(**kwargs: Any) -> Any:
"""Provide the final result and end the task."""
obj = output_schema(**kwargs)
return getattr(obj, unwrap_field) if unwrap_field else obj
terminate_tool = build_tool(
func=terminate,
name=cls.TERMINATE_TOOL_NAME,
description=f"Provide the final result ({output_schema.__name__}) and end the task.",
parameters_override=output_schema.model_json_schema(),
)
terminate_tool.terminal = True
return terminate_tool
def add_memory(self, message: Message):
self.memory.add(message)
@staticmethod
def _to_content(result: Any) -> str:
"""Render a tool's return value as text for the LLM. """
if isinstance(result, BaseModel):
return result.model_dump_json()
if isinstance(result, (dict, list)):
return json.dumps(result, ensure_ascii=False)
return str(result)
def run(self, user_input: str, max_iterations: int = 15):
self.add_memory(Message(role=Role.USER, content=user_input))
for i in range(max_iterations):
response = generate_response(self.memory, tools=self.action_registry.get_tools_schema())
assistant_msg = response.choices[0].message
# Record the assistant turn (may carry content and/or tool_calls).
self.add_memory(
Message(
role=Role.ASSISTANT,
content=assistant_msg.content,
tool_calls=[tc.model_dump() for tc in assistant_msg.tool_calls]
if assistant_msg.tool_calls else None,
)
)
if assistant_msg.tool_calls:
for tool_call in assistant_msg.tool_calls:
tool_name = tool_call.function.name
tool_args = tool_call.function.arguments
tool = self.action_registry.get_tool(tool_name)
if not tool:
raise ValueError(f"Tool '{tool_name}' not found.")
# Execute the tool in the Environment
result = self.environment.execute_action(
tool, json.loads(tool_args), self.action_context
)
self.add_memory(
Message(
role=Role.TOOL,
tool_call_id=tool_call.id,
name=tool_name,
content=self._to_content(result)
)
)
if tool.terminal:
self.result = result
return self.result
return self.result
Co pokazuje to ćwiczenie
W gruncie rzeczy agent to model językowy działający wewnątrz pętli. Odczytuje sytuację, wybiera działanie, otrzymuje wynik i powtarza ten proces aż do osiągnięcia celu. Pydantic AI, AutoGen, Microsoft Agent Framework oraz większość innych rozwiązań są zbudowane według tego samego wzorca.
Krok po kroku zwykła pętla wywoływania narzędzi rozwinęła się w coś bardziej zaawansowanego. Rejestr narzędzi umożliwił ponowne wykorzystanie działań, warstwa środowiskowa bezpiecznie wprowadzała ukryte zależności, a ustrukturyzowany mechanizm zakończenia pozwalał agentom zwracać dane typowane zamiast zwykłego tekstu. Żadna z tych koncepcji sama w sobie nie jest trudna, ale razem stanowią podstawę większości współczesnych rozwiązań dotyczących agentów.
Znajomość tych podstaw umożliwia łatwiejszą ocenę frameworków, szybsze debugowanie nieprawidłowego zachowania oraz lepsze dostosowanie własnych projektów do swoich potrzeb. Frameworki przyspieszają rozwój, ale są to abstrakcje koncepcji, które okazują się dość proste, gdy zostaną jasno przedstawione.