Semantic Kernel Python 集成 Crew AI Enterprise：把云端 Crew 封装为可调用的 Kernel Plugin

Semantic Kernel Python 集成 Crew AI Enterprise把云端 Crew 封装为可调用的 Kernel Plugin【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel本文以 Semantic Kernel Python SDK 的 Crew AI 示例 为骨架完整讲解如何将部署在 Crew AI Enterprise 云端的 Crew 接入 Semantic Kernel既可以直接调用CrewAIEnterprise对象的方法完成 Kickoff 与结果轮询也可以将其封装为标准的 Kernel Plugin让 LLM 通过函数调用Function Calling自动把 Crew 当作工具来使用。读完本文你将掌握环境配置、输入参数元数据定义、两种调用方式以及底层 HTTP 交互原理并能在自己的项目中复用这套集成方案。前置条件在 Crew AI Enterprise 云端部署 Crew本示例的核心是接入 Crew AI Enterprise 云服务因此在运行代码之前必须先在 Crew AI Enterprise 云上完成 Crew 的部署。Crew AI 官方提供大量预置的 Crew 模板如Enterprise Content Marketing Crew可直接在模板库中创建并部署。部署完成后需要从控制台收集以下三项信息信息说明endpointCrew 的基础 URLbase URL用于拼接待调用的 API 路径authentication token访问 Crew 所需的身份认证令牌required inputs启动kickoff该 Crew 时必须提供的输入参数需要知道每个输入的名称、类型与语义含义其中required inputs尤为关键不同 Crew 模板要求的输入集不同本示例基于Enterprise Content Marketing Crew模板该模板要求两个必填字符串输入company要研究的公司名称和topic要研究的主题。在后续配置 Kernel 插件时这些输入会以KernelParameterMetadata的形式声明供 LLM 理解每个参数的含义。环境变量配置端点与令牌运行示例前需要为 Crew 配置端点与认证令牌。最简单的方式是设置环境变量或在项目根目录创建.env文件写入如下内容CREW_AI_ENDPOINT{Your Crews endpoint} CREW_AI_TOKEN{Your Crews authentication token}从源码层面看这两个变量由 crew_ai_settings.py 中的CrewAISettings定义endpoint为必填字符串auth_token使用SecretStr类型存储避免日志泄露同时可选配置polling_interval默认1.0秒与polling_timeout默认30.0秒。其env_prefix为CREW_AI_因此环境变量按CREW_AI_ENDPOINT、CREW_AI_TOKEN命名即被自动识别。除环境变量外也可在构造CrewAIEnterprise时直接传入endpoint、auth_token参数详见下文直接调用一节。两种使用方式总览官方 示例脚本 演示了CrewAIEnterprise的两种用法二者都基于同一个底层 Client直接调用创建CrewAIEnterprise实例后手动调用kickoff、wait_for_crew_completion等方法适合在代码中按固定流程编排 Crew 执行。封装为 Kernel Plugin通过create_kernel_plugin生成一个输入签名与 Crew 必填输入对齐的插件注册到 Kernel 后LLM 可以在对话中根据工具描述自主决定何时以何种参数调用该 Crew——这是把 Crew 作为 AI 工具暴露给大模型的推荐方式。两种方式共享同一套 API 约定Kickoff 接口返回kickoff_id任务 ID状态查询与结果获取都围绕该 ID 展开。方式一直接调用 CrewAIEnterprise直接调用方式的核心流程是Kickoff → 轮询状态 → 获取结果代码如下节选自 crew_ai_plugin.pyasync def using_crew_ai_enterprise(): # 创建 CrewAI Enterprise Crew 实例异步上下文管理器负责会话生命周期 async with CrewAIEnterprise() as crew: # 该示例基于 Enterprise Content Marketing Crew 模板需要以下输入 inputs {company: CrewAI, topic: Agentic products for consumers} # 直接以输入参数启动 Crew返回 Kickoff ID kickoff_id await crew.kickoff(inputs) print(fCrewAI Enterprise Crew kicked off with ID: {kickoff_id}) # 等待 Crew 执行完成返回最终结果 result await crew.wait_for_crew_completion(kickoff_id) print(CrewAI Enterprise Crew completed with the following result:) print(result)构造参数说明CrewAIEnterprise的构造函数见 crew_ai_enterprise.py支持以下参数参数类型默认值说明endpointstr \| None环境变量CREW_AI_ENDPOINTCrew 的 API 端点auth_tokenstr \| None环境变量CREW_AI_TOKEN认证令牌polling_intervalfloat \| None1.0轮询状态的时间间隔秒polling_timeoutfloat \| None30.0等待完成的最长超时时间秒sessionaiohttp.ClientSession \| None自动创建可复用的 HTTP 会话env_file_pathstr \| NoneNone环境设置文件的路径作为环境变量的后备env_file_encodingstr \| NoneNone环境设置文件的编码该对象实现了异步上下文管理器协议__aenter__/__aexit__因此推荐使用async with确保底层aiohttp会话被正确创建与关闭。Kickoff 的扩展能力kickoff方法见 crew_ai_enterprise.py除inputs外还支持三个可选的 Webhook 参数用于在 Crew 运行过程中接收异步事件通知await crew.kickoff( inputs{company: CrewAI, topic: Agentic products for consumers}, task_webhook_urlNone, # 任务级事件回调地址 step_webhook_urlNone, # 步骤级事件回调地址 crew_webhook_urlNone, # Crew 级事件回调地址 )状态模型与完成判定Crew 的执行状态由 crew_ai_models.py 中的枚举CrewAIEnterpriseKickoffState定义PENDING、STARTED、RUNNING、SUCCESS、FAILED、FAILURE、NOT FOUND。wait_for_crew_completion内部以polling_interval为周期循环调用状态接口直到状态进入终态SUCCESS、FAILED、FAILURE、NOT FOUND或超过polling_timeout超时若终态为失败则抛出FunctionResultError其错误信息携带 Crew 返回的失败原因成功时返回状态响应中的result字段作为最终输出。方式二封装为 Kernel Plugin 供 LLM 调用把 Crew 变成 Kernel Plugin 的价值在于让 LLM 根据自然语言对话自主判断何时调用 Crew。示例中create_kernel_plugin的调用方式如下见 crew_ai_plugin.py# Crew 的语义描述将作为插件plugin级别的 description供 LLM 判断何时使用 crew_description ( Conducts thorough research on the specified company and topic to identify emerging trends, analyze competitor strategies, and gather># 配置 Kernel 与聊天补全服务 kernel, chat_completion, settings configure_kernel_for_chat() kernel.add_plugin(crew_plugin) # 维护会话历史系统消息 用户请求 history ChatHistory() history.add_system_message(You are an AI assistant that can help me with research.) history.add_user_message( Im looking for emerging marketplace trends about Crew AI and their consumer AI products. ) # 发起对话LLM 在需要时自动调用 EnterpriseContentMarketingCrew 插件 response await chat_completion.get_chat_message_content(history, settings, kernelkernel) print(response)configure_kernel_for_chat见 crew_ai_plugin.py展示了可复用的 Kernel 装配模式def configure_kernel_for_chat() - tuple[Kernel, ChatCompletionClientBase, PromptExecutionSettings]: kernel Kernel() # 可选聊天补全服务OPENAI / AZURE_OPENAI / AZURE_AI_INFERENCE / ANTHROPIC / # BEDROCK / GOOGLE_AI / MISTRAL_AI / OLLAMA / ONNX / VERTEX_AI / DEEPSEEK 等 # 需确保所选服务已正确配置环境变量 chat_completion_service, request_settings get_chat_completion_service_and_request_settings(Services.OPENAI) # 开启自动函数调用auto_invokeTrue 时模型自动选择并调用函数 request_settings.function_choice_behavior FunctionChoiceBehavior.Auto() kernel.add_service(chat_completion_service) return kernel, chat_completion_service, request_settings预期运行输出示例脚本的注释crew_ai_plugin.py给出了预期的日志与对话输出。可以看到当模型判断需要研究信息时会自动发起对插件kickoff_and_wait的调用参数由 LLM 从用户消息中提取INFO:semantic_kernel.kernel:Calling EnterpriseContentMarketingCrew-kickoff_and_wait function with args: {company:Crew AI,topic:emergging marketplace trends in consumer AI products} INFO:semantic_kernel.core_plugins.crew_ai.crew_ai_enterprise:CrewAI Crew kicked off with Id: ***** INFO:semantic_kernel.core_plugins.crew_ai.crew_ai_enterprise:CrewAI Crew with kickoff Id: ***** completed with status: SUCCESS INFO:semantic_kernel.functions.kernel_function:Function EnterpriseContentMarketingCrew-kickoff_and_wait succeeded. Here are some emerging marketplace trends related to Crew AI and their consumer AI products, along with suggested content pieces to explore these trends: ...整条链路为LLM 接收用户请求 → 通过函数调用机制选中kickoff_and_wait并填充company/topic参数 → Kernel 执行插件函数 → 底层触发 Crew Kickoff → 轮询至SUCCESS状态 → 将 Crew 的研究结果作为函数返回值交还给 LLM → LLM 依据结果组织最终回答。底层原理HTTP 客户端如何对接 Crew AI Enterprise APICrewAIEnterprise的 HTTP 交互由 crew_ai_enterprise_client.py 中的CrewAIEnterpriseClient完成其端点约定与请求头如下认证头每个请求都携带Authorization: Bearer {auth_token}、Content-Type: application/json并附加user_agent请求头来自SEMANTIC_KERNEL_USER_AGENT。GET{endpoint}/inputs查询 Crew 的必填输入定义返回CrewAIRequiredInputsinputs: dict[str, str]。这为提前知道 Crew 需要哪些参数提供了接口级支持。POST{endpoint}/kickoff请求体为 JSON包含inputs、taskWebhookUrl、stepWebhookUrl、crewWebhookUrl四个字段响应解析为CrewAIKickoffResponse核心字段是kickoff_id。GET{endpoint}/status/{task_id}按 Kickoff ID 查询状态响应解析为CrewAIStatusResponse包含state状态枚举、result最终结果可为空与last_step最近一步的详情可为空。三个响应模型CrewAIKickoffResponse、CrewAIStatusResponse、CrewAIRequiredInputs均基于 pydantic 定义位于 crew_ai_models.py因此所有接口返回都会经过结构化校验异常响应会由raise_for_status()直接抛出。值得留意的一个同步细节从源码结构看示例目前的直接调用与插件调用都是在同一async with CrewAIEnterprise()上下文内完成的两次 Kickoff 串行执行而create_kernel_plugin生成的插件函数捕获了同一个crew实例的引用因此若后续以不同实例注册插件需自行保证会话生命周期与连接复用。运行示例与调试要点完成 Crew 云端部署收集 endpoint、token 与必填输入清单。设置CREW_AI_ENDPOINT、CREW_AI_TOKEN环境变量或.env文件。根据自己 Crew 的实际输入修改 crew_ai_plugin.py直接调用处与crew_input_parameters插件元数据处中的输入定义——务必为每个输入提供清晰的description与正确的type这直接决定 LLM 能否准确填参。运行示例例如python crew_ai_plugin.py。观察日志先看到直接调用打印 Kickoff ID 与结果随后是插件调用阶段 LLM 自动发起kickoff_and_wait的完整链路。常见问题定位建议令牌无效或端点错误检查CREW_AI_ENDPOINT/CREW_AI_TOKEN是否与云端一致客户端对非 2xx 响应会直接抛出异常。缺少必填输入插件调用时若参数缺失会抛出PluginInitializationErrorMissing required input需核对parameters元数据是否完整覆盖 Crew 的全部必填输入。长时间无结果wait_for_crew_completion受polling_timeout默认 30 秒限制超时会中止等待对于执行时间较长的 Crew可通过构造参数调大该值或改用kickoff 独立wait_for_completion的异步编排模式。状态为失败wait_for_crew_completion在FAILED/FAILURE终态会抛出FunctionResultError错误信息中会包含 Crew 返回的失败详情。总结本示例为把外部 Agent 编排平台接入 Semantic Kernel提供了一条清晰路径通过CrewAIEnterprise直接调用满足程序化编排场景通过create_kernel_plugin生成的插件满足LLM 自主调用场景底层由CrewAIEnterpriseClient统一封装鉴权、Kickoff、状态查询等 HTTP 交互。整套实现的关键在于把 Crew 的必填输入用KernelParameterMetadata精确描述出来——类型与语义描述越准确LLM 的工具调用越可靠。如需继续深入可对照阅读 示例脚本 与 核心插件源码并参考 Python SDK 的 核心插件目录 了解其他内置插件的实现风格。【免费下载链接】semantic-kernelIntegrate cutting-edge LLM technology quickly and easily into your apps项目地址: https://gitcode.com/GitHub_Trending/se/semantic-kernel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考