1. 为什么要在 Harness 流水线里做多轮对话状态跟踪

多轮对话状态跟踪（Dialogue State Tracking，DST）说白了就是让对话机器人记住「上一句聊的是哪条流水线、哪个环境、哪次执行」，这样用户第二轮只问「怎么修复」时，系统还能接得上。它适合谁？适合正在把 AI 助手塞进 DevOps 流程的团队，尤其是已经在用 Harness 跑 CI/CD、想让排障对话不再反复追问上下文的工程师。

我在实际项目里遇到过很典型的场景：流水线挂了，运维同学在 Harness 的 AI 助手里问「今天支付服务那条流水线为什么失败」，助手答「单元测试报 NullPointerException」。接着他问「怎么修」，助手回「请问你指的是哪条流水线」。这一下上下文全丢了，用户得把项目、流水线、执行记录再报一遍。问题不在模型笨，而在于中间缺了一层状态管理：每一轮请求都是独立的，历史信息没有被结构化地保存和传递。

Harness 本身是软件交付平台，它的业务对象很多：Account、Org、Project、Pipeline、Execution、Environment、Service、Deployment。用户一句话里可能只说了「支付服务」，但系统需要把它映射到具体的 Project 和 Pipeline，还要确认这个用户有没有权限看这条流水线。如果只靠大模型自由发挥，很容易编出一个不存在的流水线 ID，后续回复就会把人带偏。所以 DST 在这里不是锦上添花，而是让 AI 助手从「能聊天」变成「能干活」的关键一层。

这一篇我不讲空泛的架构图，直接交付能跑的东西：用 TaoToken 统一 Key 和 API 通道接入对话模型，在 Harness Pipeline 里跑通多轮状态槽位更新与断言。你会看到可复制的 pipeline YAML、环境变量配置、DST 用例配置，以及本地和流水线两段验证动作。目标很明确：让「第一轮问失败原因、第二轮问修复方案」这种多轮对话，在流水线里可复现、可断言、可回归。

核心检索词先摆出来：多轮对话状态跟踪、Harness、DST、DevOps、对话状态跟踪。下面所有配置都围绕这几个词展开，不堆砌，只讲能落地的部分。

2. TaoToken 统一 Key 接入：环境变量与模型通道配置

在 Harness 里做 DST 验证，第一件麻烦事是模型通道。团队里可能有人用这个模型、有人用那个模型，Key 散落在各个地方，流水线一跑就报 401。TaoToken 在这里的作用是提供一个统一的 API 通道和 Key 管理入口，让本地调试和 Harness 流水线用同一套 Base URL 和 Key，减少「本地能跑、流水线挂掉」的扯皮。

先明确三个东西，后面所有配置都围绕它们：

• Base URL：https://taotoken.net/api
• API Key：在控制台创建，形如sk-xxxx，不要写进代码仓库
• Model ID：对话模型填你实际调用的模型标识，比如gpt-4o-mini或claude-3-5-sonnet，以控制台模型列表为准

如果你还没有 Key，可以到控制台的 API Keys 页面创建：https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。创建后先别急着塞进 Harness，本地用 curl 验一下通道是否通。

本地验证命令如下，注意把$TAOTOKEN_API_KEY换成你自己的 Key：

export TAOTOKEN_API_KEY="sk-你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api" curl -s "$TAOTOKEN_BASE_URL/v1/chat/completions" \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -H "Content-Type: application/json" \ -d '{ "model": "gpt-4o-mini", "messages": [ {"role": "user", "content": "只回复两个字：通了"} ], "temperature": 0 }'

如果返回的 JSON 里有choices[0].message.content，说明通道没问题。这一步很关键，因为后面 Harness 流水线里报的很多错，根因都是 Key 或 Base URL 配错，而不是 DST 逻辑本身。

接下来在 Harness 里配置环境变量。进入 Harness 项目设置，找到 Secrets 或 Environment Variables，建议把 Key 放 Secret，Base URL 和 Model ID 放普通变量。命名建议统一加前缀，避免和别的服务冲突：

这里有个容易踩的坑：Harness 的 Secret 在日志里会被掩码，但如果你在 shell 里用echo打印，可能触发掩码导致后续字符串拼接出错。所以脚本里不要打印 Key，直接用变量引用。

关于模型选择，DST 的实体抽取和状态更新对模型要求不一样。抽取候选实体这种任务，用便宜快的小模型就够；状态更新需要做「保留哪些旧实体、丢弃哪些」的决策，可以用稍强的模型。你可以在 TaoToken 的模型对话页面先手动试几轮，确认模型对 JSON 输出的稳定性：https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。如果模型经常在 JSON 外面包一层解释文字，后面解析就会失败，这时候要么换模型，要么在 prompt 里强制「只返回 JSON」。

