From 5c0826bc88a2c61ef048a70ace35fcf54e237566 Mon Sep 17 00:00:00 2001 From: Yizhuo Zhang Date: Sun, 18 Oct 2026 15:58:27 -0700 Subject: [PATCH 1/3] feat(aiperf): qualify native text conversation history Forward ordered text messages through the existing TextConversation SDK. Keep Qwen chat rendering and tokenization family-owned, and reuse published AIPerf multi-turn loading with live assistant responses and explicit think time. Verify every original user turn and retained native answer by conversation and request identity, and report client and Task timings per turn. Existing models, bundles, core contracts and numerical criteria remain unchanged. Refs: #1679 Signed-off-by: Yizhuo Zhang --- apps/aiperf_qual/README.md | 39 +++++ .../config/text/qwen3-0.6b-conversations.yaml | 42 +++++ .../tests/test_text_conversations.py | 92 ++++++++++ .../trtmc_aiperf_qual/text_conversations.py | 158 ++++++++++++++++++ .../trtmc_aiperf_qual/text_corpus.py | 23 ++- .../trtmc_aiperf_qual/text_profile.py | 15 +- apps/server/native_worker.cpp | 75 ++++++++- apps/server/python/tests/test_app.py | 41 +++++ apps/server/python/trtmc_server/app.py | 30 +++- apps/server/python/trtmc_server/protocol.py | 20 ++- apps/server/python/trtmc_server/schemas.py | 2 +- apps/server/tests/test_sdk_worker.cpp | 20 ++- families/qwen/runtime/chat_templates.cpp | 35 ++++ families/qwen/runtime/chat_templates.h | 6 + families/qwen/runtime/pipeline.cpp | 65 ++++++- families/qwen/runtime/pipeline.h | 4 + .../tests/cpp/test_qwen_chat_templates.cpp | 42 ++++- .../user-guides/profile-text-with-aiperf.md | 27 +++ .../docs/user-guides/serve-text-generation.md | 12 +- 19 files changed, 710 insertions(+), 38 deletions(-) create mode 100644 apps/aiperf_qual/config/text/qwen3-0.6b-conversations.yaml create mode 100644 apps/aiperf_qual/tests/test_text_conversations.py create mode 100644 apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py diff --git a/apps/aiperf_qual/README.md b/apps/aiperf_qual/README.md index ca9e4063d5..8340e40165 100644 --- a/apps/aiperf_qual/README.md +++ b/apps/aiperf_qual/README.md @@ -744,3 +744,42 @@ A selection failure or changed/missing input/output fails verification and preserves its ledger and request artifacts. Family quality criteria are unchanged; this workload verifies transport and reports performance, without adding a new numerical accuracy gate. + +### Pinned conversations + +Use the same environment and already-built Qwen3-0.6B FP16 bundle with +`config/text/qwen3-0.6b-conversations.yaml`: + +```bash +python -m trtmc_aiperf_qual text-profile \ + --environment environment.yaml --config qwen3-0.6b-conversations.yaml \ + --out artifacts/qwen-conversations +``` + +The current Qwen family library implements the existing native +`TextConversation` contract for supported ChatML templates. The server forwards +typed roles and text; Qwen owns template rendering and tokenization. Existing +single-turn generation and its correctness criteria remain unchanged. Rebuild +the server/runtime libraries for history support; existing engine bundles can +be reused. This profile uses nonstreaming chat; history streaming, tools and +non-text messages are unsupported. + +The profile pins the same ShareGPT file and tokenizer as the single-turn +example. It selects six distinct fitting human/assistant/human prefixes and +sends their first two original human strings through AIPerf's existing +`multi_turn` loader. AIPerf captures actual native answers for subsequent +history (`deltas_without_responses`); source assistant answers are not replayed. +The second turn has 50 ms of think time. Twelve measured requests cover both +turns of all six conversations, with two separate warmup requests. + +`corpus.json` preserves source indices, original user strings, considered +rejections, the selection budget and the exact delay policy. The run's +`corpus-verification.json` checks roles/order, original text, retained actual +assistant output, native dispatch hashes, and every measured turn by request +and conversation identity. Actual wire histories are counted with the pinned +tokenizer and checked against the bundle context plus requested output; +native input-token counts remain unavailable. Configured and observed +between-turn gaps are recorded separately, without claiming exact scheduler +precision. `report.json` includes client and native Task-call timings for each +turn as well as the full workload. Startup/build/load/warmup are excluded from +measured requests. No TTFT/ITL or cross-request cache acceleration is claimed. diff --git a/apps/aiperf_qual/config/text/qwen3-0.6b-conversations.yaml b/apps/aiperf_qual/config/text/qwen3-0.6b-conversations.yaml new file mode 100644 index 0000000000..c3e062fddc --- /dev/null +++ b/apps/aiperf_qual/config/text/qwen3-0.6b-conversations.yaml @@ -0,0 +1,42 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# Replace /path values with paths on the validation host. Validation must use +# this bundle and the same family DSO that the server will load. +aiperf: + version: 0.13.0 +checkpoint: + name: Qwen/Qwen3-0.6B + revision: c1899de289a04d12100db370d81485cdf75e47ca + precision: fp16 +tokenizer: + name: Qwen/Qwen3-0.6B + revision: c1899de289a04d12100db370d81485cdf75e47ca +server: + bundle: /path/bundles/qwen3-0.6b-fp16.bundle + model_name: Qwen/Qwen3-0.6B + binary: /path/build/trtmc-server + runtime_root: /path/build + replicas: 1 + port: 8000 + startup_timeout: 120 + request_timeout: 120 +validation_profile: qwen3-0.6b-fp16 +validation_commands: + - [/path/trtmc-python, -m, pytest, families/qwen/tests/test_e2e.py, --e2e-model, qwen3-0.6b-fp16, -q] +# A workload configuration for the existing full FP16 model profile above. +corpus: + format: sharegpt_multi_turn + count: 6 + think_time_ms: 50 + source: + repo: anon8231489123/ShareGPT_Vicuna_unfiltered + revision: 192ab2185289094fc556ec8ce5ce1e8e587154ca + filename: ShareGPT_V3_unfiltered_cleaned_split.json + sha256: 35f0e213ce091ed9b9af2a1f0755e9d39f9ccec34ab281cd4ca60d70f6479ba4 +workload: + endpoints: [chat] + output_tokens: [16] + concurrency: [1] + streaming: [false] + requests: 12 + warmup_requests: 2 diff --git a/apps/aiperf_qual/tests/test_text_conversations.py b/apps/aiperf_qual/tests/test_text_conversations.py new file mode 100644 index 0000000000..3aee502bc6 --- /dev/null +++ b/apps/aiperf_qual/tests/test_text_conversations.py @@ -0,0 +1,92 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +import hashlib +import json +from types import SimpleNamespace + +import pytest + +from trtmc_aiperf_qual.config import ConfigError +from trtmc_aiperf_qual.text_conversations import fingerprint, select, verify_records +from trtmc_aiperf_qual.text_profile import arguments, cases + + +def conversation(first, second): + return {"conversations": [{"from": "human", "value": first}, {"from": "gpt", "value": "source answer"}, + {"from": "human", "value": second}]} + + +def count_messages(messages): + return sum(len(message["content"]) for message in messages) + 10 + + +def test_original_user_turns_survive_selection_and_rejections_are_visible(): + original = " first\n\u2028second " + row = conversation(original, "Follow up?") + selected, ledger = select([None, conversation("x" * 200, "Q"), row, row, + conversation("another", "follow up")], 2, 100, 8, count_messages) + assert [entry["source_index"] for entry in selected] == [2, 4] + assert selected[0]["user_turns"] == [original, "Follow up?"] + assert [entry["status"] for entry in ledger] == ["rejected", "rejected", "selected", "rejected", "selected"] + + +def test_multi_turn_uses_the_existing_loader_and_counts_both_requests(tmp_path): + config = {"corpus": {"format": "sharegpt_multi_turn", "count": 2}, + "tokenizer": {"name": "pinned", "revision": "a" * 40}, + "workload": {"endpoints": ["chat"], "requests": 4, "warmup_requests": 2}} + case, = cases(config) + args = arguments("http://localhost", config, case, input_file=tmp_path / "inputs.jsonl") + assert args[args.index("--custom-dataset-type") + 1] == "multi_turn" + assert "--synthetic-input-tokens-mean" not in args + config["workload"]["requests"] = 3 + with pytest.raises(ConfigError, match="every prompt"): + cases(config) + + +@pytest.fixture +def capture(): + texts = ["Remember ORCHID.", "What word?"] + original = [{"role": "user", "content": texts[0]}] + history = [*original, {"role": "assistant", "content": "OK"}, {"role": "user", "content": texts[1]}] + raw, joined = [], [] + for turn, (messages, text) in enumerate([(original, "OK"), (history, "ORCHID")]): + metadata = {"x_request_id": str(turn), "conversation_id": "sharegpt-0", "session_num": 0, + "x_correlation_id": "conversation", "turn_index": turn, "benchmark_phase": "profiling", + "request_start_ns": 1_000_000_000 + turn * 150_000_000, + "request_end_ns": 1_100_000_000 + turn * 150_000_000} + raw.append({"metadata": metadata, "payload": {"messages": messages}, "status": 200, + "responses": [{"text": json.dumps({"choices": [{"message": {"content": text}}]})}]}) + joined.append({"request_id": str(turn), "server": {"status": 200, + "native_operation": "converse" if turn else "generate", + "prompt_sha256": hashlib.sha256(texts[0].encode()).hexdigest(), + "conversation_sha256": fingerprint(messages), "completion_sha256": hashlib.sha256(text.encode()).hexdigest()}}) + corpus = {"selected": [{"session_id": "sharegpt-0", "user_turns": texts}], + "context_limit": 256, "output_tokens": 8, "think_time_ms": 50} + return SimpleNamespace(raw_records=lambda phase: raw), joined, corpus, raw + + +def test_live_assistant_history_and_per_turn_identity_are_verified(capture): + run, joined, corpus, _ = capture + result = verify_records(run, joined, corpus, count_messages) + assert result["verified"] and result["measured_conversations"] == 1 + assert [row["turn_index"] for row in result["requests"]] == [0, 1] + assert result["requests"][1]["observed_gap_ms"] == 50 + + +@pytest.mark.parametrize("change", ["drop_history", "role", "original_text", "assistant_text", "native_history", "capacity"]) +def test_missing_or_changed_history_cannot_be_reported_as_qualified(capture, change): + run, joined, corpus, raw = capture + if change == "drop_history": + raw[1]["payload"]["messages"] = raw[1]["payload"]["messages"][-1:] + elif change == "role": + raw[1]["payload"]["messages"][1]["role"] = "user" + elif change == "original_text": + raw[1]["payload"]["messages"][2]["content"] = "rewritten" + elif change == "assistant_text": + raw[1]["payload"]["messages"][1]["content"] = "source answer" + elif change == "native_history": + joined[1]["server"]["conversation_sha256"] = "0" * 64 + else: + corpus["context_limit"] = 16 + with pytest.raises(ConfigError): + verify_records(run, joined, corpus, count_messages) diff --git a/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py b/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py new file mode 100644 index 0000000000..c4c5d86365 --- /dev/null +++ b/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py @@ -0,0 +1,158 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Fixed original user turns using AIPerf's existing live-response history.""" +from __future__ import annotations + +import hashlib +import json +import math +from pathlib import Path + +from .config import ConfigError +from .text_corpus import load_source +from .text_profile import digest, response_text + + +def fingerprint(value) -> str: + return hashlib.sha256(json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()).hexdigest() + + +def select(records, count, capacity, output, count_messages): + if any(type(value) is not int or value < 1 for value in (count, capacity, output)) or output >= capacity: + raise ConfigError("conversation count/context/output must be positive integers fitting the bundle") + selected, ledger, seen = [], [], set() + for index, row in enumerate(records): + turns = row.get("conversations", []) if isinstance(row, dict) else [] + entry = {"source_index": index} + if (not isinstance(turns, list) or len(turns) < 3 or + any(not isinstance(turns[i], dict) or turns[i].get("from") != role or + not isinstance(turns[i].get("value"), str) or not turns[i]["value"].strip() + for i, role in enumerate(("human", "gpt", "human")))): + entry.update(status="rejected", reason="no original human/assistant/human prefix") + else: + texts = (turns[0]["value"], turns[2]["value"]) + base = count_messages([{"role": "user", "content": texts[0]}, + {"role": "assistant", "content": ""}, + {"role": "user", "content": texts[1]}]) + entry["template_tokens_without_live_answer"] = base + if texts in seen or base + 2 * output > capacity: + entry.update(status="rejected", reason="duplicate turns or conversation/output reserve exceeds context") + else: + entry.update(status="selected", session_id=f"sharegpt-{index}", user_turns=list(texts)) + selected.append(entry) + seen.add(texts) + ledger.append(entry) + if len(selected) == count: + break + return selected, ledger + + +def prepare(config, capacity: int, output: int, out: Path): + source, source_hash, records, tokenizer = load_source(config) + spec = config["corpus"] + delay = spec.get("think_time_ms", 50) + if type(delay) not in (int, float) or not math.isfinite(delay) or delay < 0: + raise ConfigError("think_time_ms must be a finite nonnegative number") + def count_messages(messages): + return len(tokenizer.apply_chat_template( + messages, tokenize=True, add_generation_prompt=True, enable_thinking=False)) + selected, ledger = select(records, spec["count"], capacity, output, count_messages) + path = out / "inputs.jsonl" + path.write_text("".join(json.dumps({"session_id": row["session_id"], "turns": [ + {"text": text, "delay": 0 if index == 0 else delay, "output_length": output} + for index, text in enumerate(row["user_turns"])]}, ensure_ascii=False) + "\n" + for row in selected), encoding="utf-8") + report = {"format": "sharegpt_multi_turn", "source": source, "source_sha256": source_hash, + "tokenizer": config["tokenizer"], "context_limit": capacity, "output_tokens": output, + "think_time_ms": delay, "source_records": len(records), "considered_records": len(ledger), + "unexamined_records": len(records) - len(ledger), "selected": selected, "ledger": ledger, + "selection_policy": "first distinct fitting original human/assistant/human prefixes; send the first two human strings unchanged", + "assistant_history_policy": "AIPerf deltas_without_responses captures actual native answers; source assistant answers are not replayed", + "token_count_scope": "pinned tokenizer chat template; native input count unavailable; actual wire history counted after the run", + "input_file": str(path), "input_file_sha256": digest(path)} + (out / "corpus.json").write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") + if len(selected) != spec["count"]: + raise ConfigError("not enough original conversations fit; see corpus.json; no history was truncated") + return report + + +def messages_from_payload(payload): + messages = payload.get("messages") if isinstance(payload, dict) else None + if not isinstance(messages, list): + raise ConfigError("conversation payload is missing messages") + normalized = [] + for message in messages: + if not isinstance(message, dict): + raise ConfigError("conversation message is not an object") + content = message.get("content") + if isinstance(content, list) and all(isinstance(p, dict) and p.get("type") == "text" and + isinstance(p.get("text"), str) for p in content): + content = "".join(p["text"] for p in content) + if not isinstance(content, str): + raise ConfigError("conversation content is not text") + normalized.append({"role": message.get("role"), "content": content}) + return normalized + + +def verify(run, joined, corpus): + from transformers import AutoTokenizer + tokenizer = AutoTokenizer.from_pretrained(corpus["tokenizer"]["name"], + revision=corpus["tokenizer"]["revision"], trust_remote_code=False) + return verify_records(run, joined, corpus, lambda messages: len(tokenizer.apply_chat_template( + messages, tokenize=True, add_generation_prompt=True, enable_thinking=False))) + + +def verify_records(run, joined, corpus, count_messages): + expected = {row["session_id"]: row for row in corpus["selected"]} + servers = {row["request_id"]: row["server"] for row in joined} + groups, seen, measured, identities = {}, set(), set(), [] + raw_records = run.raw_records(phase=None) + for raw in raw_records: + meta = raw.get("metadata", {}) + if (not isinstance(meta, dict) or not isinstance(meta.get("x_request_id"), str) or + not isinstance(meta.get("conversation_id"), str) or type(meta.get("session_num")) is not int or + type(meta.get("turn_index")) is not int or meta["turn_index"] not in (0, 1) or + any(type(meta.get(name)) is not int or meta[name] < 1 for name in ("request_start_ns", "request_end_ns"))): + raise ConfigError("conversation request metadata is missing identity or timing") + for raw in sorted(raw_records, key=lambda r: (r["metadata"]["session_num"], r["metadata"]["turn_index"])): + meta = raw["metadata"] + request_id, session, turn = meta.get("x_request_id"), meta.get("conversation_id"), meta.get("turn_index") + if request_id in seen or session not in expected or type(turn) is not int or turn not in (0, 1): + raise ConfigError("unknown or repeated conversation request identity") + seen.add(request_id) + key = (meta.get("benchmark_phase"), meta["session_num"]) + previous = groups.setdefault(key, []) + if len(previous) != turn or (previous and previous[0]["session"] != session): + raise ConfigError("conversation turns are missing, repeated or out of order") + source = expected[session] + wanted = [{"role": "user", "content": source["user_turns"][0]}] + if turn: + wanted += [{"role": "assistant", "content": previous[0]["text"]}, + {"role": "user", "content": source["user_turns"][1]}] + messages = messages_from_payload(raw.get("payload")) + text = response_text(raw) + server = servers.get(request_id) + if (messages != wanted or not isinstance(text, str) or server is None or server.get("status") != 200 or + server.get("completion_sha256") != hashlib.sha256(text.encode()).hexdigest()): + raise ConfigError("conversation lost original text, roles, actual assistant history or native output") + if turn and (server.get("native_operation") != "converse" or server.get("conversation_sha256") != fingerprint(messages)): + raise ConfigError("native conversation dispatch changed history") + if not turn and server.get("prompt_sha256") != hashlib.sha256(source["user_turns"][0].encode()).hexdigest(): + raise ConfigError("native first-turn dispatch changed the original text") + tokens = count_messages(messages) + if tokens + corpus["output_tokens"] > corpus["context_limit"]: + raise ConfigError("actual conversation exceeds bundle context; no history may be dropped") + gap = ((meta["request_start_ns"] - previous[0]["end_ns"]) / 1e6) if turn else None + if gap is not None and gap < 0: + raise ConfigError("conversation dispatched a turn before its previous response completed") + previous.append({"text": text, "end_ns": meta["request_end_ns"], "session": session}) + if meta.get("benchmark_phase") == "profiling": + measured.add((session, turn)) + identities.append({"request_id": request_id, "phase": meta.get("benchmark_phase"), + "session_id": session, "session_num": meta["session_num"], "turn_index": turn, + "correlation_id": meta.get("x_correlation_id"), "input_tokens": tokens, + "messages_sha256": fingerprint(messages), "completion_sha256": server["completion_sha256"], + "configured_think_time_ms": corpus["think_time_ms"] if turn else 0, "observed_gap_ms": gap}) + if measured != {(session, turn) for session in expected for turn in (0, 1)}: + raise ConfigError("profiling did not cover both turns of every selected conversation") + return {"verified": True, "measured_conversations": len(expected), "requests": identities} diff --git a/apps/aiperf_qual/trtmc_aiperf_qual/text_corpus.py b/apps/aiperf_qual/trtmc_aiperf_qual/text_corpus.py index aab734724d..970c473977 100644 --- a/apps/aiperf_qual/trtmc_aiperf_qual/text_corpus.py +++ b/apps/aiperf_qual/trtmc_aiperf_qual/text_corpus.py @@ -43,14 +43,14 @@ def select_prompts(records: list[Any], count: int, context: int, output: int, return selected, ledger -def prepare(config: dict[str, Any], capacity: int, output: int, out: Path) -> dict[str, Any]: +def load_source(config: dict[str, Any]): from transformers import AutoTokenizer from .suites import _hf_file, _read_records spec = config["corpus"] source = spec["source"] - if spec.get("format") != "sharegpt" or not SHA.fullmatch(str(source.get("revision", ""))): - raise ConfigError("real prompts require ShareGPT format and an immutable dataset revision") + if not SHA.fullmatch(str(source.get("revision", ""))): + raise ConfigError("real prompts require an immutable dataset revision") path = _hf_file(source) source_hash = digest(path) if source.get("sha256") and source["sha256"] != source_hash: @@ -60,6 +60,17 @@ def prepare(config: dict[str, Any], capacity: int, output: int, out: Path) -> di raise ConfigError("ShareGPT source must contain a list of conversations") tokenizer = AutoTokenizer.from_pretrained(config["tokenizer"]["name"], revision=config["tokenizer"]["revision"], trust_remote_code=False) + return source, source_hash, records, tokenizer + + +def prepare(config: dict[str, Any], capacity: int, output: int, out: Path) -> dict[str, Any]: + if config["corpus"].get("format") == "sharegpt_multi_turn": + from .text_conversations import prepare as prepare_conversations + return prepare_conversations(config, capacity, output, out) + if config["corpus"].get("format") != "sharegpt": + raise ConfigError("unsupported real-prompt corpus format") + spec = config["corpus"] + source, source_hash, records, tokenizer = load_source(config) selected, ledger = select_prompts(records, spec["count"], capacity, output, lambda text: tokenizer.encode(text, add_special_tokens=False)) prepared = out / "inputs.jsonl" @@ -113,7 +124,11 @@ def verify_and_record(run, joined: list[dict[str, Any]], corpus: dict[str, Any]) """Preserve a verification verdict for missing, changed or malformed inputs.""" path = run.directory / "corpus-verification.json" try: - verified = verify(run, joined, corpus) + if corpus.get("format") == "sharegpt_multi_turn": + from .text_conversations import verify as verify_conversations + verified = verify_conversations(run, joined, corpus) + else: + verified = verify(run, joined, corpus) try: unchanged = digest(Path(corpus["input_file"])) == corpus["input_file_sha256"] except OSError as error: diff --git a/apps/aiperf_qual/trtmc_aiperf_qual/text_profile.py b/apps/aiperf_qual/trtmc_aiperf_qual/text_profile.py index 90508d522f..3dd2ce7128 100644 --- a/apps/aiperf_qual/trtmc_aiperf_qual/text_profile.py +++ b/apps/aiperf_qual/trtmc_aiperf_qual/text_profile.py @@ -102,10 +102,13 @@ def cases(config: dict[str, Any]) -> list[dict[str, Any]]: raise ConfigError(f"{name} has an invalid request count") if config.get("corpus"): count = config["corpus"].get("count") - if type(count) is not int or count < 1 or workload.get("requests", 20) < count: + multi_turn = config["corpus"].get("format") == "sharegpt_multi_turn" + if multi_turn and len({item["output_tokens"] for item in result}) != 1: + raise ConfigError("conversation profiles require one output-token cap for their per-turn settings") + if type(count) is not int or count < 1 or workload.get("requests", 20) < count * (2 if multi_turn else 1): raise ConfigError("corpus count must be positive and every prompt needs a measured request") - if any(item["endpoint"] != "completions" or item["streaming"] for item in result): - raise ConfigError("real-prompt profiles currently support nonstreaming completions") + if any(item["endpoint"] != ("chat" if multi_turn else "completions") or item["streaming"] for item in result): + raise ConfigError("real-prompt profiles require nonstreaming chat for conversations or completions for single turns") return result @@ -121,7 +124,8 @@ def arguments(url: str, config: dict[str, Any], case: dict[str, Any], *, input_f if input_file is None: result += ["--synthetic-input-tokens-mean", str(case["input_tokens"]), "--synthetic-input-tokens-stddev", "0"] else: - result += ["--input-file", str(input_file), "--custom-dataset-type", "single_turn"] + result += ["--input-file", str(input_file), "--custom-dataset-type", + "multi_turn" if config.get("corpus", {}).get("format") == "sharegpt_multi_turn" else "single_turn"] if workload.get("warmup_requests", 3): result += ["--warmup-request-count", str(workload.get("warmup_requests", 3)), "--warmup-concurrency", "1"] result += ["--extra-inputs", "temperature:0", "seed:0", "top_k:1"] @@ -305,6 +309,9 @@ def profile(environment: Environment, config_path: Path, out: Path) -> dict[str, entries.append({"case": case, "exit_code": run.exit_code or int(incomplete), "incomplete_export": incomplete, "command": run.command, "summary": summarize(rows), "aiperf_summary": run.summary}) + if corpus and corpus.get("format") == "sharegpt_multi_turn": + entries[-1]["per_turn"] = {str(turn): summarize([row for row in rows + if row["client_metadata"].get("turn_index") == turn]) for turn in (0, 1)} (out / "report.json").write_text(json.dumps({"runs": entries}, indent=2) + "\n") if entries[-1]["exit_code"]: break diff --git a/apps/server/native_worker.cpp b/apps/server/native_worker.cpp index ca68a1dcfe..6bf35d81fd 100644 --- a/apps/server/native_worker.cpp +++ b/apps/server/native_worker.cpp @@ -279,20 +279,21 @@ int run_protocol(const Json& ready, const Operations& operations, std::istream& template int run_text_protocol(std::int64_t default_tokens, Generate&& generate, std::istream& input, - std::ostream& output, StreamGenerate streaming = {}) { + std::ostream& output, StreamGenerate streaming = {}, + Operations operations = {}) { Json capabilities = Json::array({"text_generation"}); if (streaming) capabilities.push_back("streaming_text_continuation"); + if (operations.count("converse")) + capabilities.push_back("text_conversation"); const Json ready{{"event", "ready"}, {"protocol_version", 2}, {"capabilities", capabilities}, {"default_max_new_tokens", default_tokens}}; - return run_protocol(ready, - {{"generate", - [&](const Json& request) { - return generate(string_field(request, "prompt", true), request); - }}}, - input, output, std::move(streaming)); + operations.emplace("generate", [&](const Json& request) { + return generate(string_field(request, "prompt", true), request); + }); + return run_protocol(ready, operations, input, output, std::move(streaming)); } const std::vector& checked_embedding_values(const EmbeddingResult& result) { @@ -943,6 +944,64 @@ int run_model_worker(const Model& model, std::istream& input, std::ostream& outp return run_text_worker(model, input, output); } +Operations conversation_operations(const Model* model) { + if (model == nullptr || !model->supports()) + return {}; + const auto task = model->task(); + const auto fields = task.config_fields(); + return {{"converse", [task, fields](const Json& request) { + const auto found = request.find("messages"); + if (found == request.end() || !found->is_array() || found->empty() || + found->size() > 64) + throw ProtocolError("messages must contain between 1 and 64 text messages"); + TextConversationRequest conversation; + for (const auto& message : *found) { + const auto role = string_field(message, "role"); + MessageRole typed; + if (role == "system") + typed = MessageRole::System; + else if (role == "user") + typed = MessageRole::User; + else if (role == "assistant") + typed = MessageRole::Assistant; + else + throw ProtocolError("unsupported conversation role"); + conversation.messages.push_back( + {typed, {TextPart{string_field(message, "content", true)}}}); + } + const auto config = parse_sdk_config(request, fields); + try { + const auto started = std::chrono::steady_clock::now(); + const auto result = task.run(conversation, config); + const auto wall_ms = std::chrono::duration( + std::chrono::steady_clock::now() - started) + .count(); + std::string text; + const auto message = result.message(); + for (const auto& part : message.parts) { + const auto* value = std::get_if(&part); + if (value == nullptr) + throw std::runtime_error( + "conversation Task returned a non-text assistant part"); + text += value->text; + } + const auto view = result.result().c_view(); + return Json{{"text", text}, + {"completion_tokens", result.result().token_ids().size()}, + {"setup_ms", view.setup_ms}, + {"prefill_ms", view.prefill_ms}, + {"decode_ms", view.decode_ms}, + {"model_call_ms", wall_ms}, + {"timing_scope", "public_task_call_wall"}}; + } catch (const Error& error) { + if (error.code() == TRTMC_INVALID_ARGUMENT || + error.code() == TRTMC_INVALID_CONFIG) + throw ProtocolError("conversation request is invalid for this model"); + throw; + } + }}}; +} + template int run_sdk_worker(const Task& task, Request&& make_request, std::istream& input, std::ostream& output, const Model* model = nullptr) { @@ -1025,7 +1084,7 @@ int run_sdk_worker(const Task& task, Request&& make_request, std::istream& input throw; } }, - input, output, std::move(streaming)); + input, output, std::move(streaming), conversation_operations(model)); } } // namespace diff --git a/apps/server/python/tests/test_app.py b/apps/server/python/tests/test_app.py index 0a88c718bc..170ba0a8cc 100644 --- a/apps/server/python/tests/test_app.py +++ b/apps/server/python/tests/test_app.py @@ -87,6 +87,9 @@ def models(self) -> list[dict[str, Any]]: def supports_streaming(self, model: str) -> bool: return False + def supports(self, model: str, capability: str) -> bool: + return capability == "text_conversation" and getattr(self, "conversation", False) + def max_tokens(self, model: str, hard_cap: int) -> int: if model != "test/model": raise ModelNotFoundError(model) @@ -133,6 +136,44 @@ def test_real_prompt_and_native_output_identity_survive_http(tmp_path): assert "prompt" not in row and "text" not in row +def test_history_forwards_typed_roles_and_content_without_a_client_template(tmp_path): + import hashlib + + registry = FakeRegistry() + registry.conversation = True + path = tmp_path / "records.jsonl" + messages = [{"role": "system", "content": " system\n"}, + {"role": "user", "content": "Remember ORCHID."}, + {"role": "assistant", "content": "OK\u2028"}, + {"role": "user", "content": "What word?"}] + app = create_app(registry, ServerConfig(max_body_bytes=4096, max_prompt_bytes=1024, + max_generation_tokens=64, records=path)) + with TestClient(app) as client: + response = client.post("/v1/chat/completions", json={"model": "test/model", "messages": messages}) + assert response.status_code == 200 + operation, payload = registry.requests[0] + assert operation == "converse" and payload["messages"] == messages + assert "prompt" not in payload and "use_chat_template" not in payload["config"] + row, = [json.loads(line) for line in path.read_text().split("\n") if line] + assert row["conversation_sha256"] == hashlib.sha256(json.dumps( + messages, ensure_ascii=False, separators=(",", ":")).encode()).hexdigest() + assert row["conversation_roles"] == ["system", "user", "assistant", "user"] + assert "messages" not in row and row["native_operation"] == "converse" + + +def test_history_rejects_unsupported_models_streaming_and_full_history_limits(): + registry = FakeRegistry() + messages = [{"role": "user", "content": "first"}, {"role": "assistant", "content": "answer"}, + {"role": "user", "content": "last"}] + with make_client(registry) as client: + assert client.post("/v1/chat/completions", json={"model": "test/model", "messages": messages}).status_code == 400 + registry.conversation = True + assert client.post("/v1/chat/completions", json={"model": "test/model", "messages": messages, "stream": True}).status_code == 400 + messages[0]["content"] = "x" * 2048 + assert client.post("/v1/chat/completions", json={"model": "test/model", "messages": messages}).status_code == 413 + assert registry.requests == [] + + def test_completion_and_chat_keep_model_semantics_in_worker(caplog: Any) -> None: registry = FakeRegistry() caplog.set_level(logging.INFO, logger="uvicorn.error") diff --git a/apps/server/python/trtmc_server/app.py b/apps/server/python/trtmc_server/app.py index 887daa35de..5a02c843e3 100644 --- a/apps/server/python/trtmc_server/app.py +++ b/apps/server/python/trtmc_server/app.py @@ -35,7 +35,7 @@ WorkerTimeoutError, ) from .metrics import Metrics -from .protocol import chat_prompt, chat_image, extract_vision_result, extract_embedding, extract_ranking, extract_result, extract_transcription, extract_detection, generation_config, public_worker_error +from .protocol import chat_prompt, chat_history, chat_image, extract_vision_result, extract_embedding, extract_ranking, extract_result, extract_transcription, extract_detection, generation_config, public_worker_error from .records import RequestPolicy, RequestRecords from .registry import ModelRegistry from .schemas import ChatCompletionRequest, CompletionRequest, EmbeddingRequest, GenerationRequest, RankingRequest, TranscriptionRequest, DetectionRequest @@ -318,6 +318,7 @@ async def execute( system_prompt: str = "", chat: bool, chat_max_tokens: int | None = None, + messages: list[dict[str, str]] | None = None, ) -> Any: context["model"] = request.model context.update(prompt_sha256=hashlib.sha256(prompt.encode("utf-8")).hexdigest(), @@ -338,7 +339,9 @@ async def execute( return error_response( 400, "unsupported_parameter", "stop is not supported", param="stop" ) - if len(prompt.encode("utf-8")) + len(system_prompt.encode("utf-8")) > config.max_prompt_bytes: + input_bytes = (sum(len(message["content"].encode("utf-8")) for message in messages) if messages is not None + else len(prompt.encode("utf-8")) + len(system_prompt.encode("utf-8"))) + if input_bytes > config.max_prompt_bytes: metrics.reject(route, 413) return error_response( 413, "content_too_large", "prompt exceeds the server limit", param="prompt" @@ -349,6 +352,14 @@ async def execute( except ModelNotFoundError as error: metrics.reject(route, 404) return error_response(404, error.code, str(error), param="model") + if messages is not None: + if request.stream or not registry.supports(request.model, "text_conversation"): + metrics.reject(route, 400) + return error_response(400, "unsupported_parameter", "this model requires nonstreaming native conversation support for history", + param="stream" if request.stream else "messages") + encoded = json.dumps(messages, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + context.update(conversation_sha256=hashlib.sha256(encoded).hexdigest(), + conversation_messages=len(messages), conversation_roles=[message["role"] for message in messages]) max_tokens = requested_tokens or default_tokens if max_tokens > config.max_generation_tokens: metrics.reject(route, 400) @@ -382,13 +393,17 @@ async def execute( context["replica"] = getattr(session, "replica_name", None) context["admission_ms"] = queue_seconds * 1000 worker_config = generation_config(request, max_tokens) - worker_config["use_chat_template"] = chat - if system_prompt: - worker_config["system_prompt"] = system_prompt + if messages is None: + worker_config["use_chat_template"] = chat + if system_prompt: + worker_config["system_prompt"] = system_prompt context["generation_config"] = {key: value for key, value in worker_config.items() if key != "system_prompt"} context["has_system_prompt"] = bool(system_prompt) metrics.begin() inference_started = time.monotonic() + operation = "converse" if messages is not None else "generate" + context["native_operation"] = operation + payload = {"messages": messages, "config": worker_config} if messages is not None else {"prompt": prompt, "config": worker_config} if request.stream and registry.supports_streaming(request.model): return await incremental_response(session, {"prompt": prompt, "config": worker_config}, model=request.model, chat=chat, include_usage=(request.stream_options is not None @@ -396,7 +411,7 @@ async def execute( route=route, queue_seconds=queue_seconds, started=inference_started, metrics=metrics) try: result = await _worker_request( - session, {"prompt": prompt, "config": worker_config} + session, payload, operation=operation ) text, completion_tokens, timings = extract_result(result) except asyncio.CancelledError: @@ -863,8 +878,8 @@ def validate(value): @app.post("/v1/chat/completions") async def chat_completions(http_request: Request, request: ChatCompletionRequest) -> Any: try: - prompt, system_prompt = chat_prompt(request) image_url = chat_image(request) + prompt, system_prompt = chat_prompt(request) except ValueError as error: metrics.reject("/v1/chat/completions", 400) return error_response(400, "invalid_request", str(error), param="messages") @@ -886,6 +901,7 @@ async def chat_completions(http_request: Request, request: ChatCompletionRequest system_prompt=system_prompt, chat=True, chat_max_tokens=request.max_completion_tokens, + messages=chat_history(request), ) videos = VideoJobs(registry, config, records, execute_task, error_response) diff --git a/apps/server/python/trtmc_server/protocol.py b/apps/server/python/trtmc_server/protocol.py index 852f8f270b..5c145c4681 100644 --- a/apps/server/python/trtmc_server/protocol.py +++ b/apps/server/python/trtmc_server/protocol.py @@ -24,15 +24,19 @@ def generation_config(request: GenerationRequest, max_tokens: int) -> dict[str, def chat_prompt(request: ChatCompletionRequest) -> tuple[str, str]: messages = request.messages - if len(messages) == 1 and messages[0].role == "user": - return _message_text(messages[0].content), "" - if ( - len(messages) == 2 - and messages[0].role == "system" - and messages[1].role == "user" + if messages[-1].role != "user" or any( + message.role not in {"user", "assistant"} and not (index == 0 and message.role == "system") + for index, message in enumerate(messages) ): - return _message_text(messages[1].content), _message_text(messages[0].content) - raise ValueError("messages must be one user message with an optional preceding system message") + raise ValueError("messages require user/assistant roles, an optional leading system, and a final user") + return _message_text(messages[-1].content), ( + _message_text(messages[0].content) if messages[0].role == "system" else "") + + +def chat_history(request: ChatCompletionRequest) -> list[dict[str, str]] | None: + if len(request.messages) == 1 or (len(request.messages) == 2 and request.messages[0].role == "system"): + return None + return [{"role": message.role, "content": _message_text(message.content)} for message in request.messages] def chat_image(request: ChatCompletionRequest) -> str | None: diff --git a/apps/server/python/trtmc_server/schemas.py b/apps/server/python/trtmc_server/schemas.py index 94af0e491e..47b051b0c5 100644 --- a/apps/server/python/trtmc_server/schemas.py +++ b/apps/server/python/trtmc_server/schemas.py @@ -58,7 +58,7 @@ class CompletionRequest(GenerationRequest): class ChatCompletionRequest(GenerationRequest): - messages: list[ChatMessage] = Field(min_length=1, max_length=2) + messages: list[ChatMessage] = Field(min_length=1, max_length=64) max_completion_tokens: int | None = Field(default=None, ge=1) diff --git a/apps/server/tests/test_sdk_worker.cpp b/apps/server/tests/test_sdk_worker.cpp index 2cdfa52e9e..44dee9fa1c 100644 --- a/apps/server/tests/test_sdk_worker.cpp +++ b/apps/server/tests/test_sdk_worker.cpp @@ -213,7 +213,8 @@ void test_incremental_stream(const std::filesystem::path& root, const trtmc::Loa "stream emits one delta and one terminal then shutdown"); require(run.messages[0].at("protocol_version") == 2 && run.messages[0].at("capabilities") == - Json::array({"text_generation", "streaming_text_continuation"}), + Json::array( + {"text_generation", "streaming_text_continuation", "text_conversation"}), "ready advertises actual SDK stream capability"); require(run.messages[1].at("id") == "stream" && run.messages[1].at("event") == "delta" && run.messages[1].at("result").at("text_delta") == "hello!", @@ -224,6 +225,23 @@ void test_incremental_stream(const std::filesystem::path& root, const trtmc::Loa "public_task_stream_wall_including_backpressure" && run.messages[2].at("result").at("model_call_ms").get() >= 0, "terminal result carries scoped timing"); + + const auto invalid_history = + invoke(bundle, options, + {{{"id", "bad-history"}, {"op", "converse"}, {"messages", Json::array()}}, + {{"id", "bad-role"}, + {"op", "converse"}, + {"messages", Json::array({{{"role", "tool"}, {"content", "unsupported"}}})}}, + {{"id", "after-history-errors"}, {"op", "generate"}, {"prompt", "hello"}}}); + require(invalid_history.status == 0 && invalid_history.messages.size() == 4, + "invalid conversation requests preserve the worker"); + for (std::size_t index = 1; index < 3; ++index) + require(invalid_history.messages[index].at("ok") == false && + invalid_history.messages[index].at("error").at("type") == + "invalid_request_error", + "bad histories are recoverable request errors"); + require(invalid_history.messages.back().at("result").at("text") == "sync", + "text generation remains usable after conversation errors"); } void test_audio_task(const std::filesystem::path& root, const trtmc::LoadOptions& options) { const auto bundle = write_bundle(root, "speech_transcription", "audio_fixture"); diff --git a/families/qwen/runtime/chat_templates.cpp b/families/qwen/runtime/chat_templates.cpp index 14adaf80ff..8b88105c1d 100644 --- a/families/qwen/runtime/chat_templates.cpp +++ b/families/qwen/runtime/chat_templates.cpp @@ -5,6 +5,7 @@ #include "families/qwen/runtime/chat_templates.h" +#include #include namespace trtmc { @@ -47,4 +48,38 @@ std::string qwen_apply_chat_template(const std::string& format, const std::strin return prompt; } +std::string +qwen_apply_conversation_template(const std::string& format, + const std::vector>& messages, + bool enable_thinking) { + if (format != "chatml" && format != "chatml_thinking") + throw std::invalid_argument("Qwen conversation requires a supported chat template"); + if (messages.empty() || messages.back().first != "user") + throw std::invalid_argument("Qwen conversation must end with a user message"); + std::string result; + for (std::size_t index = 0; index < messages.size(); ++index) { + const auto& [role, original] = messages[index]; + if (role != "user" && role != "assistant" && !(role == "system" && index == 0)) + throw std::invalid_argument( + "Qwen conversation supports user/assistant and a leading system message"); + auto content = original; + // The pinned hybrid Qwen3 template omits reasoning from assistant turns + // before the final user query. Preserve the original typed input; only + // this family-owned rendering applies that model-specific convention. + if (role == "assistant" && format == "chatml_thinking") { + const auto end = content.rfind(""); + if (end != std::string::npos) { + content.erase(0, end + std::string("").size()); + const auto first = content.find_first_not_of('\n'); + content.erase(0, first == std::string::npos ? content.size() : first); + } + } + result += "<|im_start|>" + role + "\n" + content + "<|im_end|>\n"; + } + result += "<|im_start|>assistant\n"; + if (format == "chatml_thinking" && !enable_thinking) + result += "\n\n\n\n"; + return result; +} + } // namespace trtmc diff --git a/families/qwen/runtime/chat_templates.h b/families/qwen/runtime/chat_templates.h index 7e5c787afd..7ff188e8d2 100644 --- a/families/qwen/runtime/chat_templates.h +++ b/families/qwen/runtime/chat_templates.h @@ -6,6 +6,8 @@ #pragma once #include +#include +#include namespace trtmc { @@ -13,5 +15,9 @@ std::string qwen_detect_chat_template_format(const std::string& jinja_template); std::string qwen_apply_chat_template(const std::string& format, const std::string& prompt, bool enable_thinking = true, const std::string& system_prompt = {}); +std::string +qwen_apply_conversation_template(const std::string& format, + const std::vector>& messages, + bool enable_thinking = true); } // namespace trtmc diff --git a/families/qwen/runtime/pipeline.cpp b/families/qwen/runtime/pipeline.cpp index 7cbd70dfd4..1b0024bb20 100644 --- a/families/qwen/runtime/pipeline.cpp +++ b/families/qwen/runtime/pipeline.cpp @@ -139,8 +139,16 @@ const ConfigField qwen_text_fields[] = { {"enable_thinking", ConfigKind::Bool, true, "Enable Qwen thinking"}, {"system_prompt", ConfigKind::String, std::string_view{}, "ChatML system message"}, }; -TextGenerationConfig qwen_text_config(internal::ConfigView supplied) { - internal::validate_config(qwen_text_fields, supplied); +const auto qwen_conversation_fields = [] { + std::vector fields; + for (const auto& field : qwen_text_fields) + if (field.name != "use_chat_template" && field.name != "system_prompt") + fields.push_back(field); + return fields; +}(); +TextGenerationConfig qwen_text_config(internal::ConfigView supplied, + Span fields = qwen_text_fields) { + internal::validate_config(fields, supplied); TextGenerationConfig result; const auto integer = [&](std::string_view name) { const auto value = *internal::config_get(supplied, qwen_text_fields, name); @@ -180,12 +188,65 @@ std::vector QwenTextGenerationPipeline::task_bindings() // move a tensor-parallel communicator onto a producer thread. std::vector result{ internal::bind(*this, qwen_text_fields)}; + if (config_.chat_template_format == "chatml" || + config_.chat_template_format == "chatml_thinking") + result.push_back(internal::bind( + *this, {qwen_conversation_fields.data(), qwen_conversation_fields.size()})); if (!distributed_owner_) result.push_back( internal::bind(*this, qwen_text_fields)); return result; } +internal::ConversationResult +QwenTextGenerationPipeline::run(const internal::TextConversationRequest& request, + internal::ConfigView config) { + if (!request.tools.empty()) + throw std::invalid_argument("Qwen conversation tools are unsupported"); + const auto cfg = qwen_text_config( + config, {qwen_conversation_fields.data(), qwen_conversation_fields.size()}); + std::vector> messages; + for (const auto& message : request.messages) { + std::string role; + switch (message.role) { + case internal::MessageRole::System: + role = "system"; + break; + case internal::MessageRole::User: + role = "user"; + break; + case internal::MessageRole::Assistant: + role = "assistant"; + break; + default: + throw std::invalid_argument("Qwen conversation role is unsupported"); + } + std::string text; + for (const auto& part : message.parts) { + const auto* value = std::get_if(&part); + if (value == nullptr) + throw std::invalid_argument("Qwen conversation requires text-only message parts"); + text += value->text; + } + messages.emplace_back(std::move(role), std::move(text)); + } + const auto rendered = qwen_apply_conversation_template(config_.chat_template_format, messages, + cfg.enable_thinking); + try { + const auto generated = generate(rendered, cfg); + internal::ConversationResult result; + result.parts.emplace_back(internal::TextPart{generated.text}); + result.token_ids = generated.token_ids; + result.setup_ms = generated.setup_ms; + result.prefill_ms = generated.prefill_ms; + result.decode_ms = generated.decode_ms; + result.usage.output_tokens = generated.token_ids.size(); + return result; + } catch (const QwenGenerationCapacityError& error) { + throw std::invalid_argument(error.what()); + } +} + TextResult QwenTextGenerationPipeline::run(const internal::TextContinuationRequest& request, internal::ConfigView config) { const auto cfg = qwen_text_config(config); diff --git a/families/qwen/runtime/pipeline.h b/families/qwen/runtime/pipeline.h index 7b5c0ba55a..cc726a21ad 100644 --- a/families/qwen/runtime/pipeline.h +++ b/families/qwen/runtime/pipeline.h @@ -14,6 +14,7 @@ #include "families/qwen/runtime/inference_state.h" #include "families/qwen/runtime/sampler.h" #include "families/qwen/runtime/tokenizer.h" +#include "trtmc/internal/language.h" #include "trtmc/internal/model.h" #include "trtmc/internal/stream.h" #include "trtmc/runtime/trt_module.h" @@ -48,6 +49,7 @@ struct QwenTextGenConfig { class QwenTextGenerationPipeline final : public ITextGeneration, public internal::IModel, public internal::ITextContinuation, + public internal::ITextConversation, public internal::IStreamingTextContinuation { public: QwenTextGenerationPipeline(std::unique_ptr decoder, @@ -61,6 +63,8 @@ class QwenTextGenerationPipeline final : public ITextGeneration, int32_t default_max_new_tokens() const override { return 128; } std::vector task_bindings() override; TextResult run(const internal::TextContinuationRequest&, internal::ConfigView) override; + internal::ConversationResult run(const internal::TextConversationRequest&, + internal::ConfigView) override; std::unique_ptr start(const internal::TextContinuationRequest&, internal::ConfigView) override; diff --git a/families/qwen/tests/cpp/test_qwen_chat_templates.cpp b/families/qwen/tests/cpp/test_qwen_chat_templates.cpp index c1d7d26e60..3f53a772cd 100644 --- a/families/qwen/tests/cpp/test_qwen_chat_templates.cpp +++ b/families/qwen/tests/cpp/test_qwen_chat_templates.cpp @@ -6,6 +6,7 @@ #include "families/qwen/runtime/chat_templates.h" #include +#include #include namespace { @@ -62,7 +63,46 @@ int main() { check_render("{{ messages[0]['content'] }}", false, "Q", "unsupported template preserves the prompt"); + const std::vector> history{ + {"system", " system\n"}, + {"user", "Remember ORCHID."}, + {"assistant", "\nreasoning\n\n\nOK"}, + {"user", "What word?"}}; + const std::string history_prefix = "<|im_start|>system\n system\n<|im_end|>\n" + "<|im_start|>user\nRemember ORCHID.<|im_end|>\n"; + const std::string history_suffix = + "<|im_end|>\n<|im_start|>user\nWhat word?<|im_end|>\n<|im_start|>assistant\n"; + if (trtmc::qwen_apply_conversation_template("chatml_thinking", history, false) != + history_prefix + "<|im_start|>assistant\nOK" + history_suffix + empty_think) { + std::fprintf( + stderr, + "FAIL: hybrid conversation retains roles/history and removes prior reasoning\n"); + ++failures; + } + if (trtmc::qwen_apply_conversation_template("chatml", history, false) != + history_prefix + "<|im_start|>assistant\n\nreasoning\n\n\nOK" + + history_suffix) { + std::fprintf(stderr, "FAIL: ordinary ChatML preserves assistant content\n"); + ++failures; + } + for (const auto& invalid : std::vector>>{ + {}, + {{"assistant", "unfinished"}}, + {{"tool", "unsupported"}, {"user", "Q"}}, + {{"user", "Q"}, {"system", "misplaced"}, {"user", "Q"}}}) { + bool rejected = false; + try { + trtmc::qwen_apply_conversation_template("chatml_thinking", invalid, false); + } catch (const std::invalid_argument&) { + rejected = true; + } + if (!rejected) { + std::fprintf(stderr, "FAIL: invalid conversation accepted\n"); + ++failures; + } + } + if (failures == 0) - std::fprintf(stderr, "All 13 Qwen chat template cases passed.\n"); + std::fprintf(stderr, "All Qwen single-turn and conversation template cases passed.\n"); return failures; } diff --git a/website/docs/user-guides/profile-text-with-aiperf.md b/website/docs/user-guides/profile-text-with-aiperf.md index c11629123c..dd4347aca6 100644 --- a/website/docs/user-guides/profile-text-with-aiperf.md +++ b/website/docs/user-guides/profile-text-with-aiperf.md @@ -137,3 +137,30 @@ For a reproducible selection of real ShareGPT prompts, the optional provides a pinned Qwen3-0.6B workload. It preserves selected strings, records over-limit rejections, and verifies HTTP input/output against native dispatch records. It uses the same published AIPerf package. + +## Benchmark a conversation + +With the current Qwen server and runtime, save this as `conversations.jsonl`: + +```json +{"session_id":"example","turns":[{"text":"Remember the code word ORCHID. Reply OK."},{"text":"What code word did I give you?","delay":50}]} +``` + +Use the benchmark command above with `--endpoint-type chat`, replace the two +synthetic-input options with +`--input-file conversations.jsonl --custom-dataset-type multi_turn`, and use +`--extra-inputs temperature:0 enable_thinking:false`. Each turn is one request; +`--request-count 2` measures one complete conversation. The delay is 50 +milliseconds after the first response. AIPerf includes the original user turn +and the actual model answer in the follow-up request. + +Keep conversation requests nonstreaming. History requires the loaded model's +native `TextConversation` capability; unsupported models return an explicit +error. Qwen owns chat formatting and tokenization. Its existing bundles work +with the updated family library; follow the existing build guide for the +server/runtime installation. The complete history and requested output must +fit the bundle's context limit; history is never silently dropped. + +For pinned ShareGPT conversations, per-turn timings, and automatic history +verification, use the optional +[conversation profile](https://github.com/NVIDIA/TensorRT-Model-Connect/blob/main/apps/aiperf_qual/README.md#pinned-conversations). diff --git a/website/docs/user-guides/serve-text-generation.md b/website/docs/user-guides/serve-text-generation.md index 56926a478f..8001476937 100644 --- a/website/docs/user-guides/serve-text-generation.md +++ b/website/docs/user-guides/serve-text-generation.md @@ -96,7 +96,7 @@ support that placement. | Route | Supported behavior | | --- | --- | | `POST /v1/completions` | One string `prompt`; `model` is required. | -| `POST /v1/chat/completions` | One text `user` message and optional preceding `system` message; vision-language bundles also accept one user message with text and one embedded image. | +| `POST /v1/chat/completions` | One text `user` message and optional preceding `system` message. Native `TextConversation` models also accept nonstreaming text history (up to 64 messages, final role `user`). Vision-language bundles accept one user message with text and one embedded image. | | `GET /v1/models` | Configured models and their native capabilities. | | `GET /health/live` | Control-plane process liveness. | | `GET /health/ready` | Worker readiness and admission state. | @@ -107,9 +107,15 @@ Generation accepts `temperature`, `top_p`, `min_p`, `top_k`, `seed`, Completions accept `max_tokens`; chat accepts either `max_tokens` or `max_completion_tokens`. The MVP requires `n=1`. It rejects unknown fields, prompt arrays, unsupported media content, -multi-turn chat, stop sequences, tools, and log probabilities instead +unsupported history, stop sequences, tools, and log probabilities instead of silently ignoring them. +For text history, the current Qwen family implements the existing +`TextConversation` Task for its supported ChatML templates. Send user/assistant +messages in their original order, optionally preceded by a system message, +and end with a user message. History requests are nonstreaming and must fit +the complete context limit; unsupported models return an explicit error. + For single-image chat, see [Profile Image Understanding with AIPerf](./profile-vision-with-aiperf.md). Image requests require a vision-language bundle and accept one embedded RGB PNG/JPEG plus text. System prompts, streaming, and thinking controls are rejected for this workload. @@ -160,6 +166,8 @@ composition, and validation remain inside the family selected by the bundle. For migrated bundles, the native worker calls the public header-only C++ Task wrapper over the stable C ABI. It uses `TextContinuation`, or the primary `ConditionalTextGeneration` / `TextTranslation` contract for those bundles. +History calls the existing `TextConversation` contract only when the loaded +model advertises it. The family formats and tokenizes the typed messages. Translation uses the family's bundled language defaults; this HTTP protocol does not add source/target-language parameters. From 9141cac416c16ce4e491835a20822e7ff69a3428 Mon Sep 17 00:00:00 2001 From: Yizhuo Zhang Date: Sun, 18 Oct 2026 16:02:46 -0700 Subject: [PATCH 2/3] fix(aiperf): count and correlate actual conversation inputs Count explicit encoded token IDs from the rendered pinned template and group live turns by their correlation ID. Preserve the upstream request ordinal without using it as a conversation identity. Signed-off-by: Yizhuo Zhang --- .../tests/test_text_conversations.py | 17 +++++++++++++++-- .../trtmc_aiperf_qual/text_conversations.py | 17 +++++++++++------ 2 files changed, 26 insertions(+), 8 deletions(-) diff --git a/apps/aiperf_qual/tests/test_text_conversations.py b/apps/aiperf_qual/tests/test_text_conversations.py index 3aee502bc6..0b8a99fe1f 100644 --- a/apps/aiperf_qual/tests/test_text_conversations.py +++ b/apps/aiperf_qual/tests/test_text_conversations.py @@ -7,7 +7,7 @@ import pytest from trtmc_aiperf_qual.config import ConfigError -from trtmc_aiperf_qual.text_conversations import fingerprint, select, verify_records +from trtmc_aiperf_qual.text_conversations import fingerprint, select, template_tokens, verify_records from trtmc_aiperf_qual.text_profile import arguments, cases @@ -20,6 +20,19 @@ def count_messages(messages): return sum(len(message["content"]) for message in messages) + 10 +def test_context_budget_counts_token_ids_instead_of_tokenizer_container_fields(): + class Tokenizer: + def apply_chat_template(self, messages, **options): + assert options == {"tokenize": False, "add_generation_prompt": True, "enable_thinking": False} + return "rendered original history" + + def encode(self, text, *, add_special_tokens): + assert text == "rendered original history" and add_special_tokens is False + return [100, 200, 300, 400] + + assert template_tokens(Tokenizer(), []) == 4 + + def test_original_user_turns_survive_selection_and_rejections_are_visible(): original = " first\n\u2028second " row = conversation(original, "Follow up?") @@ -50,7 +63,7 @@ def capture(): history = [*original, {"role": "assistant", "content": "OK"}, {"role": "user", "content": texts[1]}] raw, joined = [], [] for turn, (messages, text) in enumerate([(original, "OK"), (history, "ORCHID")]): - metadata = {"x_request_id": str(turn), "conversation_id": "sharegpt-0", "session_num": 0, + metadata = {"x_request_id": str(turn), "conversation_id": "sharegpt-0", "session_num": turn, "x_correlation_id": "conversation", "turn_index": turn, "benchmark_phase": "profiling", "request_start_ns": 1_000_000_000 + turn * 150_000_000, "request_end_ns": 1_100_000_000 + turn * 150_000_000} diff --git a/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py b/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py index c4c5d86365..25933cce39 100644 --- a/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py +++ b/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py @@ -17,6 +17,12 @@ def fingerprint(value) -> str: return hashlib.sha256(json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()).hexdigest() +def template_tokens(tokenizer, messages) -> int: + rendered = tokenizer.apply_chat_template(messages, tokenize=False, + add_generation_prompt=True, enable_thinking=False) + return len(tokenizer.encode(rendered, add_special_tokens=False)) + + def select(records, count, capacity, output, count_messages): if any(type(value) is not int or value < 1 for value in (count, capacity, output)) or output >= capacity: raise ConfigError("conversation count/context/output must be positive integers fitting the bundle") @@ -54,8 +60,7 @@ def prepare(config, capacity: int, output: int, out: Path): if type(delay) not in (int, float) or not math.isfinite(delay) or delay < 0: raise ConfigError("think_time_ms must be a finite nonnegative number") def count_messages(messages): - return len(tokenizer.apply_chat_template( - messages, tokenize=True, add_generation_prompt=True, enable_thinking=False)) + return template_tokens(tokenizer, messages) selected, ledger = select(records, spec["count"], capacity, output, count_messages) path = out / "inputs.jsonl" path.write_text("".join(json.dumps({"session_id": row["session_id"], "turns": [ @@ -98,8 +103,7 @@ def verify(run, joined, corpus): from transformers import AutoTokenizer tokenizer = AutoTokenizer.from_pretrained(corpus["tokenizer"]["name"], revision=corpus["tokenizer"]["revision"], trust_remote_code=False) - return verify_records(run, joined, corpus, lambda messages: len(tokenizer.apply_chat_template( - messages, tokenize=True, add_generation_prompt=True, enable_thinking=False))) + return verify_records(run, joined, corpus, lambda messages: template_tokens(tokenizer, messages)) def verify_records(run, joined, corpus, count_messages): @@ -110,17 +114,18 @@ def verify_records(run, joined, corpus, count_messages): for raw in raw_records: meta = raw.get("metadata", {}) if (not isinstance(meta, dict) or not isinstance(meta.get("x_request_id"), str) or + not isinstance(meta.get("x_correlation_id"), str) or not meta["x_correlation_id"] or not isinstance(meta.get("conversation_id"), str) or type(meta.get("session_num")) is not int or type(meta.get("turn_index")) is not int or meta["turn_index"] not in (0, 1) or any(type(meta.get(name)) is not int or meta[name] < 1 for name in ("request_start_ns", "request_end_ns"))): raise ConfigError("conversation request metadata is missing identity or timing") - for raw in sorted(raw_records, key=lambda r: (r["metadata"]["session_num"], r["metadata"]["turn_index"])): + for raw in sorted(raw_records, key=lambda r: r["metadata"]["request_start_ns"]): meta = raw["metadata"] request_id, session, turn = meta.get("x_request_id"), meta.get("conversation_id"), meta.get("turn_index") if request_id in seen or session not in expected or type(turn) is not int or turn not in (0, 1): raise ConfigError("unknown or repeated conversation request identity") seen.add(request_id) - key = (meta.get("benchmark_phase"), meta["session_num"]) + key = (meta.get("benchmark_phase"), meta["x_correlation_id"]) previous = groups.setdefault(key, []) if len(previous) != turn or (previous and previous[0]["session"] != session): raise ConfigError("conversation turns are missing, repeated or out of order") From 8187eb46d2f2959c617d20805dcfba4110e3126e Mon Sep 17 00:00:00 2001 From: Yizhuo Zhang Date: Sun, 18 Oct 2026 17:16:07 -0700 Subject: [PATCH 3/3] fix(aiperf): reject incomplete measured conversations Signed-off-by: Yizhuo Zhang --- .../tests/test_text_conversations.py | 17 +++++++++++++++++ .../trtmc_aiperf_qual/text_conversations.py | 3 ++- 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/apps/aiperf_qual/tests/test_text_conversations.py b/apps/aiperf_qual/tests/test_text_conversations.py index 0b8a99fe1f..da1946752c 100644 --- a/apps/aiperf_qual/tests/test_text_conversations.py +++ b/apps/aiperf_qual/tests/test_text_conversations.py @@ -1,5 +1,6 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import copy import hashlib import json from types import SimpleNamespace @@ -86,6 +87,22 @@ def test_live_assistant_history_and_per_turn_identity_are_verified(capture): assert result["requests"][1]["observed_gap_ms"] == 50 +@pytest.mark.parametrize("phase", ["profiling", "warmup"]) +def test_incomplete_replayed_profiling_group_is_rejected_even_after_full_coverage(capture, phase): + run, joined, corpus, raw = capture + replay = copy.deepcopy(raw[0]) + replay["metadata"].update(x_request_id="replay-first-only", x_correlation_id="another-instance", + session_num=2, benchmark_phase=phase, request_start_ns=1_400_000_000, request_end_ns=1_500_000_000) + raw.append(replay) + joined.append({"request_id": "replay-first-only", "server": copy.deepcopy(joined[0]["server"])}) + if phase == "profiling": + with pytest.raises(ConfigError, match="both turns"): + verify_records(run, joined, corpus, count_messages) + else: + # Warmup request limits may intentionally end before a conversation does. + assert verify_records(run, joined, corpus, count_messages)["verified"] + + @pytest.mark.parametrize("change", ["drop_history", "role", "original_text", "assistant_text", "native_history", "capacity"]) def test_missing_or_changed_history_cannot_be_reported_as_qualified(capture, change): run, joined, corpus, raw = capture diff --git a/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py b/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py index 25933cce39..d2be5851d1 100644 --- a/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py +++ b/apps/aiperf_qual/trtmc_aiperf_qual/text_conversations.py @@ -158,6 +158,7 @@ def verify_records(run, joined, corpus, count_messages): "correlation_id": meta.get("x_correlation_id"), "input_tokens": tokens, "messages_sha256": fingerprint(messages), "completion_sha256": server["completion_sha256"], "configured_think_time_ms": corpus["think_time_ms"] if turn else 0, "observed_gap_ms": gap}) - if measured != {(session, turn) for session in expected for turn in (0, 1)}: + if (any(len(group) != 2 for (phase, _), group in groups.items() if phase == "profiling") or + measured != {(session, turn) for session in expected for turn in (0, 1)}): raise ConfigError("profiling did not cover both turns of every selected conversation") return {"verified": True, "measured_conversations": len(expected), "requests": identities}