_agent_instantiation.py代码

from contextlib import contextmanager
from contextvars import ContextVar
from typing import Any, ClassVar, Generator

from ._agent_id import AgentId
from ._agent_runtime import AgentRuntime


class AgentInstantiationContext:
    def __init__(self) -> None:
        raise RuntimeError(
            "AgentInstantiationContext cannot be instantiated. It is a static class that provides context management for agent instantiation."
        )

    _AGENT_INSTANTIATION_CONTEXT_VAR: ClassVar[ContextVar[tuple[AgentRuntime, AgentId]]] = ContextVar(
        "_AGENT_INSTANTIATION_CONTEXT_VAR"
    )

    @classmethod
    @contextmanager
    def populate_context(cls, ctx: tuple[AgentRuntime, AgentId]) -> Generator[None, Any, None]:
        """:meta private:"""
        token = AgentInstantiationContext._AGENT_INSTANTIATION_CONTEXT_VAR.set(ctx)
        try:
            yield
        finally:
            AgentInstantiationContext._AGENT_INSTANTIATION_CONTEXT_VAR.reset(token)

    @classmethod
    def current_runtime(cls) -> AgentRuntime:
        try:
            return cls._AGENT_INSTANTIATION_CONTEXT_VAR.get()[0]
        except LookupError as e:
            raise RuntimeError(
                "AgentInstantiationContext.runtime() must be called within an instantiation context such as when the AgentRuntime is instantiating an agent. Mostly likely this was caused by directly instantiating an agent instead of using the AgentRuntime to do so."
            ) from e

    @classmethod
    def current_agent_id(cls) -> AgentId:
        try:
            return cls._AGENT_INSTANTIATION_CONTEXT_VAR.get()[1]
        except LookupError as e:
            raise RuntimeError(
                "AgentInstantiationContext.agent_id() must be called within an instantiation context such as when the AgentRuntime is instantiating an agent. Mostly likely this was caused by directly instantiating an agent instead of using the AgentRuntime to do so."
            ) from e

代码解释

这段代码定义了一个名为 AgentInstantiationContext 的类，用于管理智能体（Agent）实例化过程中的上下文信息。它主要解决了以下问题：

问题背景:

在智能体框架中，当创建一个新的智能体实例时，我们需要访问一些与当前实例化过程相关的信息，例如：

• 当前正在创建智能体的运行时环境 (AgentRuntime): 这可能包含智能体执行所需的各种资源、工具和服务。
• 当前正在创建智能体的 ID (AgentId): 每个智能体通常都有一个唯一的标识符。

如果直接实例化智能体，这些信息可能无法直接获取。我们需要一种机制来存储和检索这些信息，以便在智能体创建过程中使用。

代码逻辑及功能:

1. 禁止直接实例化:

   • __init__ 方法抛出 RuntimeError，明确表示 AgentInstantiationContext 不能被直接实例化。它是一个静态类，提供上下文管理功能。

2. 使用 ContextVar 存储上下文信息:

   • _AGENT_INSTANTIATION_CONTEXT_VAR 是一个类变量，类型为 ContextVar。ContextVar 是 Python 中用于在协程和异步编程中存储上下文信息的特殊变量。它能够跨异步任务边界保持值的不变性。
   • _AGENT_INSTANTIATION_CONTEXT_VAR 存储一个元组 (AgentRuntime, AgentId)，表示当前实例化上下文中的运行时环境和智能体 ID。

3. 上下文管理器 populate_context:

   • populate_context 是一个类方法，同时也是一个上下文管理器 (使用 @contextmanager 装饰器)。
   • 它接收一个包含 AgentRuntime 和 AgentId 的元组 ctx 作为参数。
   • 在进入 with 语句块时，它会使用 set 方法将 ctx 设置到 _AGENT_INSTANTIATION_CONTEXT_VAR 中，并返回一个令牌 (token)。
   • 在退出 with 语句块时，它会使用 reset 方法和之前获取的令牌来恢复 _AGENT_INSTANTIATION_CONTEXT_VAR 到之前的状态。