如果你打算长期在 Harness 里跑 Agent 类任务，比如让 AI 自动分析失败日志并给出修复建议，可以了解 Coding Plan 的额度方式：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。不过本篇的重点是 DST 验证链路，先把通道和状态跑通，再考虑规模化。

配置完成后，建议在 Harness 里加一个最简的连通性检查步骤，作为流水线的第一道门。这样后面 DST 用例失败时，你能快速判断是通道问题还是逻辑问题。

3. 可复制的 Harness Pipeline YAML 与 DST 用例配置

这一节是核心，直接给可复制的配置。整体思路：Harness Pipeline 里跑一个 Python 脚本，脚本负责调用 TaoToken 通道，执行两轮对话，维护一个 DST 状态对象，最后对状态里的槽位做断言。断言失败，流水线就红，这样多轮状态跟踪的结果就可回归。

先看目录结构，建议在仓库里这样放：

dst-harness/ ├── pipeline/ │ └── dst-verify.yaml ├── scripts/ │ └── dst_runner.py ├── config/ │ └── dst_cases.json └── requirements.txt

requirements.txt内容很简单：

openai>=1.0.0 pyyaml

注意这里用的是 OpenAI 兼容 SDK，通过base_url指向 TaoToken 通道，不需要额外装奇怪的包。

先写 DST 用例配置config/dst_cases.json。这个文件定义多轮对话的输入和期望状态，是断言的依据：

{ "cases": [ { "case_id": "dst_pipeline_repair", "description": "第一轮问失败原因，第二轮问修复方案，验证流水线槽位不丢失", "turns": [ { "user": "今天支付服务那条 Java 流水线为什么失败？", "expect_slots": { "service": "支付服务", "pipeline_type": "Java" } }, { "user": "怎么修复？", "expect_slots": { "service": "支付服务", "pipeline_type": "Java", "intent": "repair" } } ] } ] }

这里的关键是expect_slots：第一轮结束后，状态里应该有service和pipeline_type；第二轮用户没有重复这两个词，但状态里必须还在，同时新增intent=repair。这就是 DST 要验证的核心：槽位跨轮保留与更新。

接下来是scripts/dst_runner.py。它做四件事：读用例、调模型、更新状态、断言。为了让你能直接跑，我把状态更新写成「模型抽取 + 本地合并」的混合方式，既用模型理解自然语言，又用代码保证槽位不丢：

import json import os import sys from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ.get("TAOTOKEN_BASE_URL", "https://taotoken.net/api"), ) MODEL_ID = os.environ.get("DST_MODEL_ID", "gpt-4o-mini") def extract_slots(user_text, history_summary): prompt = f""" 你是 DevOps 对话状态抽取器。根据用户当前输入和历史摘要，抽取槽位。 只返回 JSON，格式：{{"slots": {{"key": "value"}}, "intent": "字符串"}} 历史摘要：{history_summary} 当前输入：{user_text} """ resp = client.chat.completions.create( model=MODEL_ID, messages=[{"role": "user", "content": prompt}], temperature=0, response_format={"type": "json_object"}, ) content = resp.choices[0].message.content return json.loads(content) def merge_state(old_state, new_slots, new_intent): merged = dict(old_state) for k, v in new_slots.items(): if v: merged[k] = v if new_intent: merged["intent"] = new_intent return merged def run_case(case): state = {} history_summary = "" for idx, turn in enumerate(case["turns"]): result = extract_slots(turn["user"], history_summary) state = merge_state(state, result.get("slots", {}), result.get("intent")) history_summary = f"用户说：{turn['user']}；当前槽位：{json.dumps(state, ensure_ascii=False)}" print(f"[turn {idx+1}] state={json.dumps(state, ensure_ascii=False)}") for key, expected in turn["expect_slots"].items(): actual = state.get(key) if actual != expected: print(f"ASSERT FAIL: case={case['case_id']} turn={idx+1} key={key} expected={expected} actual={actual}") return False print(f"ASSERT PASS: {case['case_id']}") return True def main(): with open("config/dst_cases.json", "r", encoding="utf-8") as f: cases = json.load(f)["cases"] all_pass = True for case in cases: if not run_case(case): all_pass = False sys.exit(0 if all_pass else 1) if __name__ == "__main__": main()

这段代码里，response_format={"type": "json_object"}是保证解析稳定的关键。如果模型不支持这个参数，就去掉它，但在 prompt 里必须强调「只返回 JSON，不要任何解释」。

然后是 Harness Pipeline YAML。这里用 Harness 的 CI 阶段，跑一个 Run 步骤。注意把 Secret 引用写对，Harness 里通常用<+secrets.getValue("...")>的表达式：

