FastAPI中如何将请求上下文(session_id)传递给Semantic Kernel的OpenAI插件

方式一：通过KernelArguments传递上下文参数 #

这是Semantic Kernel中传递请求级上下文的标准方式，无需修改插件注册逻辑，只需在调用时传入参数，并在插件函数中读取。

1. 修改MovesPlugin的工具函数，支持读取session_id #

更新插件中的业务函数，通过KernelArguments获取session_id，同时保留初始化时传入的openai_client：

from semantic_kernel.kernel_function_decorator import kernel_function
from semantic_kernel.arguments import KernelArguments

class MovesPlugin:
    def __init__(self, openai_client):
        # 复用全局的openai_client，避免重复创建实例
        self.openai_client = openai_client

    @kernel_function(description="获取指定会话的用户动作数据")
    async def get_user_moves(self, context: KernelArguments) -> str:
        # 从上下文取出session_id
        session_id = context.get("session_id")
        if not session_id:
            raise ValueError("请求上下文缺少session_id")
        
        # 结合openai_client和session_id执行业务逻辑
        # 示例：调用OpenAI API或查询会话关联数据
        response = await self.openai_client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": f"获取会话{session_id}的用户动作"}]
        )
        return response.choices[0].message.content

2. 在FastAPI端点中传入session_id到KernelArguments #

在/chat端点中，从请求提取session_id后，将其封装到KernelArguments中，再传入Kernel的执行方法：

@app.post("/chat/")
async def chat(request: Request):
    req_data = await request.json()
    session_id = req_data.get('session_id')
    chat_history = req_data.get('chat_history', [])

    # 构造包含session_id的上下文参数
    kernel_args = KernelArguments(session_id=session_id)
    
    # 获取聊天完成服务并执行请求
    _chat_completion_service = kernel.get_service(type=ChatCompletionClientBase)
    response = await _chat_completion_service.get_chat_message_content(
        chat_history=chat_history,
        kernel=kernel,
        settings=execution_settings,
        arguments=kernel_args  # 传入上下文参数
    )
    
    return {"response": response.content}

方式二：动态创建插件实例（适用于需会话级插件状态的场景） #

如果MovesPlugin需要为每个会话维护独立状态，可以在请求时动态创建插件实例并注册到Kernel（注意：需确保Kernel是请求级实例，而非全局单例）。

1. 修改FastAPI的Kernel创建逻辑 #

将全局Kernel改为请求级创建，避免跨请求状态污染：

from fastapi import Depends

def get_kernel(api_key: str, model_id: str, service_id: str):
    """创建请求级的Kernel实例"""
    kernel = Kernel()
    openai_client = OpenAI(api_key=api_key)
    chat_completion_service = OpenAIChatCompletion(
        ai_model_id=model_id, 
        api_key=api_key, 
        service_id=service_id 
    )
    kernel.add_service(chat_completion_service)
    return kernel, openai_client

2. 在端点中动态注册带session_id的插件 #

@app.post("/chat/")
async def chat(
    request: Request,
    kernel_deps: tuple[Kernel, OpenAI] = Depends(get_kernel)
):
    kernel, openai_client = kernel_deps
    req_data = await request.json()
    session_id = req_data.get('session_id')
    chat_history = req_data.get('chat_history', [])

    # 动态创建带session_id和openai_client的插件实例
    moves_plugin = MovesPlugin(openai_client, session_id)
    kernel.add_plugin(moves_plugin, plugin_name='MovesPlugin')
    
    # 执行聊天请求
    _chat_completion_service = kernel.get_service(type=ChatCompletionClientBase)
    response = await _chat_completion_service.get_chat_message_content(
        chat_history=chat_history,
        kernel=kernel,
        settings=execution_settings
    )
    
    return {"response": response.content}

3. 更新MovesPlugin的初始化逻辑 #

class MovesPlugin:
    def __init__(self, openai_client, session_id):
        self.openai_client = openai_client
        self.session_id = session_id  # 会话级状态

    @kernel_function(description="获取当前会话的用户动作数据")
    async def get_user_moves(self) -> str:
        # 直接使用实例中的session_id和openai_client
        response = await self.openai_client.chat.completions.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": f"获取会话{self.session_id}的用户动作"}]
        )
        return response.choices[0].message.content

关键注意事项 #

• 复用OpenAI Client：无论哪种方式，都应避免在每个请求中重复创建openai_client，减少资源开销。
• 上下文参数校验：在插件函数中务必校验session_id的存在性，避免空指针异常。
• Kernel实例作用域：如果使用方式二，确保Kernel是请求级实例，不要用全局单例，防止跨请求的session_id污染。

内容的提问来源于stack exchange，提问作者Prasanth Rao