企业官网建设流程全解析

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），可直接在模板库中创建并部署。部署完成后，需要从控制台收集以下三项信息：

其中"required inputs"尤为关键：不同 Crew 模板要求的输入集不同，本示例基于Enterprise Content Marketing Crew模板，该模板要求两个必填字符串输入：company（要研究的公司名称）和topic（要研究的主题）。在后续配置 Kernel 插件时，这些输入会以KernelParameterMetadata的形式声明，供 LLM 理解每个参数的含义。

环境变量配置：端点与令牌

运行示例前，需要为 Crew 配置端点与认证令牌。最简单的方式是设置环境变量，或在项目根目录创建.env文件，写入如下内容：

CREW_AI_ENDPOINT="{Your Crew's endpoint}" CREW_AI_TOKEN="{Your Crew's 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：

1. 直接调用：创建CrewAIEnterprise实例后，手动调用kickoff、wait_for_crew_completion等方法，适合在代码中按固定流程编排 Crew 执行。
2. 封装为 Kernel Plugin：通过create_kernel_plugin生成一个输入签名与 Crew 必填输入对齐的插件，注册到 Kernel 后，LLM 可以在对话中根据工具描述自主决定何时以何种参数调用该 Crew——这是把 Crew 作为 AI 工具暴露给大模型的推荐方式。

两种方式共享同一套 API 约定：Kickoff 接口返回kickoff_id（任务 ID），状态查询与结果获取都围绕该 ID 展开。

方式一：直接调用 CrewAIEnterprise

直接调用方式的核心流程是"Kickoff → 轮询状态 → 获取结果"，代码如下（节选自 crew_ai_plugin.py）：

async 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(f"CrewAI 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）支持以下参数：

该对象实现了异步上下文管理器协议（__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_url=None, # 任务级事件回调地址 step_webhook_url=None, # 步骤级事件回调地址 crew_webhook_url=None, # 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( "I'm 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, kernel=kernel) 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_invoke=True 时模型自动选择并调用函数 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 API

CrewAIEnterprise的 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 的必填输入定义，返回CrewAIRequiredInputs（inputs: 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实例的引用，因此若后续以不同实例注册插件，需自行保证会话生命周期与连接复用。

运行示例与调试要点

1. 完成 Crew 云端部署，收集 endpoint、token 与必填输入清单。
2. 设置CREW_AI_ENDPOINT、CREW_AI_TOKEN环境变量（或.env文件）。
3. 根据自己 Crew 的实际输入，修改 crew_ai_plugin.py（直接调用处）与crew_input_parameters（插件元数据处）中的输入定义——务必为每个输入提供清晰的description与正确的type，这直接决定 LLM 能否准确填参。
4. 运行示例（例如python crew_ai_plugin.py）。观察日志：先看到直接调用打印 Kickoff ID 与结果，随后是插件调用阶段 LLM 自动发起kickoff_and_wait的完整链路。

常见问题定位建议：

• 令牌无效或端点错误：检查CREW_AI_ENDPOINT/CREW_AI_TOKEN是否与云端一致，客户端对非 2xx 响应会直接抛出异常。
• 缺少必填输入：插件调用时若参数缺失，会抛出PluginInitializationError（"Missing 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