pipeline: name: dst-verify-pipeline identifier: dst_verify_pipeline projectIdentifier: your_project orgIdentifier: your_org tags: {} stages: - stage: name: dst-verify identifier: dst_verify type: CI spec: cloneCodebase: true execution: steps: - step: type: Run name: run-dst-cases identifier: run_dst_cases spec: connectorRef: your_connector image: python:3.11-slim shell: Bash command: | set -e pip install -r requirements.txt export TAOTOKEN_API_KEY=<+secrets.getValue("taotoken_api_key")> export TAOTOKEN_BASE_URL=<+pipeline.variables.taotoken_base_url> export DST_MODEL_ID=<+pipeline.variables.dst_model_id> python scripts/dst_runner.py infrastructure: type: KubernetesDirect spec: connectorRef: your_k8s_connector namespace: default variables: - name: taotoken_base_url type: String value: https://taotoken.net/api - name: dst_model_id type: String value: gpt-4o-mini

如果你用的是 Harness Cloud 或者别的执行环境，infrastructure那段按你实际的环境改。核心是command里的三步：装依赖、注入环境变量、跑脚本。脚本退出码非 0，Harness 步骤就失败，流水线就红。

这里要提醒一个细节：Harness 的变量表达式在 YAML 里是<+...>，如果你在 shell 里用单引号包住，可能不会被替换。所以export那几行不要加单引号，直接写表达式。

另外，如果你在 Harness 里用 Cline MCP 或类似的工具做 AI 辅助，记得把三件套配全：Base URL、Key、Model ID。缺一个都会报连接错误。Base URL 用https://taotoken.net/api，Key 用 Secret，Model ID 用你实际选的模型。这三者在本地和流水线里必须一致，否则就会出现「本地通、流水线 401」的经典问题。

4. 验证请求与成功结果：本地与流水线两段动作

配置写完了，得验证。我习惯分两段：先在本地把 DST 逻辑跑通，再推到 Harness 流水线里跑。这样出问题时，能快速定位是模型通道、脚本逻辑还是 Harness 环境的问题。

本地验证第一步，确认通道。前面第 2 节的 curl 已经做过，这里再做一次带 JSON 输出的调用，确认模型能稳定返回结构化内容：

export TAOTOKEN_API_KEY="sk-你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api" export DST_MODEL_ID="gpt-4o-mini" python -c " from openai import OpenAI import os, json client = OpenAI(api_key=os.environ['TAOTOKEN_API_KEY'], base_url=os.environ['TAOTOKEN_BASE_URL']) resp = client.chat.completions.create( model=os.environ['DST_MODEL_ID'], messages=[{'role':'user','content':'返回JSON：{\"slots\":{\"service\":\"支付服务\"},\"intent\":\"query\"}'}], temperature=0, response_format={'type':'json_object'} ) print(resp.choices[0].message.content) "

如果输出是合法的 JSON，说明通道和模型都 OK。如果报local proxy failed或连接超时，先检查 Base URL 是不是写成了https://taotoken.net/api/v1这种重复路径，正确写法是https://taotoken.net/api，SDK 会自动拼/v1/chat/completions。

本地验证第二步，跑 DST 用例：

cd dst-harness pip install -r requirements.txt python scripts/dst_runner.py

期望输出类似：

[turn 1] state={"service": "支付服务", "pipeline_type": "Java"} [turn 2] state={"service": "支付服务", "pipeline_type": "Java", "intent": "repair"} ASSERT PASS: dst_pipeline_repair

看到ASSERT PASS，说明多轮状态跟踪在本地是通的。注意第二轮 state 里service和pipeline_type还在，这就是 DST 的价值：用户没说，但状态记住了。

本地通过后，提交代码，触发 Harness 流水线。在 Harness 的 Execution 页面看run-dst-cases步骤的日志。成功时你会看到同样的ASSERT PASS，并且步骤状态是 Success。如果失败，日志里会打印ASSERT FAIL，告诉你哪个槽位对不上。

这里有个实用技巧：在 Harness 日志里搜state=，能快速看到每一轮的状态快照。如果第一轮就缺槽位，说明模型抽取有问题；如果第一轮有、第二轮丢了，说明状态合并逻辑有问题。把这两类问题分开，排查效率会高很多。

流水线跑通后，建议把dst_cases.json当成回归用例库。每次改 prompt 或换模型，都跑一遍。DST 这种东西，最怕的就是「这次好了，下次换个说法又丢了」。有了断言，回归就有依据。

如果你在验证过程中想手动试几轮对话，看看模型对槽位的理解，可以到模型对话页面直接聊：https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。手动试出来的好 prompt，再固化到脚本里。

