AI-агент з нуля на Python: цыкл GAME крок за крокам
Будуем працоўнага агента з выклікам інструментаў на чыстым Python з мэтамі, дзеяннямі, памяццю і асяроддзем, а потым дадаём рэестр інструментаў, схаванае ўкараненне залежнасцей і тыпізаваны вывад.
Фрэймворкі агентаў, такія як Pydantic AI, AutoGen і Microsoft Agent Framework, падходзяць для шырокага прыемлівання пратотыпа за скоры час. Аднак пасля выкарыстоўвання калькі з іх, часта выяўляюцца адзін-два прыблудныя моменты.
- Экасістэма яшчэ развіваецца. Кожны фрэймворк падтрымляе частку неабходных функцыяй і не падтрымляе іншых, а іншы фрэймворк мае тыя элементы, якіх не хапяць у першам. Гэта нормальна для моладога направлення, і з часам большасць з іх будзе справляцца з распашчытанымі кейсамі выкарыстоўвання.
- Абстракцыі накапліваюцца. Фрэймворкі ставяць за мету падтрымка вельмі шырокага спектру дызайнаў агентаў, таму яны дадаюць шары, якія ўскладнююць розумеў і дыбаггіранне. LangChain — тыповы прыклад: калі сістэма працюе некорэктна, гэтыя шары ствараюць пераканстанні для выяўлення прычыны.
Пахвальнае ў тым, што агент не павінен быць складным. Для багато задач дастатнека простай рэалізацыі. Стварэнне агента з нуля дае вам палную контроль над тым, што прайшоў да модэля, як прымуцца рашэнні і як кантролюваць вітраты, а знанне прынцыпаў спрыяе большай доверлівасці да будзь-каго пазнейшага фрэймворку. Далей паказвана пошаговая стварэння працоўнага Python-агента на адных засадах.
Асновная ідэя: цыкл агента
Пачатакніце з спрошчэння ідэі. Агент AI — гэта большая мадэль мовы, якая можа выконваць дзеяння і выкарыстоўваць ўсунутыя рэзультаты, каб адлучыць, што робіць далей. У традыцыйным абмене з 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, у тым чым 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 для парадоксу і перадае рэзультат да наступнага крока ціклу.
Памяць
Памяць зберагае тое, што было скарыстана ў пакульных ітэрацыях, таму агент можа взяць у расчытанне пакульныя назірання пры выборэ наступнага крока.
Наразы ёй выступае мінімум: новыя паведамлення проста дадаюцца да спісу. Вікна кантэксту LLM ёсць скончанымі, таму рэальная сістэма патрэбуе чагось цікавейшага для скарычвання чы ўзводу, што ў гэтым прыкладзе не расследжваецца.
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 LLM, таму стварэнне запытоў на аднойчынае ўзятыя з історыі ёсць простым, а памяць лёгка розшырваецца.
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)}
Стварэнне цыклу агента
Пасля таго, як вярнуліся чатыры элемента ГЕЙМ, можна запісаць основны цыкл. Ён складае запіт, прасіць 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 для LLM выкарыстоваеяць стандарт, па якому рэзультат абладнання павінен быць прыязначаны да абладнання, якое гэта практыкувалася ранейш, зазвычай праз tool_id; такім чынам той самы ідэнтыфікатор праказуецца два разы ў історыі: аднажды ў запіце і аднажды ў адпаведным выніку.
Сучасныя мадэлі таксама атрыбутаваны значна на працэсах вызывання абладнань. Павторны выкарыстоўкі ролей і шаблонаў паведамленняў, якія яны бачылі пад час атрыбутавання, падходзяць да ўсіх іх аспактаванняў і зазвычай даюць стабільнейшае вызывання абладнань і лепшую загальную карыстнасць.
Спеціальній інструмент finish таксама заслуговуе на увагу. Ён дае мадэлі чыткій спосіб позначыты завершэння задачы, шчоб яна не застрагала у безкінечніх вікліках інструментаў. Ён рэеструецца як будзь-які іншый інструмент у кожному агенты, што ёсць перэвагою для навчання.
Дэ перша версія мае недолікі
Гэты агент працюе, але ён розкрывае всю структуру цыклу без жодніх прыхованіх элементаў у рамках фрэймворку, што ёсць ідэальнім для навчання. Однак у прыкладах з мінулого швідко проявляюцца тры обмежэння.
- Візначэння інструментаў не масштабуюцца. Кожна дыя вімагае ручного напісання JSON Schema, якое мае буты сінхронізоване з відповідною функціею на 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
):
...
Якбы кантэкст быў проста ўсьмою іншым аргументам інструмента, LLM быў змушаны сам ствараць гэтыя значэнні, што не мае сэнсу, адколі яны павінны быць заданы ззаўні. Сюды і падае на дапамогу 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-шыму, таму LLM не ведае пра яго наявнасць і не можа прабавіцца яго стварыць. 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, які змушвае LLM выдаваць JSON, які адпавядае шыму. Гэта суперчыць цыклу вызываў інструментаў. Калі ўвімкнуты response_format, мадэль зазвычай адразу піша свой заканчыцковы адказ уместо таго, каб вызваць інструменты, і агент зупіняецца пасля першага заполнення. Спалучэнне гэтых двух падходаў таксама практычна выявілася некомфортным і нестабільным.
Альтэрнатыва — паўтарна адналіквеннае вызыванне інструментаў. Шым выходу стаюць параметры спецыяльнага інструмента, і калі праца завершана, LLM вызывае гэты інструмент з аргументамі, якія адпавядаюць бажанаму шыму. Агент выкарыстоўвае гэтыя аргументы і стварае заканчыцковы об’ект выходу.
Памагальны элемент класу 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 і большасць іншых рашыткі створаны на адной і той жа модэлі.
Пашагоўна просты цікл вызову інструмента ператворыўся на болей можнацькі механізм. Рэестр інструментаў дазволіў вярнаваць іспользоўваныя дзеянні, шар срэдавішча безпечна дадаў скрытыя залежнасці, а структураваны механізм завершэння дазволіў агентам вярнаваць выходны тэкст у фармаце з пазначаннямі тыпаў. Ніякі з гэтых прыемаў сам по сабе не ёсць складным, але разам яны станавяць адну з базаў сучаснага дызайну агентоў.
Знання цых основ допомагае легшэ оцінюваты фрэймворкі, простышэ відловлюваты незвічайну поведынку і ефектывнішэ прыстосовуваты власні праекты під вашы потрэбі. Фрэймворкі прыскоруюць распрацоўку, але гэта абстракціі концепцій, які віявляюцца досіць простымі, колі іх розглядаты відкрыто.