常用接入术语

接入架构

接入优势

• Se**ntic Kernel 用 service_id 区分不同模型服务，测试、生产或不同模型可以独立配置。
• ChatHistory 明确保存系统、用户和助手消息，适合构建可检查的多轮对话。
• 自定义 AsyncOpenAI 客户端可以显式设置兼容端点，也可以统一设置超时和有限重试。
• Kernel 还能继续接入提示函数和插件；基础连接验证通过后再增加自动函数调用更容易排错。
• KingFlow 控制台集中管理模型、密钥和用量，便于隔离不同环境。

开始前的准备

• 准备 Python 3.10 或更高版本，并安装 uv。
• 在 KingFlow AI 控制台创建以 sk- 开头的 API Key，确认余额和模型分组可用。
• 复制目标模型的完整模型 ID，并确认上下文窗口满足待处理文档长度。
• 准备少量脱敏测试文本，首次联调不要直接加载生产知识库。
• 确认运行环境可以访问 api.**nju**shen.com。

创建 Se**ntic Kernel 测试项目

uv init se**ntic-kernel-kingflow-demo
cd se**ntic-kernel-kingflow-demo
uv add se**ntic-kernel openai python-dotenv

验证 KingFlow 基础接口

curl https://api.**nju**shen.com/v1/chat/completions \
  -H "Authorization: *earer sk-你的密钥" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "<KingFlow控制台中的完整模型ID>",
    "messages": [{"role": "user", "content": "只回复 SK_READY"}],
    "stream": false
  }'

配置环境变量

KINGFLOW_*ASE_**L=https://api.**nju**shen.com/v1
KINGFLOW_API_KEY=sk-你的密钥
KINGFLOW_MODEL=<KingFlow控制台中的完整模型ID>

.env
.venv/
__pycache__/
logs/

编写多轮对话示例

import asyncio
import os

from dotenv import load_dotenv
from openai import AsyncOpenAI
from se**ntic_kernel import Kernel
from se**ntic_kernel.connectors.ai.open_ai import (
    OpenAIChatCompletion,
    OpenAIChatPromptExecutionSettings,
)
from se**ntic_kernel.contents import ChatHistory

load_dotenv()

async def **in():
    client = AsyncOpenAI(
        api_key=os.environ["KINGFLOW_API_KEY"],
        *ase_url=os.environ["KINGFLOW_*ASE_**L"],
        timeout=60.0,
        **x_retries=2,
    )
    service = OpenAIChatCompletion(
        ai_model_id=os.environ["KINGFLOW_MODEL"],
        service_id="kingflow-chat",
        async_client=client,
    )
    kernel = Kernel()
    kernel.add_service(service)
    settings = OpenAIChatPromptExecutionSettings(
        service_id="kingflow-chat",
        temperature=0.2,
        **x_tokens=800,
    )

    history = ChatHistory(system_message="回答简洁；不确定时明确说明。")
    history.add_user_message("只回复 SEMANTIC_KERNEL_CONNECTED")
    first = await service.get_chat_message_content(
        chat_history=history, settings=settings, kernel=kernel
    )
    print(first.content)
    history.add_assistant_message(first.content or "")

    history.add_user_message("上一条要求你回复什么？")
    second = await service.get_chat_message_content(
        chat_history=history, settings=settings, kernel=kernel
    )
    print(second.content)

asyncio.run(**in())

观察流式输出

async for chunk in service.get_streaming_chat_message_content(
    chat_history=history,
    settings=settings,
    kernel=kernel,
):
    print(chunk.content or "", end="", flush=True)

运行和验证

uv run python **in.py

1. 先执行 KingFlow 基础请求，确认返回 SK_READY。
2. 创建 .env 和 .gitignore，再将完整示例保存为 **in.py。
3. 运行程序，确认第一轮返回 SEMANTIC_KERNEL_CONNECTED。
4. 检查第二轮是否能根据 ChatHistory 复述上一条要求。
5. 在 KingFlow 控制台核对模型、密钥、调用次数和 Token 用量。

正式环境配置

• ChatHistory 会随轮次增长；生产环境应按 Token 预算裁剪、摘要或归档旧消息。
• 不同环境使用独立 service_id、密钥和额度，避免测试流量进入生产账单。
• 先验证聊天补全，再启用插件和自动函数调用；函数调用还需要模型支持与参数校验。
• 工具或插件执行前检查用户权限，高风险操作需要人工确认并记录审计事件。
• 为 429 和 503 设置有限退避重试，并同时限制并发、请求时长和 Token 预算。
• 记录查询 ID、文档版本、模型、状态码、耗时和用量，但不要记录完整密钥和敏感原文。

文档与密钥安全

• 只在 .env、密钥管理器或部署平台 Secret 中保存 API Key，并定期轮换。
• 把系统提示、用户输入和工具返回值视为不可信数据，进入日志前完成脱敏。
• 插件执行前验证参数、身份和权限，禁止模型直接拼接系统命令或数据库语句。
• 多轮会话设置过期时间和删除机制，不把敏感历史长期保存在进程内存。
• 为开发、测试和生产设置独立密钥、额度和告警，异常用量出现后立即停用旧密钥。

上线检查清单

• 基础接口已经返回 SK_READY。
• KINGFLOW_*ASE_**L 为 https://api.**nju**shen.com/v1。
• KINGFLOW_MODEL 使用控制台中的完整模型 ID。
• AsyncOpenAI 已设置 *ase_url、超时和有限重试。
• .env 已被 .gitignore 排除，仓库中没有明文密钥。
• Se**ntic Kernel 使用明确的 service_id 注册服务。
• ChatHistory 已通过两轮对话验证上下文连续性。
• KingFlow 控制台的模型、密钥和用量符合预期。

常见故障排查