4. 获取当前上下文信息:

   • current_runtime 和 current_agent_id 都是类方法，用于获取当前上下文中的 AgentRuntime 和 AgentId。
   • 它们通过 get 方法从 _AGENT_INSTANTIATION_CONTEXT_VAR 获取当前值。
   • 如果 _AGENT_INSTANTIATION_CONTEXT_VAR 中没有值 (例如，不在 populate_context 创建的上下文中)，它们会抛出 RuntimeError，明确指出必须在智能体实例化上下文中调用这些方法。

总结:

这段代码通过 ContextVar 和上下文管理器机制，提供了一种在智能体实例化过程中安全地存储和访问上下文信息 (AgentRuntime 和 AgentId) 的方法。它确保了在智能体创建过程中可以可靠地获取这些信息，避免了直接实例化智能体可能带来的问题。这对于构建可靠的智能体框架至关重要。

代码示例

这段代码定义了一个名为 AgentInstantiationContext 的类，用于在智能体（Agent）实例化过程中提供上下文管理。它使用 contextvars.ContextVar 来存储当前实例化智能体的运行时环境和智能体 ID，并确保这些信息在实例化过程中始终可用。

以下是几个体现该代码功能的代码示例：

示例 1：使用 populate_context 正确设置上下文

from autogen_core import AgentInstantiationContext, SingleThreadedAgentRuntime, AgentId,AgentRuntime

# 假设 AgentRuntime 和 AgentId 已经定义

runtime = SingleThreadedAgentRuntime()
agent_id = AgentId("agent_123","faf")

with AgentInstantiationContext.populate_context((runtime, agent_id)):
    # 在这个上下文中，current_runtime 和 current_agent_id 可以正常工作
    print(AgentInstantiationContext.current_runtime())  # 输出: <AgentRuntime object>
    print(AgentInstantiationContext.current_agent_id())  # 输出: agent_123

<autogen_core._single_threaded_agent_runtime.SingleThreadedAgentRuntime object at 0x000002316819F260>
agent_123/faf

示例 2：尝试在上下文之外调用 current_runtime 和 current_agent_id

try:
    AgentInstantiationContext.current_runtime()
except RuntimeError as e:
    print(e)  # 输出错误信息，说明必须在上下文中调用

try:
    AgentInstantiationContext.current_agent_id()
except RuntimeError as e:
    print(e)  # 输出错误信息，说明必须在上下文中调用

AgentInstantiationContext.runtime() must be called within an instantiation context such as when the AgentRuntime is instantiating an agent. Mostly likely this was caused by directly instantiating an agent instead of using the AgentRuntime to do so.
AgentInstantiationContext.agent_id() must be called within an instantiation context such as when the AgentRuntime is instantiating an agent. Mostly likely this was caused by directly instantiating an agent instead of using the AgentRuntime to do so.

示例 3：模拟 AgentRuntime 使用 AgentInstantiationContext

class MyAgentRuntime(AgentRuntime):
    def create_agent(self, agent_id: AgentId):
        with AgentInstantiationContext.populate_context((self, agent_id)):
            # 在这里，我们可以使用 AgentInstantiationContext.current_runtime() 和 AgentInstantiationContext.current_agent_id()
            print(AgentInstantiationContext.current_runtime())  # 输出: <AgentRuntime object>
            print(AgentInstantiationContext.current_agent_id())

runtime = MyAgentRuntime()
agent_id = AgentId("agent_123","faf")
runtime.create_agent(agent_id)

<__main__.MyAgentRuntime object at 0x0000023166A731A0>
agent_123/faf

这些示例展示了如何正确使用 AgentInstantiationContext 来管理智能体实例化过程中的上下文，以及如果在上下文之外使用它会发生什么错误。