Створення AI-агента з нуля на Python: цикл GAME крок за кроком
Створіть працездатного агента для виклику інструментів на звичайному Python із елементами Goals, Actions, Memory та Environment, а потім додайте реєстр інструментів, приховану ін’єкцію залежностей та типізований вивід.
Фреймворки агентів, такі як Pydantic AI, AutoGen та Microsoft Agent Framework, ідеально підходять для швидкого створення прототипу. Однак після використання кількох з них зазвичай з’являються певні проблеми.
- Екосистема все ще розвивається. Кожен фреймворк покриває частину необхідних функцій та пропускає інші, а інший фреймворк має те, чого бракує першому. Це нормально для молодої галузі, і з часом більшість з них зможе вирішувати поширені сценарії використання.
- Накопичується абстракція. Фреймворки прагнуть підтримувати величезний спектр дизайнів агентів, тому вони додають шари, які ускладнюють розуміння та дебагування. LangChain є типовим прикладом: коли система поводиться некоректно, ці шари заважають з’ясувати причину.
Позитивною стороною є те, що агент не обов’язково має бути складним. Для багатьох завдань достатньо простої реалізації. Створення агента з нуля дає вам повний контроль над тим, що потрапляє до моделі, як приймаються рішення та як керується витратами, а знання основ допомагає легше довіряти будь-яким наступним фреймворкам. Далі наведено поетапне створення працездатного Python-агента на основі первинних принципів.
Основна ідея: цикл агента
Почніть з спрощення ідеї. Штучний інтелект-агент — це велика мовна модель, яка може виконувати дії та використовувати їхні результати для прийняття рішення про наступні кроки. У традиційному обміні з LLM ви надсилаєте запит та отримуєте одну відповідь. Агент же працює у циклі: він аналізує поточний стан, обирає дію, її виконує, отримує зворотний зв’язок та повторює процес, поки завдання не буде виконане.
GAME: чотирикомпонентна структура
Щоб побудувати цей цикл, ми використаємо структуру GAME, яка розділяє агента на Цілі, Дії, Пам’ять та Середовище. Вона створює компактну ментальну модель та забезпечує модульний код, який легко розширювати.
Ціль
Ціль визначає, чого намагається досягти агент. Вона містить опис завдання та будь-які інструкції, які керують його поведінкою, і стає системним запитом.
class Goal(BaseModel):
name: str
description: str
Дія (інструмент)
У цій базі коду її називають Інструментом, але ці терміни означають одне й те саме. Дія — це щось, що може виконувати агент: виклик API, пошук документів, запит до бази даних чи виконання обчислень.
class Tool(BaseModel):
name: str
description: str
parameters: Dict[str, Any]
function: Callable
tags: Optional[List[str]] = None
terminal: bool = False
Сучасні API LLM, зокрема OpenAI, приймають визначення інструментів разом із запитом. Ці визначення мають бути написані у форматі JSON Schema, зазначаючи назву кожного інструменту, його параметри та очікуваний вхідний даний.
Як перший інструмент візьмемо функцію, яка додає два числа:
def add(a: float, b: float) -> float:
return a + b
Потім описайте його у форматі, який може прочитати модель:
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,
)
Коли формується запит, ці об’єкти дій перетворюються на визначення інструментів та надсилаються до моделі. Модель ніколи не виконує саму функцію. Вона лише вирішує, який інструмент викликати, та генерує для нього аргументи.
Припустимо, користувач запитує: «Що дорівнює 10 + 20?» Модель може відповісти щось на кшталт цього:
{
"name": "add",
"arguments": {
"a": 10,
"b": 20
}
}
Вибір дії та надання її аргументів — це вся робота моделі.
Надсилання дій до моделі
Цей помічник передає історію розмови та доступні дії до 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": []
}
Потім агент запускає відповідну функцію на Python та передає результат у наступний етап циклу.
Пам’ять
У пам’яті зберігається інформація, зібрана попередніми ітераціями, тож агент може враховувати попередні спостереження при виборі наступного кроку.
Наразі вона залишається мінімальною: нові повідомлення просто додаються до списку. Вікна контексту ШІ мають обмежену кількість простору, тому справжня система потребує щось більш розумне для скорочення чи узагальнення інформації, що в цьому прикладі не розглядається.
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
Модель Message представляє кожну взаємодію у циклі. Спостереження, результати інструментів, вхідні дані користувача та відповіді асистента — усе це стає повідомленнями. Її формат точно відповідає формату повідомлень у сучасних API ШИ, тому створення запитів на основі історії є простим, а пам’ять легко розширюється.
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
Середовище
Ви можете запитати, навіщо взагалі потрібне середовище, адже дію можна просто виконати всередині його власного класу.
class Action:
...
def execute_action(self, **args):
return function(**args)
Це підходить для прикладів-іграшок. Однак це стає обмеженням, коли потрібна ін’єкція залежностей, прихований контекст, підключення до баз даних, сесії користувачів чи інші зовнішні ресурси, і наступні розділи показують, як це реалізувати.
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)}
Створення циклу агента
Після визначення чотирьох компонентів GAME можна написати основний цикл. Він формує запит, просить LLM надати відповідь, виконує обрані дії, збирає дані спостережень та повторює цей процес до досягнення умови зупинки.
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"
}
]
Кілька зауважень щодо деталей. Формат повідомлення може відрізнятися від формату, який використовується тут для ролі "tool", але структуру доведеться трохи змінити. Більшість API ШІ мають спільну конвенцію: результат використання інструменту має бути пов’язаний із запитом на використання цього інструменту, зазвичай через tool_id; тому той самий ідентифікатор з’являється двічі в історії — один раз у запиті та один раз у відповідному результаті.
Сучасні моделі також інтенсивно навчаються на прикладах виклику інструментів. Повторне використання ролей та шаблонів повідомлень, які вони бачили під час навчання, відповідає їхнім очікуванням та зазвичай забезпечує більш стабільні виклики інструментів та кращу загальну продуктивність.
Особливий інструмент finish також заслуговує на увагу. Він надає моделі чіткий спосіб повідомити, що завдання завершене, щоб вона не застрягла у безкінечних викликах інструментів. Він реєструється, як і будь-який інший інструмент, у кожному агенті, що є аспектом дизайну, який незабаром буде вдосконалений.
Чим не вистачає першої версії
Цей агент працює та відкриває весь цикл без жодних приховувань за фреймворком, що ідеально підходить для навчання. Окрім простих прикладів, швидко проявляються три обмеження.
- Визначення інструментів не масштабуються. Кожній дії потрібна вручну написана схема JSON, яка має бути синхронізована з відповідною функцією на Python. У міру збільшення кількості інструментів це стає повторюваним та схильним до помилок, а навіть незначна зміна сигнатури може зробити схему застарілою, через що модель створює недійсні виклики.
finish зупиняє цикл та повертає результат, але кожен агент мусить це зареєструвати, і він може повертати лише просте значення. Більш здатні агенти будуть потребувати більш детальних правил зупинки та структурованого виводу.Кожен аспект розглядається по черзі, починаючи з того, як керуються та реєструються інструменти.
Проблема 1: реєстр інструментів
Почнемо з найбільш проблематичного аспекту — вручну підтримувати узгодженість схем та функцій. API для завершення роботи вимагають списку JSON-схем, які описують доступні інструменти.
У першій версії кожен інструмент був представлений у вигляді двох окремо написаних елементів: функції Python, яка виконує роботу, та JSON Schema, який пояснює LLM, як її викликати:
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,
)
Відповідь полягає у створенні схеми з функції. Функція стає єдиним джерелом істини, а схема автоматично генерується під час реєстрації інструменту.
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)
Далі цей конструктор має запускатися автоматично щоразу, коли визначається функція інструменту, тож цю роботу виконує декоратор:
@tool()
def add(a: float, b: float) -> float:
"""Adds two floats."""
return a + b
Декоратор навмисно залишається скромним. Він створює об’єкт Tool, приєднує його до об’єкта функції та повертає функцію без змін:
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)
Спробуємо це:
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
}
Коли існує багато інструментів та кілька агентів, які діляться цими інструментами, корисно мати централізоване місце для їх керування. Реєстр дій виконує саме цю функцію, відстежуючи, які інструменти може використовувати певний агент.
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()
]
Реєстрація функції тоді — це просто один виклик:
registry.register(add)
Реєстр читає метадані, які додав @tool, та перетворює їх на екземпляр Tool. Для функції, яка не була оформлена спеціальними декораторами, він автоматично звертається до функції build_tool(). Це зручний стандартний варіант, а ті, хто хоче повного контролю, все одно можуть вручну створювати об’єкти Tool.
Також зверніть увагу, що значення terminal встановлюється під час реєстрації, а не для самого інструменту:
registry.register(tool, terminal=True)
Те, чи завершує інструмент цикл, є властивістю агента, а не функції. Той самий інструмент може бути термінальним для одного агента та звичайним для іншого. Коли потрібне інше значення terminal, реєстр копіює Tool замість того, щоб змінювати оригінал, тож конфігурація не витікає між реєстрами, і одне визначення може безпечно використовуватися в різних агентах.
Проблема 2: вставка залежностей
Інструменти наразі працюють, але існує одне важливе обмеження: вони бачать лише аргументи, створені LLM.
Практичні інструменти часто потребують ресурсів, які не повинні бути видимими для моделі, зокрема підключення до баз даних, клієнти API, об’єкти конфігурації, сеанси користувачів та стан додатку. Деякі з них також потребують доступу до пам’яті агента. Уявіть інструмент, який зберігає наукові статті в базі даних:
@tool()
def save_paper(
title: str,
content: str,
database_url: str,
api_key: str,
):
...
Щоб краще це ілюструвати, клас несе контекст дії:
class ActionContext(BaseModel):
database_url: str,
api_key: str
# Then the tool becomes:
@tool()
def save_paper(
title: str,
content: str,
action_context: ActionContext
):
...
Якби контекст був просто ще одним аргументом інструменту, моделі ШІ довелося б вигадувати ці значення, що не має сенсу, адже вони мають бути введені ззовні. Саме тут допомагає Environment. Параметри контексту приховані від моделі та вводяться вручну під час виконання.
Оскільки деякі інструменти потребують контексту, а інші — ні, першим кроком є створення функції, яка на основі анотацій типів перевіряє, чи приймає функція параметр ActionContext чи його підклас:
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 []
)
Інструмент, який використовує параметр ActionContext, не включає цей параметр у свою схему JSON, тому ШІ не знає про його існування та не може спробувати його створити. Environment потім повертає цей контекст під час виконання:
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)}
Ось у чому секрет. Середовище виступає як межа між моделлю та зовнішнім світом. Модель надає ті аргументи, про які знає, а Середовище додає ресурси, про які вона не повинна знати. Ця проста ідея робить агентів набагато практичнішими.
Проблема 3: структурований вихід та вбудований інструмент завершення
Залишилося два питання. По-перше, наразі агент може повертати лише рядки, тоді як у справжніх робочих процесах часто потрібні структуровані дані. По-друге, інструмент завершення має бути вбудованим у агента, а не чимось, що потрібно реєструвати ззовні щоразу.
Обидва підходи вирішуються разом. Для структурованого виводу багато API для завершення завдань пропонують параметр response_format, який змушує ШІ генерувати JSON, що відповідає певній схемі. Це суперечить циклу виклику інструментів. Як тільки активується response_format, модель зазвичай одразу записує свою кінцеву відповідь замість того, щоб викликати інструменти, а агент припиняє роботу після першого завершення. Поєднання цих двох підходів на практиці також виявилося незручним та неоднозначним.
Альтернативою є повторне використання механізму виклику інструментів. Схема виводу стає параметрами спеціального інструменту, і коли робота завершена, ШІ викликає цей інструмент з аргументами, що відповідають бажаній схемі. Агент бере ці аргументи та створює кінцевий об’єкт виводу.
Допоміжний елемент у класі Agent автоматично реєструє вбудований інструмент 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
Ось готовий цикл агента:
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
Що показує цей приклад
По суті, агент — це мовна модель, яка працює в межах циклу. Вона аналізує ситуацію, обирає дію, отримує результат та повторює процес до досягнення мети. Pydantic AI, AutoGen, Microsoft Agent Framework та більшість інших рішень побудовані за цим самим принципом.
Крок за кроком звичайний цикл виклику інструментів перетворився на щось більш потужне. Реєстр інструментів дозволяє повторно використовувати дії, шар середовища безпечно вставляє приховані залежності, а структурований механізм завершення дозволяє агентам повертати типізовані результати замість простого тексту. Жодна з цих ідей сама по собі не є складною, проте разом вони лежать в основі більшої частини сучасного проектування агентів.
Знання цих примітивів допомагає краще оцінювати фреймворки, легше відлагоджувати дивну поведінку та адаптувати власні рішення під свої потреби. Фреймворки прискорюють розробку, але це абстракції концепцій, які виявляються досить простими, коли їх розкривають повністю.