Создание ИИ-агента с нуля на Python: цикл GAME шаг за шагом
Создайте рабочего агента для вызова инструментов на чистом Python с механизмами целей, действий, памяти и окружения, затем добавьте реестр инструментов, скрытую инъекцию зависимостей и типизированный вывод.
Источники: How to build AI Agent From Scratch in Python
Фреймворки агентов, такие как Pydantic AI, AutoGen и Microsoft Agent Framework, отлично подходят для быстрой разработки прототипов. Однако после использования нескольких из них обычно возникают определённые проблемы.
- Экосистема всё ещё находится в развитии. Каждый фреймворк покрывает часть необходимых функций и упускает другие, а другой фреймворк имеет те компоненты, которых нет в первом. Это нормально для молодой области, и со временем большинство из них смогут обрабатывать распространённые сценарии использования.
- Накапливается абстракция. Фреймворки стремятся обслуживать огромное количество вариантов проектирования агентов, поэтому они добавляют слои, которые сложно понимать и ещё сложнее отлаживать. LangChain — типичный пример: когда система ведёт себя некорректно, эти слои мешают выяснить причину.
Положительным моментом является то, что агент не обязательно должен быть сложным. Для многих задач достаточно простой реализации. Разработка агента с нуля позволяет полностью контролировать то, что поступает в модель, как принимаются решения и как управляются затраты, а знание основных принципов делает любую последующую платформу более надежной. Далее приведена пошаговая инструкция по созданию рабочего Python-агента на основе первоначальных принципов.
Основная идея: цикл агента
Начните с упрощения концепции. ИИ-агент — это большая модель языка, способная выполнять действия и использовать полученные результаты для принятия решения о дальнейших шагах. В традиционном взаимодействии с большой моделью языка вы отправляете запрос и получаете один ответ. Агент же работает в цикле: он анализирует текущее состояние, выбирает действие, выполняет его, получает обратную связь и снова возвращается к циклу до завершения задачи.
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 больших языковых моделей, включая 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 Schema, которая должна синхронизироваться с соответствующей функцией на Python. По мере увеличения количества инструментов это становится трудоемким и подверженным ошибкам, а небольшое изменение сигнатуры может привести к устареванию структуры, в результате чего модель будет формировать недействительные вызовы.
finish останавливает цикл и возвращает результат, но каждый агент должен его зарегистрировать, причем он может возвращать только простое значение. Более мощные агенты будут нуждаться в более сложных правилах остановки и структурированном выводе.Каждый аспект рассматривается по очереди, начиная с способов управления и регистрации инструментов.
Проблема 1: реестр инструментов
Начнем с самой проблематичной части — вручную синхронизируем схемы и функции. API для завершения работы требуют списка JSON-схем, описывающих доступные инструменты.
В первой версии каждый инструмент представлял собой два отдельно написанных элемента: функцию на Python, выполняющую работу, и JSON Schema, указывающего большую языковую модель, как её вызвать:
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: внедрение зависимостей
В настоящее время инструменты работают при соблюдении одного важного ограничения: они могут обрабатывать только аргументы, генерируемые большой языковой моделью.
Практические инструменты часто нуждаются в ресурсах, которые никогда не должны быть видны модели, включая подключения к базам данных, клиенты 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 и большинство других решений построены на основе этой же схемы.
Шаг за шагом простой цикл вызова инструментов превратился в более мощный механизм. Реестр инструментов позволил сделать действия повторно используемыми, слой окружения безопасно внедрял скрытые зависимости, а структурированный механизм завершения позволял агентам возвращать типизированные данные вместо простого текста. Ни одна из этих идей сама по себе не является сложной, но вместе они лежат в основе современного проектирования агентов.
Знание этих основ позволяет легче оценивать фреймворки, быстрее отлаживать необычное поведение и адаптировать собственные решения под свои потребности. Фреймворки ускоряют разработку, но это абстракции концепций, которые оказываются довольно простыми, когда их полностью раскрывают.