通过

运行时上下文

运行时上下文提供中间件对有关当前执行环境和请求的信息的访问权限。 这可实现基于运行时条件的模式，例如按会话配置、特定于用户的行为和动态中间件行为。

在 C# 中，运行时上下文通常通过 AgentRunOptions 或自定义会话状态传递。 中间件可以访问会话属性并运行选项以做出运行时决策。

小窍门

有关中间件范围如何影响对运行时上下文的访问的信息，请参阅 “代理与运行范围 ”页。

Python 运行时上下文拆分为三个公共图面：

  • session= 用于聊天状态和历史记录。
  • function_invocation_kwargs= 对于仅应看到工具或函数中间件的值。
  • client_kwargs= 用于聊天客户端特定的数据或客户端中间件配置。

使用适合数据的最小图面。 这会使工具输入保持显式，并避免将仅客户端元数据泄露到工具执行中。

小窍门

被视为function_invocation_kwargs将任意公共**kwargsagent.run()传递给或get_response()传递旧模式的替代。

选择正确的运行时存储桶

用例 API 图面 访问自
共享会话状态、服务会话 ID 或历史记录 session= ctx.session、AgentContext.session
仅传递工具或函数中间件所需的运行时值 function_invocation_kwargs= FunctionInvocationContext.kwargs
传递特定于客户端的运行时值或客户端中间件配置 client_kwargs= 自定义 get_response(..., client_kwargs=...) 实现

传递仅限工具的运行时值

from typing import Annotated

from agent_framework import FunctionInvocationContext, tool
from agent_framework.openai import OpenAIResponsesClient


@tool(approval_mode="never_require")
def send_email(
    address: Annotated[str, "Recipient email address."],
    ctx: FunctionInvocationContext,
) -> str:
    user_id = ctx.kwargs["user_id"]
    tenant = ctx.kwargs.get("tenant", "default")
    return f"Queued email for {address} from {user_id} ({tenant})"


agent = OpenAIResponsesClient().as_agent(
    name="Notifier",
    instructions="Send email updates.",
    tools=[send_email],
)

response = await agent.run(
    "Email the launch update to finance@example.com",
    function_invocation_kwargs={
        "user_id": "user-123",
        "tenant": "contoso",
    },
)

print(response.text)

在工具内部使用 ctx.kwargs ，而不是在可调用的工具上声明毛毯 **kwargs 。 旧 **kwargs 版工具仍可用于兼容性，但在正式发布之前将被删除。

任何注释为 FunctionInvocationContext 注释的参数都被视为注入的运行时上下文参数，而不考虑其名称，也不会在向模型显示的 JSON 架构中公开。 如果提供显式架构/输入模型，则命名 ctx 为纯非批注参数也将被识别为注入的上下文参数。

如果值是生存期较长的工具状态或依赖项，而不是按调用数据，请将其保留在工具类实例上，而不是传递 function_invocation_kwargs它。 有关该模式，请参阅 使用多个函数工具创建类。

函数中间件接收相同的上下文

函数中间件使用工具接收的同一 FunctionInvocationContext 对象。 这意味着中间件可以检查context.arguments、context.kwargs和context.sessioncontext.result。

from collections.abc import Awaitable, Callable

from agent_framework import FunctionInvocationContext
from agent_framework.openai import OpenAIResponsesClient


async def enrich_tool_runtime_context(
    context: FunctionInvocationContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    context.kwargs.setdefault("tenant", "contoso")
    context.kwargs.setdefault("request_source", "middleware")
    await call_next()


agent = OpenAIResponsesClient().as_agent(
    name="Notifier",
    instructions="Send email updates.",
    tools=[send_email],
    middleware=[enrich_tool_runtime_context],
)

中间件协定不使用 call_next() 任何参数。 在调用它之前发生 context.kwargs 变异，所选工具通过注入 FunctionInvocationContext来查看这些值。

用于 session= 共享运行时状态

from typing import Annotated

from agent_framework import FunctionInvocationContext, tool
from agent_framework.openai import OpenAIResponsesClient


@tool(approval_mode="never_require")
def remember_topic(
    topic: Annotated[str, "Topic to remember."],
    ctx: FunctionInvocationContext,
) -> str:
    if ctx.session is None:
        return "No session available."

    ctx.session.state["topic"] = topic
    return f"Stored {topic!r} in session state."


agent = OpenAIResponsesClient().as_agent(
    name="MemoryAgent",
    instructions="Remember important topics.",
    tools=[remember_topic],
)

session = agent.create_session()
await agent.run("Remember that the budget review is on Friday.", session=session)
print(session.state["topic"])

显式 session= 传递会话并从中 ctx.session读取会话。 会话访问不再需要通过运行时 kwargs 进行。

与委派代理共享会话状态

当代理作为 as_tool()工具公开时，运行时函数 kwargs 已流经 ctx.kwargs。 仅当子代理应共享调用方AgentSession时添加propagate_session=True。

from agent_framework import FunctionInvocationContext, tool
from agent_framework.openai import OpenAIResponsesClient


@tool(description="Store findings for later steps.")
def store_findings(findings: str, ctx: FunctionInvocationContext) -> None:
    if ctx.session is not None:
        ctx.session.state["findings"] = findings


client = OpenAIResponsesClient()

research_agent = client.as_agent(
    name="ResearchAgent",
    instructions="Research the topic and store findings.",
    tools=[store_findings],
)

research_tool = research_agent.as_tool(
    name="research",
    description="Research a topic and store findings.",
    arg_name="query",
    propagate_session=True,
)

使用 propagate_session=True时，委托的代理会看到与调用方相同的 ctx.session 状态。 False将其保留为在其自己的会话中隔离子代理。

自定义聊天客户端和代理

如果实现自定义公共 run() 或 get_response() 方法，请将显式运行时存储桶添加到签名。

from collections.abc import Mapping, Sequence
from typing import Any

from agent_framework import ChatOptions, Message


async def get_response(
    self,
    messages: Sequence[Message],
    *,
    options: ChatOptions[Any] | None = None,
    function_invocation_kwargs: Mapping[str, Any] | None = None,
    client_kwargs: Mapping[str, Any] | None = None,
    **kwargs: Any,
):
    ...

用于 function_invocation_kwargs 工具调用流和 client_kwargs 特定于客户端的行为。 直接通过公共 **kwargs 传递特定于客户端的值只是一个兼容性路径，应被视为已弃用。 同样，使用仅迁移兼容性定义新工具 **kwargs - 请改为通过注入的上下文对象使用运行时数据。

后续步骤

中间件概述

  • Last updated on