使用 Semantic Kernel 插件编排 AI 代理的技术深度解析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

class OrchestratorAgent:
    async def invoke(self, message: ChatMessage, context: Context) -> ChatResponseMessage:
        # 设置系统提示，引导 LLM 进行编排
        self._history.add_system_message(self.system_prompt)
        self._history.add_user_message(message.user_query)

        # 初始化代理插件
        org_policy_plugin = OrgPolicyAgentInvokingPlugin(kernel=self._kernel, message=message)
        web_search_plugin = WebSearchAgentInvokingPlugin(kernel=self._kernel, message=message)
        private_data_plugin = PrivateDataAgentInvokingPlugin(kernel=self._kernel, message=message)
        generic_plugin = GenericAgentInvokingPlugin(kernel=self._kernel, message=message)

        # 注册插件到 Kernel
        self._kernel.add_plugin(org_policy_plugin, "ORG_POLICY_AGENT")
        self._kernel.add_plugin(web_search_plugin, "WEB_SEARCH_AGENT")
        self._kernel.add_plugin(private_data_plugin, "PRIVATE_DATA_AGENT")
        self._kernel.add_plugin(generic_plugin, "GENERIC_LLM_AGENT")

        # 调用 Kernel 对问题进行编排
        results = []
        async for content in self._sk_agent.invoke(self._history):
            results.append(content)

        # 整合结果并返回响应
        final_content = results[-1].content if results else "无法生成响应。"
        return ChatResponseMessage(content=final_content)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

class WebSearchAgentInvokingPlugin(AgentInvokingPlugin):
    @kernel_function(
        name="invoke_web_search_agent",
        description="""WebSearch插件。用此插件回答需要当前公共信息的问题，例如：
        - 一般知识问题或事实（如"法国的首都是什么？"）
        - 时事、市场数据或行业趋势（如"MSFT最新股价？"）
        - 可能在网上找到的技术信息（如"如何配置Flask服务器？"）
        - 发表在学术论文或公共行业报告中的信息。""",
    )
    async def invoke_web_search_agent(self, query: Annotated[str, "要在网上搜索的特定问题或主题。"]):
        """调用外部Web搜索代理。"""
        # self.was_invoked = True # 标记为已调用
        # 创建搜索请求
        message_body = self.create_message_body(query)
        recipient = AgentId(WEBSEARCH_AGENT, str(self.message.metadata.session_id))
        # 目标代理
        
        # 通过运行时发送消息并等待响应
        result = await self.runtime.send_message(message_body, recipient)
        # 处理结果，更新内部状态（token使用情况、搜索结果），并返回
        processed_response = self.proccess_and_create_response(result)
        # self.token_usage = ... # 根据结果更新token使用情况
        # self.search_result = ... # 根据结果更新搜索结果
        return processed_response
        # 返回处理后的数据，通常只是一个摘要字符串给编排器LLM

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# 编排器 LLM 的系统提示示例
system_prompt = """您是一个编排专门代理的 AI 助手。分析用户查询。
基于查询和可用工具（插件）的描述，选择最合适的工具。

可用工具：
- ORG_POLICY_AGENT：用于关于内部公司政策和流程的问题。
- WEB_SEARCH_AGENT：用于当前事件、公共事实和一般网络知识。
- PRIVATE_DATA_AGENT：用于需要访问内部数据库的查询（例如销售数据、用户数据）。
- GENERIC_LLM_AGENT：作为一般对话或没有其他合适工具时的备选方案。

调用所选工具。如果需要多个工具，规划其执行。
将它们的输出合成为最终的综合响应。
"""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

from semantic_kernel.kernel_pugin import KernelPlugin
# ... 其他导入 ...

class AgentInvokingPlugin(KernelPlugin):
    def __init__(self, kernel, message, runtime):
        self.kernel = kernel
        self.message = message
        self.runtime = runtime  # 代理通信运行时
        self.token_usage = TokenUsage(prompt_token=0, completion_token=0)
        self.search_result = SearchResult(docs=[], citations=[])
        self.was_invoked = False
        
    # 可能添加用于处理结果等的抽象方法
    
    def create_message_body(self, query: str) -> dict:
        # 标准化消息创建
        pass
    
    def proccess_and_create_response(self, agent_result: dict) -> str:
        # 标准化响应处理，更新状态（令牌、结果）
        # 返回适合编排器 LLM 的字符串摘要
        pass
        
    def get_summary(self) -> str:
        # 如果已调用，返回结果摘要
        pass
    
    # ... 其他公共方法 ...