用Python从零构建AI智能体：逐步讲解游戏循环
使用包含目标、动作、记忆体和环境结构的纯Python构建一个可运行的工具调用代理，然后再添加工具注册表、隐藏的依赖注入机制以及类型化输出功能。
Pydantic AI、AutoGen以及Microsoft Agent Framework这类智能体框架有助于快速构建原型。不过在试用了多个之后，往往会发现一些问题。
- 生态系统仍在发展之中。每个框架都能覆盖部分所需功能，同时也会遗漏其他功能，而另一个框架则可能具备前一个所缺乏的功能。对于一个新兴领域来说这是正常现象，随着时间推移，大多数框架都能应对常见的使用场景。
- 抽象层次日益增多。这些框架旨在支持多种多样的智能体设计，因此会添加许多难以理解且更难调试的层次结构。LangChain就是典型的例子：当系统出现异常时，这些层次结构反而会妨碍人们找出问题所在。
令人欣慰的是，代理并不一定非得十分复杂。对于许多任务来说，简单的实现就已经足够。从零开始编写代理可以让你对输入模型的内容、决策方式以及成本管理拥有完全的控制权，而且了解相关原理也能让后续使用的任何框架都更易被信任。以下将从基础原理出发，逐步介绍如何构建一个可用的Python代理。
核心理念：代理循环
首先需要简化这个概念。AI智能体是一种大型语言模型，它能够执行动作，并利用行动结果来决定下一步该做什么。在传统的LLM交互中，你发送一个提示语就会得到一个回复。而智能体则是以循环方式运行：它查看当前状态，选择一项动作并执行，接收反馈后再次循环，直到任务完成。
GAME：四部分设计结构
为了构建这种循环机制，我们将使用GAME结构，该结构将智能体划分为目标、动作、记忆和环境四个部分。这种结构能够提供简洁的思维模型，同时让代码保持模块化且易于扩展。
目标
目标明确了智能体试图实现的内容。它包含任务描述以及用于指导行为的各种指令，同时也会成为系统的提示语。
class Goal(BaseModel):
name: str
description: str
操作（工具）
这里的代码库将其称为Tool，但这两个术语含义相同。操作指的是智能体能够执行的任务：调用API、搜索文档、查询数据库或进行计算。
class Tool(BaseModel):
name: str
description: str
parameters: Dict[str, Any]
function: Callable
tags: Optional[List[str]] = None
terminal: bool = False
当前的LLM API，包括OpenAI的API，允许在提示语中同时提交工具定义。这些定义必须以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
}
}
选择操作并提供其参数就是模型的全部工作。
将操作发送给模型
此辅助函数会将对话历史和可用操作传递给大语言模型：
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组件之后，就可以编写核心循环了。该循环负责生成提示语、向大语言模型请求回复、执行选定的动作、收集观察结果，直至满足停止条件为止。
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 函数保持同步。随着工具数量的增加，这种做法既重复又容易出错，而且细微的签名变化都可能导致 Schema 过时，从而使模型生成无效的调用。
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以及大多数其他智能体框架都是基于相同的模式构建的。
最初只是一个简单的工具调用循环，后来逐渐发展出更强大的功能。工具注册表使得各种行动可以重复使用，环境层能够安全地注入隐藏的依赖项，而结构化的终止机制则让智能体能够返回带类型信息的输出而非纯文本。这些概念单独来看都不难，但结合起来却构成了现代智能体设计的基础。
了解这些基础概念后，就能更轻松地评估框架性能、调试异常行为，也能更好地让自己的设计符合需求。框架虽能加快开发速度，但它们本质上是概念的抽象表现，一旦详细展示出来就会发现其实相当简单。