5. 常见报错排查：401、local proxy failed、reading choices、OAuth

这一节按真实报错来。你在 Harness 里跑 DST，大概率会遇到下面几类问题。我把现象、原因、处理方式列清楚，方便你对照。

401 Unauthorized。现象是模型调用返回 401，日志里可能只写Authentication failed。原因通常是 Key 没注入、Key 过期、或者 Harness Secret 引用写错。处理：先在本地用同一个 Key 跑 curl，确认 Key 本身有效；再检查 Harness 里TAOTOKEN_API_KEY的 Secret 名称和表达式是否一致。注意 Harness 的 Secret 在不同 scope（Project/Org/Account）下名称可能冲突，引用时最好带上完整路径。

local proxy failed。这个报错通常出现在网络层，意思是请求没到达目标地址。原因可能是 Base URL 写错、执行环境没有出网权限、或者把 Base URL 配成了带/v1的完整路径导致 SDK 拼接后变成/v1/v1/...。处理：确认TAOTOKEN_BASE_URL=https://taotoken.net/api，不要加/v1；确认 Harness 执行集群能访问外网；如果公司网络有出口限制，需要让网络同学放行。

reading 'choices'。典型报错是TypeError: Cannot read properties of undefined (reading 'choices')或 Python 里的KeyError: 'choices'。这说明返回体里没有choices字段，通常是请求根本没成功，返回的是错误 JSON，但代码直接去取choices了。处理：在解析前先打印完整响应，判断是不是 401 或 429。429 是限流，需要降低并发或检查额度。另外，如果模型返回的是流式响应而你没处理，也会出现类似问题，DST 场景建议先用非流式。

OAuth 相关报错。如果你在 Harness 里用 OAuth 方式连接代码仓库或外部服务，可能会看到OAuth token expired或invalid_grant。这跟 TaoToken 通道无关，是 Harness 连接器的问题。处理：到 Harness 的 Connectors 页面重新授权。注意区分：模型通道用 API Key，代码仓库用 OAuth，两者不要混。

模型返回非 JSON。现象是json.loads抛异常。原因可能是模型不支持response_format，或者 prompt 不够强硬。处理：去掉response_format，在 prompt 末尾加「只返回 JSON，不要 markdown 代码块，不要解释」。如果模型还是加代码块，就在解析前用正则把json 和去掉。

槽位第二轮丢失。这不是报错，但属于逻辑失败。原因通常是merge_state没做，或者每轮都新建了 state。处理：确认 state 在循环外初始化，每轮用merge_state合并，而不是覆盖。另外，如果模型在第二轮返回了空槽位，merge_state里的if v判断会跳过空值，保证旧槽位不被清掉。

Harness 变量没替换。现象是脚本里拿到的 Base URL 是字面量<+pipeline.variables.taotoken_base_url>。原因是在 YAML 里用了单引号，或者变量定义在错误的 scope。处理：去掉单引号，确认变量定义在 pipeline 级别，并且表达式拼写正确。

权限校验失败。如果你在 DST 里加了 Harness RBAC 校验，可能会遇到 403。处理：确认调用 Harness API 的 Service Account 有对应权限，并且实体确实属于当前租户。DST 状态里不要存跨租户的实体，这是硬性要求。

排查时记住一个顺序：先确认通道（curl 通不通），再确认脚本（本地跑不跑得通），最后确认 Harness 环境（变量、Secret、网络）。这个顺序能帮你少走很多弯路。

6. 把 DST 验证链路固化到日常交付

走到这里，你已经有了可复制的 pipeline YAML、环境变量配置、DST 用例和两段验证动作。接下来最重要的是把它变成日常习惯，而不是一次性 demo。

我的做法是：把dst_cases.json当成产品需求来维护。每次线上出现「AI 助手答非所问」的反馈，就把它转成一条 DST 用例，补上期望槽位，然后跑流水线。这样 DST 的覆盖会越来越厚，回归也越来越稳。

另外，模型和 prompt 是会变的。今天用gpt-4o-mini抽取稳定，明天换个模型可能就飘。所以流水线里的断言不能省。宁可多花几分钟跑用例，也不要等用户投诉了才发现状态丢了。

如果你需要更细的接入文档，包括不同语言的 SDK 示例和错误码说明，可以看接入文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite 。里面关于 Base URL 和鉴权的部分，和本篇配置是对应的。

最后留一个实用技巧：在 Harness 流水线里加一个「状态快照归档」步骤，把每轮state=的日志存成 artifact。这样当 DST 断言失败时，你能直接对比历史快照，看是哪个槽位在哪一轮开始漂移的。这个动作很小，但排查效率提升很明显。