常用接入术语

接入架构

接入优势

• ChatPromptTemplate、ChatOpenAI 和输出解析器都遵循 Runna*le 接口，可以用竖线运算符组合。
• 提示、模型和返回结构相互分离，业务代码更容易测试、替换和复用。
• ChatOpenAI 支持显式设置 *ase_url、密钥、超时、重试次数和生成参数。
• 模型支持工具调用或结构化输出时，with_structured_output 可直接返回经过 Py**ntic 校验的对象。
• KingFlow 控制台集中管理模型、密钥和用量，便于隔离不同环境。

开始前的准备

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

创建 LangChain 测试项目

uv init langchain-kingflow-demo
cd langchain-kingflow-demo
uv add langchain-openai langchain-core py**ntic python-dotenv

验证 KingFlow 基础接口

先验证端点、密钥和模型；基础请求失败时不要继续排查 LangChain 代码。

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": "只回复 LANGCHAIN_READY"}],
    "stream": false
  }'

配置环境变量

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

.env
.venv/
__pycache__/
artifacts/

编写结构化输出示例

目标模型必须支持工具调用或结构化输出；否则应先使用普通模型链并自行解析、校验结果。

import os

from dotenv import load_dotenv
from py**ntic import *aseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

load_dotenv()

model = ChatOpenAI(
    model=os.environ["KINGFLOW_MODEL"],
    api_key=os.environ["KINGFLOW_API_KEY"],
    *ase_url=os.environ["KINGFLOW_*ASE_**L"],
    temperature=0.2,
    **x_tokens=800,
    timeout=60,
    **x_retries=2,
    use_responses_api=False,
    stream_usage=False,
)

ready = model.invoke("只回复 LANGCHAIN_CONNECTED")
print(ready.content)

class ReleaseReview(*aseModel):
    checks: list[str] = Field(description="可执行的上线检查项")
    risks: list[str] = Field(description="仍需处理的风险")
    verdict: str = Field(description="只能是 pass 或 review")

prompt = ChatPromptTemplate.from_messages([
    ("system", "你负责上线**；信息不足时列入风险。"),
    ("hu**n", "系统信息：{system_info}"),
])

structured = model.with_structured_output(ReleaseReview)
chain = prompt | structured
result = chain.invoke({
    "system_info": "密钥放入 Secret；已设置额度；尚未配置 429 重试。"
})
print(result)

增加流式输出

兼容**不支持流式 usage 数据时，保留 stream_usage=False。

for chunk in model.stream("用三点说明上线验收方法"):
    print(chunk.content or "", end="", flush=True)

运行和验证

uv run python **in.py

1. 先执行 KingFlow 基础请求，确认返回 LANGCHAIN_READY。
2. 创建 .env 和 .gitignore，再将完整示例保存为 **in.py。
3. 运行程序，确认直接调用返回 LANGCHAIN_CONNECTED。
4. 检查结构化链是否返回 checks、risks 和 verdict 三个字段。
5. 在 KingFlow 控制台核对模型、密钥、调用次数和 Token 用量。

正式环境配置

• 显式设置 use_responses_api=False，固定使用 OpenAI Chat Completions 兼容路径。
• 结构化输出依赖目标模型的工具调用或结构化能力；不支持时改用普通文本链并在应用层解析。
• 类型化输出仍需业务校验；列表可能为空，verdict 也要在应用层限制为允许值。
• **不支持流式 usage 数据时设置 stream_usage=False，并单独核对平台用量。
• 为 429 和 503 设置有限退避重试，并同时限制并发、请求时长和 Token 预算。
• 记录查询 ID、文档版本、模型、状态码、耗时和用量，但不要记录完整密钥和敏感原文。

文档与密钥安全

• 只在 .env、密钥管理器或部署平台 Secret 中保存 API Key，并定期轮换。
• 把输入样本和模型输出视为不可信数据，进入日志或评估集前完成脱敏。
• 提示模板、测试数据和生产输入分开管理，样例进入文档前先完成脱敏。
• 保存追踪记录或演示结果时检查敏感字段，禁止把真实密钥写入 artifact。
• 为开发、测试和生产设置独立密钥、额度和告警，异常用量出现后立即停用旧密钥。

上线检查清单

• 基础接口已经返回 LANGCHAIN_READY。
• KINGFLOW_*ASE_**L 为 https://api.**nju**shen.com/v1。
• KINGFLOW_MODEL 使用控制台中的完整模型 ID。
• ChatOpenAI 已设置 *ase_url、模型 ID、超时和有限重试。
• .env 已被 .gitignore 排除，仓库中没有明文密钥。
• 提示模板变量和 Py**ntic 字段描述已经明确。
• 结构化链已返回 checks、risks 和 verdict。
• 已确认目标模型支持结构化输出，或准备了普通文本回退方案。
• KingFlow 控制台的模型、密钥和用量符合预期。

常见故障排查