Ein KI-Agent von Grund auf in Python entwickeln: Der GAME-Loop Schritt für Schritt
Erstellen Sie einen funktionsfähigen Tool-Aufruf-Agenten in reinem Python mit Zielen, Aktionen, Speicher und Umgebung, fügen Sie anschließend ein Tool-Register, versteckte Abhängigkeitsinjektion sowie typisierte Ausgaben hinzu.
Agent-Frameworks wie Pydantic AI, AutoGen und Microsoft Agent Framework eignen sich gut, um schnell ein Prototyp zu erstellen. Nach der Verwendung mehrerer solcher Frameworks tauchen jedoch oft einige Probleme auf.
- Das Ökosystem entwickelt sich weiter. Jedes Framework deckt einige der benötigten Funktionen ab und lässt andere aus, während ein anderes Framework die fehlenden Komponenten bietet. Das ist bei einem jungen Feld normal, und mit der Zeit werden die meisten Frameworks die gängigsten Anwendungsfälle bewältigen können.
- Es sammelt sich Abstraktion an. Frameworks sollen eine breite Palette an Agent-Designs unterstützen, weshalb sie Schichten hinzufügen, die schwer nachvollziehbar und noch schwieriger zu debuggen sind. LangChain ist ein typisches Beispiel: Wenn das System falsch funktioniert, behindern diese Schichten die Diagnose des Problems.
Die ermutigende Seite ist, dass ein Agent nicht unbedingt kompliziert sein muss. Für viele Aufgaben reicht eine einfache Implementierung aus. Wenn man einen Agent von Grund auf entwickelt, hat man volle Kontrolle darüber, was ins Modell eingeht, wie Entscheidungen getroffen werden und wie Kosten gemanagt werden. Zudem macht das Verständnis der zugrundeliegenden Prinzipien es später einfacher, Frameworks zu vertrauen. Im Folgenden wird schrittweise gezeigt, wie man einen funktionsfähigen Python-Agenten von den Grundlagen aus erstellt.
Die Kernidee: der Agent-Zyklus
Beginnen Sie damit, die Idee zu vereinfachen. Ein KI-Agent ist ein großes Sprachmodell, das Aktionen ausführen und die Ergebnisse nutzen kann, um zu entscheiden, was als Nächstes getan werden soll. Bei einem herkömmlichen Austausch mit einem LLM senden Sie eine Anfrage und erhalten eine Antwort zurück. Ein Agent hingegen arbeitet in einer Schleife: Er betrachtet den aktuellen Zustand, wählt eine Aktion aus, führt sie aus, erhält Feedback und beginnt den Prozess erneut, bis die Aufgabe abgeschlossen ist.
GAME: ein vierteiliges Design
Um diese Schleife zu erstellen, verwenden wir die GAME-Struktur, die einen Agenten in Ziele, Aktionen, Gedächtnis und Umgebung unterteilt. Sie liefert ein kompaktes mentales Modell und sorgt dafür, dass der Code modular bleibt und leicht erweiterbar ist.
Ziel
Das Ziel beschreibt, was der Agent erreichen soll. Es enthält die Beschreibung der Aufgabe sowie alle Anweisungen zur Steuerung des Verhaltens und wird zum Systemprompt.
class Goal(BaseModel):
name: str
description: str
Aktion (Werkzeug)
In der Codebasis wird dies als Tool bezeichnet, doch die Begriffe bedeuten dasselbe. Eine Aktion ist etwas, was der Agent ausführen kann: eine API aufrufen, Dokumente durchsuchen, in einer Datenbank nachfragen oder eine Berechnung durchführen.
class Tool(BaseModel):
name: str
description: str
parameters: Dict[str, Any]
function: Callable
tags: Optional[List[str]] = None
terminal: bool = False
Die aktuellen LLM-APIs, einschließlich denen von OpenAI, akzeptieren zusammen mit dem Prompt auch Tool-Definitionen. Diese Definitionen müssen im JSON Schema-Format verfasst sein und den Namen jedes Tools, seine Parameter sowie die erwarteten Eingaben angeben.
Als erstes Tool nehmen wir eine Funktion, die zwei Zahlen addiert:
def add(a: float, b: float) -> float:
return a + b
Dann beschreiben Sie es in einer Form, die das Modell lesen kann:
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,
)
Wenn die Anfrage erstellt wird, werden diese Aktionselemente in Tool-Definitionen umgewandelt und an das Modell gesendet. Das Modell führt die Funktion selbst niemals aus. Es entscheidet lediglich, welches Tool aufgerufen werden soll, und generiert die entsprechenden Argumente dafür.
Nehmen wir an, der Benutzer fragt „Wie viel ist 10 + 20?“ Das Modell könnte mit etwas Ähnlichem antworten:
{
"name": "add",
"arguments": {
"a": 10,
"b": 20
}
}
Die Auswahl der Aktion und das Bereitstellen ihrer Argumente ist die gesamte Aufgabe des Modells.
Aktionen an das Modell senden
Dieser Hilfsfunktion übermittelt den Konversationsverlauf sowie die verfügbaren Aktionen an das 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": []
}
Dann ruft der Agent die entsprechende Python-Funktion auf und gibt das Ergebnis in den nächsten Schritt des Loops ein.
Gedächtnis
Das Gedächtnis speichert das, was in früheren Iterationen gesammelt wurde, sodass der Agent vergangene Beobachtungen berücksichtigen kann, wenn er seinen nächsten Schritt wählt.
Für den Moment bleibt es minimal: Neue Nachrichten werden einfach einer Liste hinzugefügt. Die Kontextfenster von LLMs sind begrenzt, weshalb ein echtes System etwas Intelligenteres zum Kürzen oder Zusammenfassen benötigt – was in diesem Beispiel weggelassen wird.
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
Ein Message-Modell repräsentiert jede Interaktion im Loop. Beobachtungen, Tool-Ergebnisse, Benutzereingaben sowie Assistentenantworten werden alle als Nachrichten abgebildet. Die Struktur folgt eng dem Nachrichtenformat der aktuellen LLM-APIs, sodass das Erstellen von Prompts auf Basis der Historie unkompliziert ist und das Speichersystem einfach erweiterbar bleibt.
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
Umgebung
Sie fragen sich vielleicht, warum überhaupt eine Umgebung notwendig ist, wenn die Aktion einfach in ihrer eigenen Klasse ausgeführt werden könnte.
class Action:
...
def execute_action(self, **args):
return function(**args)
Das ist in Ordnung für Beispiele mit Spielzeugkomponenten. Es wird jedoch begrenzend, sobald man Abhängigkeitsinjektion, versteckten Kontext, Datenbankverbindungen, Benutzersitzungen oder andere externe Ressourcen benötigt – die späteren Abschnitte zeigen, wie das funktioniert.
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)}
Aufbau des Agenten-Zyklus
Sobald die vier GAME-Bestandteile definiert sind, kann der Kernzyklus geschrieben werden. Er erstellt eine Anfrage, bittet den LLM um eine Antwort, führt die ausgewählten Aktionen aus, sammelt Beobachtungen ein und wiederholt diesen Vorgang, bis eine Stoppsbedingung erfüllt ist.
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"
}
]
Eine kurze Anmerkung zu den Details. Das Nachrichtenformat könnte anders als die hier verwendete "tool"-Rolle gestaltet werden, doch die Struktur würde dann kleine Anpassungen erfordern. Die meisten LLM-APIs folgen der Konvention, dass das Ergebnis eines Tools mit dem vorangegangenen Toolaufruf verknüpft werden muss, in der Regel über einen tool_id; daher erscheint derselbe Identifikator zweimal in der Historie: einmal im Anfrage- und einmal im entsprechenden Ergebnis.
Moderne Modelle werden außerdem intensiv an Tool-Aufruf-Workflows trainiert. Die Wiederverwendung der in der Ausbildung gesehenen Rollen und Nachrichtenmuster entspricht ihren Erwartungen und führt in der Regel zu stabileren Tool-Aufrufe sowie einer besseren Gesamtleistung.
Das spezielle finish-Tool verdient ebenfalls Erwähnung. Es gibt dem Modell eine explizite Möglichkeit, anzuzeigen, dass die Aufgabe abgeschlossen ist, sodass es nicht ewig damit beschäftigt ist, Tools aufzurufen. Es wird wie jedes andere Tool bei jedem Agenten registriert, was ein Designaspekt ist, der bald verbessert werden wird.
Wo die erste Version nachlässt
Dieser Agent funktioniert und macht den gesamten Ablauf sichtbar, ohne etwas hinter einem Framework zu verbergen – was ideal zum Lernen ist. Abgesehen von den ersten Beispielen treten schnell drei Einschränkungen auf.
- Die Tool-Definitionen sind nicht skalierbar. Jede Aktion erfordert ein manuell erstelltes JSON Schema, das mit der zugehörigen Python-Funktion synchron gehalten werden muss. Wenn die Anzahl der Tools zunimmt, wird dies repetitiv und fehleranfällig; außerdem kann eine kleine Änderung in der Signatur dazu führen, dass das Schema veraltet wird, wodurch das Modell ungültige Aufrufe erstellt.
finish-Tool stoppt die Schleife und gibt ein Ergebnis zurück, doch jedes Agent muss dies registrieren, und es kann nur einen einfachen Wert liefern. Leistungsfähigere Agenten benötigen detailliertere Beendigungsvorschriften sowie strukturierte Ausgaben.Jeder Punkt wird nacheinander behandelt, beginnend mit der Verwaltung und Registrierung von Tools.
Problem 1: ein Tool-Register
Man beginnt mit dem lästigsten Aspekt und passt Schema sowie Funktionen manuell an. Completion-APIs verlangen eine Liste von JSON-Schemata, die die verfügbaren Tools beschreiben.
In der ersten Version kam jedes Tool als zwei getrennt geschriebene Komponenten: eine Python-Funktion, die die Arbeit ausführt, und ein JSON Schema, das dem LLM mitteilt, wie es darauf zugreifen soll:
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,
)
Die Lösung besteht darin, das Schema aus der Funktion zu erzeugen. Die Funktion wird zur einzigen Quelle der Wahrheit, und das Schema wird automatisch abgeleitet, wenn das Tool registriert wird.
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)
Danach sollte dieser Builder automatisch ausgeführt werden, sobald eine Tool-Funktion definiert wird – daher übernimmt ein Dekorator diese Aufgabe:
@tool()
def add(a: float, b: float) -> float:
"""Adds two floats."""
return a + b
Der Dekorator ist absichtlich bescheiden. Er erstellt das Tool, fügt es dem Funktionselement hinzu und gibt die Funktion unverändert zurück:
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)
Outprobieren:
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
}
Sobald es viele Tools und mehrere Agenten gibt, die die Tools teilen, hilft ein zentrales Ort zur Verwaltung. Eine Action Registry übernimmt diese Rolle, indem sie verfolgt, welche Tools ein bestimmter Agent verwenden darf.
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()
]
Die Registrierung einer Funktion erfolgt dann durch einen einzigen Aufruf:
registry.register(add)
Das Registry liest die Metadaten, die @tool hinzugefügt hat, und wandelt sie in eine Tool-Instanz um. Für Funktionen, die nicht dekoriert wurden, greift es automatisch auf den Aufruf von build_tool() zurück. Das ist eine praktische Standardlösung, und diejenigen, die vollständige Kontrolle wünschen, können Tool-Objekte weiterhin manuell erstellen.
Bemerken Sie außerdem, dass terminal bei der Registrierung festgelegt wird und nicht am Tool selbst:
registry.register(tool, terminal=True)
Ob ein Werkzeug den Loop beendet, ist eine Eigenschaft des Agents und nicht der Funktion. Dasselbe Werkzeug kann für einen Agenten terminierend sein und für einen anderen gewöhnlich. Wenn ein anderer terminal-Wert angefordert wird, kopiert das Register das Tool anstelle dessen, den Originalwert zu ändern – so dringen Konfigurationen nicht zwischen Registern durch, und eine Definition kann sicher über verschiedene Agenten wiederverwendet werden.
Problem 2: Einfügen von Abhängigkeiten
Werkzeuge funktionieren derzeit unter einer wichtigen Einschränkung: Sie können nur Argumente sehen, die vom LLM erzeugt werden.
Praktische Werkzeuge benötigen oft Ressourcen, die dem Modell niemals gezeigt werden sollten – darunter Datenbankverbindungen, API-Cliente, Konfigurationsobjekte, Benutzersitzungen und Anwendungsstatus. Einige müssen außerdem auf das Gedächtnis des Agents zugreifen. Stellen Sie sich ein Werkzeug vor, das Forschungspapiere in eine Datenbank speichert:
@tool()
def save_paper(
title: str,
content: str,
database_url: str,
api_key: str,
):
...
Um das besser darzustellen, enthält eine Klasse den Aktionskontext:
class ActionContext(BaseModel):
database_url: str,
api_key: str
# Then the tool becomes:
@tool()
def save_paper(
title: str,
content: str,
action_context: ActionContext
):
...
Falls der Kontext einfach ein weiteres Argument für ein Tool wäre, müsste das LLM diese Werte selbst erfinden, was keinen Sinn macht, da sie eigentlich injiziert werden sollen. Hier kommt Environment zum Einsatz. Die Kontextparameter bleiben vor dem Modell verborgen und werden zur Ausführungszeit manuell eingefüllt.
Da einige Tools einen Kontext benötigen und andere nicht, ist der erste Schritt eine Funktion, die anhand der Typangaben prüft, ob eine Funktion einen ActionContext oder eine seiner Unterklassen entgegennimmt:
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 []
)
Ein Tool, das einen ActionContext-Parameter verwendet, lässt diesen Parameter in seinem JSON-Schema aus, sodass das LLM nicht weiß, dass er existiert, und ihn daher nicht generieren kann. Der Environment fügt den Kontext dann zur Ausführungszeit wieder hinzu:
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)}
Das ist der Trick. Die Umgebung fungiert als Grenze zwischen dem Modell und der Außenwelt. Das Modell liefert die Argumente, von denen es weiß, und die Umgebung fügt die Ressourcen hinzu, von denen es nichts wissen sollte. Es handelt sich um eine einfache Idee, die Agenten erheblich praktischer macht.
Problem 3: strukturierte Ausgabe und ein integriertes Beendigungstool
Zwei Probleme bleiben bestehen. Erstens kann der Agent derzeit nur Zeichenketten zurückgeben, während reale Arbeitsabläufe oft strukturierte Daten benötigen. Zweitens sollte das Beendigungstool in den Agenten integriert sein und nicht jedes Mal extern registriert werden.
Sowohl Probleme werden gemeinsam gelöst. Für eine strukturierte Ausgabe bieten viele Completion-APIs einen response_format-Parameter, der das LLM zwingt, JSON im Einklang mit einem Schema auszugeben. Dies steht im Widerspruch zu einem Tool-Aufruf-Loop. Sobald response_format aktiviert ist, schreibt das Modell in der Regel sofort seine endgültige Antwort anstelle dessen, Tools aufzurufen, und der Agent stoppt nach der ersten Vervollständigung. Die Kombination beider Ansätze hat sich in der Praxis ebenfalls als unpraktisch und inkonsistent erwiesen.
Die Alternative besteht darin, die Tool-Aufrufe erneut zu nutzen. Das Ausgabeschema wird zu den Parametern eines speziellen Tools, und sobald die Arbeit abgeschlossen ist, ruft das LLM dieses Tool mit Argumenten auf, die dem gewünschten Schema entsprechen. Der Agent extrahiert diese Argumente und erstellt das endgültige Ausgabeobjekt.
Ein Hilfsfunktion in der Agent-Klasse registriert automatisch ein eingebautes terminate-Tool:
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
Und hier ist der fertige Agenten-Loop:
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
Was die Übung zeigt
Im Grunde genommen ist ein Agent ein Sprachmodell, das in einer Schleife läuft. Es liest die Situation ein, wählt eine Aktion aus, erhält das Ergebnis und wiederholt den Vorgang, bis das Ziel erreicht ist. Pydantic AI, AutoGen, Microsoft Agent Framework sowie die meisten anderen Lösungen basieren auf diesemselben Muster.
Schritt für Schritt entwickelte sich eine einfache Schleife zur Aufrufung von Tools zu einer leistungsfähigeren Struktur. Ein Tool-Register machte Aktionen wiederverwendbar, eine Umgebungs-Schicht injizierte versteckte Abhängigkeiten sicher, und ein strukturierter Beendigungsmechanismus ermöglichte es Agenten, anstelle von reinem Text typisierte Ausgaben zurückzugeben. Keine dieser Ideen ist für sich genommen schwierig, doch zusammen bilden sie die Grundlage vieler moderner Agentenkonzepte.
Diese Grundkonzepte zu kennen macht es einfacher, Frameworks zu beurteilen, ungewöhnliches Verhalten zu debuggen und eigene Designs an die eigenen Bedürfnisse anzupassen. Frameworks beschleunigen die Entwicklung, doch es handelt sich dabei um Abstraktionen von Konzepten, die, sobald sie offen dargestellt werden, als ziemlich einfach erweisen.