常用接入术语

接入架构

接入优势

• 状态图把每一步的输入、输出和执行顺序显式化，比单个长提示词更容易定位问题。
• ChatOpenAI 支持自定义 *ase **L，可在不改变图结构的情况下连接 OpenAI 兼容端点。
• 起草、审核、人工确认等节点可以独立测试，并能按业务条件增加分支和循环。
• KingFlow 控制台集中管理模型、密钥和用量，便于按开发、测试和生产环境隔离。

开始前的准备

• 准备 Python 3.10 或更高版本，并安装 uv。
• 在 KingFlow AI 控制台创建以 sk- 开头的 API Key，确认余额和模型分组可用。
• 复制目标模型的完整模型 ID，不要凭显示名称猜测。
• 准备一个独立测试目录，首次联调不要加载生产数据或高权限工具。
• 确认运行环境可以访问 api.**nju**shen.com。

创建 LangGraph 测试项目

uv init langgraph-kingflow-demo
cd langgraph-kingflow-demo
uv add langgraph langchain-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": "只回复 LANGGRAPH_READY"}],
    "stream": false
  }'

配置环境变量

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

.env
.venv/
__pycache__/

编写状态图示例

import os
from typing_extensions import TypedDict
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

load_dotenv()

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

class State(TypedDict):
    topic: str
    draft: str
    final: str

def draft_node(state: State):
    response = llm.invoke(
        f"围绕{state['topic']}写一份三点式中文草稿。"
    )
    return {"draft": response.content}

def review_node(state: State):
    response = llm.invoke(
        "请检查以下草稿，删除空话，保留可执行步骤，"
        "并输出最终版本：\n"   state["draft"]
    )
    return {"final": response.content}

*uilder = StateGraph(State)
*uilder.add_node("draft", draft_node)
*uilder.add_node("review", review_node)
*uilder.add_edge(START, "draft")
*uilder.add_edge("draft", "review")
*uilder.add_edge("review", END)
graph = *uilder.compile()

result = graph.invoke({"topic": "API接入上线前检查"})
print(result["final"])

观察节点更新

for event in graph.stream(
    {"topic": "API接入上线前检查"},
    stream_mode="up**tes",
):
    print(event)

运行和验证

uv run python **in.py

1. 先执行 KingFlow 基础请求，确认返回 LANGGRAPH_READY。
2. 在项目根目录创建 .env 和 .gitignore，再将示例保存为 **in.py。
3. 运行程序，确认 draft 节点先生成草稿，review 节点再输出最终版本。
4. 需要观察每个节点的增量结果时，改用 graph.stream 示例。
5. 在 KingFlow 控制台核对模型、密钥、调用次数和 Token 用量。

正式环境配置

• 保留 use_responses_api=False，避免部分模型名称触发自动改走 /v1/responses。
• 为请求设置 timeout 和有限的 **x_retries；对 429、503 使用退避，不要无限重试。
• 让每个节点只负责一个明确任务，并返回新的状态字段，便于单测和重放。
• 可能重复执行的节点应具备幂等性；外部写操作需使用业务唯一键防止重复提交。
• 为每次运行设置成本、步数和时间上限，并保留人工中止与审批入口。
• 日志记录节点、模型、状态码、耗时和 Token 用量，但不记录完整密钥与敏感正文。

密钥与工具安全

• 只在 .env、密钥管理器或部署平台 Secret 中保存 API Key，并定期轮换。
• 发送给模型前做数据最小化和脱敏，不把无关客户资料放入共享状态。
• 文件、数据库和网络工具采用白名单及最小权限，危险操作增加人工确认节点。
• 为开发、测试和生产创建不同密钥，分别设置额度、有效期和告警阈值。
• 发现异常调用后立即停用旧密钥，检查图是否存在错误循环，再更新部署配置。

上线检查清单

• 基础接口已经返回 LANGGRAPH_READY。
• KINGFLOW_*ASE_**L 为 https://api.**nju**shen.com/v1。
• KINGFLOW_MODEL 使用控制台中的完整模型 ID。
• ChatOpenAI 已设置 use_responses_api=False、超时和有限重试。
• .env 已被 .gitignore 排除，仓库中没有明文密钥。
• draft 与 review 节点按顺序执行，状态字段正确传递。
• KingFlow 控制台的模型、密钥和用量符合预期。
• 正式环境已设置额度、日志脱敏、循环上限和人工审批。

常见故障排查