This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 2113455e3f fix(workflow-operator): carry structured task context into
chat fallback payloads (#7798)
2113455e3f is described below
commit 2113455e3f384912d721893147e3dfd329df1126
Author: Prateek Ganigi <[email protected]>
AuthorDate: Mon Aug 24 00:35:45 2026 +0000
fix(workflow-operator): carry structured task context into chat fallback
payloads (#7798)
### What changes were proposed in this PR?
The five structured Hugging Face tasks (`question-answering`,
`table-question-answering`, `zero-shot-classification`,
`sentence-similarity`, `text-ranking`) build a native HF pipeline
payload that only `hf-inference` understands. When the operator fell
back to a third-party chat-completions provider, all three chat branches
sent just the prompt cell, dropping the
context, table, candidate labels, or sentence list, so the model
answered without the input that defines the task.
This adds a `_chat_content_for_task` helper to the generated operator
that reformulates each of the five tasks into a chat prompt carrying its
full context, and routes all three chat branches (zai-org,
OpenAI-compatible, unknown-provider fallback) through it. Other tasks
pass through unchanged. `QaRankingCodegen.parsePython` is also extended
to read `choices[0].message.content` from chat responses, keeping the
native `{"answer": ...}` shape as the primary path, the same idiom
`ImageTaskCodegen` already uses.
### Any related issues?
Closes #7195
### How was this PR tested?
131 tests pass in the `WorkflowOperator` Hugging Face suites,
`PythonCodeRawInvalidTextSpec` py-compiles the generated Python for all
117 operators, and `scalafmtCheck` is clean for main and test sources. A
new spec test asserts the helper is emitted and used by every chat
branch, with one assertion per task reformulation; two existing tests
were updated where they pinned the old parse expression.
### Was this PR authored or co-authored using generative AI tooling?
Yes, this PR was co-authored with Claude in compliance with ASF policy.
---
.../codegen/HuggingFaceCodegenBase.scala | 73 +++++++++++++++++++++-
.../huggingFace/codegen/QaRankingCodegen.scala | 16 ++++-
.../HuggingFaceInferenceOpDescSpec.scala | 28 +++++++--
.../huggingFace/codegen/QaRankingCodegenSpec.scala | 5 +-
4 files changed, 110 insertions(+), 12 deletions(-)
diff --git
a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/codegen/HuggingFaceCodegenBase.scala
b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/codegen/HuggingFaceCodegenBase.scala
index 043e713363..a40d4aeac7 100644
---
a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/codegen/HuggingFaceCodegenBase.scala
+++
b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/codegen/HuggingFaceCodegenBase.scala
@@ -260,6 +260,73 @@ object HuggingFaceCodegenBase {
| summary = "; ".join(errors) if errors else "no providers
available"
| return last_resp, summary
|
+ | def _chat_content_for_task(self, pipeline_payload, prompt_value):
+ | '''Reformulate a structured task (question-answering,
+ | table-question-answering, zero-shot-classification,
+ | sentence-similarity, text-ranking) as a chat prompt so
third-party
+ | chat-completions providers receive the full context. The native
+ | pipeline_payload (question+context, table, labels, sentences)
is
+ | only understood by hf-inference; chat providers otherwise get
only
+ | prompt_value and hallucinate. Non-structured tasks are
unchanged.
+ | '''
+ | task = self.TASK
+ | inputs = pipeline_payload.get("inputs") if
isinstance(pipeline_payload, dict) else None
+ | if task == "question-answering" and isinstance(inputs, dict):
+ | question = inputs.get("question", prompt_value)
+ | context = inputs.get("context", "")
+ | if context:
+ | return (
+ | "Answer the question using only the context below.
"
+ | "If the answer is not in the context, say so.\n\n"
+ | f"Context:\n{context}\n\nQuestion: {question}"
+ | )
+ | return question or prompt_value
+ | if task == "table-question-answering" and isinstance(inputs,
dict):
+ | query = inputs.get("query", prompt_value)
+ | table = inputs.get("table")
+ | if table:
+ | return (
+ | "Answer the question using the table below, given
as JSON "
+ | "mapping each column to its list of cell
values.\n\n"
+ | f"Table:\n{json.dumps(table)}\n\nQuestion: {query}"
+ | )
+ | return query or prompt_value
+ | if task == "zero-shot-classification":
+ | params = pipeline_payload.get("parameters") if
isinstance(pipeline_payload, dict) else None
+ | labels = params.get("candidate_labels", []) if
isinstance(params, dict) else []
+ | text = inputs if isinstance(inputs, str) else prompt_value
+ | if labels:
+ | return (
+ | "Classify the text into exactly one of these
labels: "
+ | f"{', '.join(str(l) for l in labels)}. Respond
with only the chosen label.\n\n"
+ | f"Text: {text}"
+ | )
+ | return text or prompt_value
+ | if task == "sentence-similarity" and isinstance(inputs, dict):
+ | source = inputs.get("source_sentence", prompt_value)
+ | sentences = inputs.get("sentences") or []
+ | if sentences:
+ | numbered = "\n".join(f"{i}. {s}" for i, s in
enumerate(sentences, 1))
+ | return (
+ | "Rate how semantically similar the source sentence
is to "
+ | "each candidate below, from 0.0 (unrelated) to 1.0
"
+ | "(identical meaning). Give one score per
candidate.\n\n"
+ | f"Source: {source}\n\nCandidates:\n{numbered}"
+ | )
+ | return source or prompt_value
+ | if task == "text-ranking" and isinstance(inputs, dict):
+ | query = inputs.get("query", prompt_value)
+ | texts = inputs.get("texts") or []
+ | if texts:
+ | numbered = "\n".join(f"{i}. {t}" for i, t in
enumerate(texts, 1))
+ | return (
+ | "Rank the passages below by relevance to the
query, most "
+ | "relevant first, and return the ranking.\n\n"
+ | f"Query: {query}\n\nPassages:\n{numbered}"
+ | )
+ | return query or prompt_value
+ | return prompt_value
+ |
| def _call_provider(self, provider_name, provider_id, json_headers,
raw_binary_headers, pipeline_payload, use_raw_binary_body, prompt_value):
| '''Route to a third-party provider using its native API format.
| Handles OpenAI-compatible chat providers for text-gen,
zai-org's
@@ -297,7 +364,7 @@ object HuggingFaceCodegenBase {
| file_data = f"data:image/png;base64,{img_b64}" if
img_b64 else ""
| return requests.post(url, headers=zai_headers,
json={"model": provider_id, "file": file_data}, timeout=120)
| url = f"{base}/api/paas/v4/chat/completions"
- | messages = [{"role": "user", "content": prompt_value}]
+ | messages = [{"role": "user", "content":
self._chat_content_for_task(pipeline_payload, prompt_value)}]
| if img_b64:
| messages = [{"role": "user", "content": [
| {"type": "image_url", "image_url": {"url":
f"data:image/png;base64,{img_b64}"}},
@@ -439,7 +506,7 @@ object HuggingFaceCodegenBase {
| url = f"{base}/v1/audio/speech"
| return requests.post(url, headers=json_headers,
json={"model": provider_id, "input": prompt_value}, timeout=120)
| url = f"{base}/{self.CHAT_ROUTES.get(provider_name,
'v1/chat/completions')}"
- | messages = [{"role": "user", "content": prompt_value}]
+ | messages = [{"role": "user", "content":
self._chat_content_for_task(pipeline_payload, prompt_value)}]
| if img_b64:
| messages = [{"role": "user", "content": [
| {"type": "image_url", "image_url": {"url":
f"data:image/png;base64,{img_b64}"}},
@@ -460,7 +527,7 @@ object HuggingFaceCodegenBase {
| resp = requests.post(url, headers=json_headers,
json=pipeline_payload, timeout=120)
| if resp.status_code in (400, 404, 422):
| url = f"{base}/v1/chat/completions"
- | messages = [{"role": "user", "content": prompt_value}]
+ | messages = [{"role": "user", "content":
self._chat_content_for_task(pipeline_payload, prompt_value)}]
| if img_b64:
| messages = [{"role": "user", "content": [
| {"type": "image_url", "image_url": {"url":
f"data:image/png;base64,{img_b64}"}},
diff --git
a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/codegen/QaRankingCodegen.scala
b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/codegen/QaRankingCodegen.scala
index 18c6897499..bdc266080a 100644
---
a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/codegen/QaRankingCodegen.scala
+++
b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/huggingFace/codegen/QaRankingCodegen.scala
@@ -76,9 +76,21 @@ object QaRankingCodegen extends TaskCodegen {
override def parsePython(ctx: CodegenContext): String =
""" if task == "question-answering":
- | return body.get("answer", json.dumps(body)) if
isinstance(body, dict) else json.dumps(body)
+ | if isinstance(body, dict):
+ | # Third-party chat providers answer via
choices[0].message;
+ | # hf-inference returns the native {"answer": ...}
shape.
+ | if "choices" in body:
+ | return body["choices"][0]["message"]["content"]
+ | return body.get("answer", json.dumps(body))
+ | return json.dumps(body)
| elif task == "table-question-answering":
- | return body.get("answer", json.dumps(body)) if
isinstance(body, dict) else json.dumps(body)
+ | if isinstance(body, dict):
+ | if "choices" in body:
+ | return body["choices"][0]["message"]["content"]
+ | return body.get("answer", json.dumps(body))
+ | return json.dumps(body)
| elif task in ("zero-shot-classification",
"sentence-similarity", "text-ranking"):
+ | if isinstance(body, dict) and "choices" in body:
+ | return body["choices"][0]["message"]["content"]
| return json.dumps(body)""".stripMargin
}
diff --git
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceInferenceOpDescSpec.scala
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceInferenceOpDescSpec.scala
index e23a9141fd..56b4b65ebf 100644
---
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceInferenceOpDescSpec.scala
+++
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/HuggingFaceInferenceOpDescSpec.scala
@@ -567,9 +567,8 @@ class HuggingFaceInferenceOpDescSpec extends AnyFlatSpec
with Matchers {
code should include("ctx_col = self.CONTEXT_COLUMN")
code should include("Context column")
code should include("""payload = {"inputs": {"question": prompt_value,
"context": ctx_val}}""")
- code should include(
- """return body.get("answer", json.dumps(body)) if isinstance(body, dict)
else json.dumps(body)"""
- )
+ code should include("""body.get("answer", json.dumps(body))""")
+ code should include("""body["choices"][0]["message"]["content"]""")
}
it should "route table-question-answering with a precomputed table payload"
in {
@@ -577,9 +576,8 @@ class HuggingFaceInferenceOpDescSpec extends AnyFlatSpec
with Matchers {
code should include("""if task == "table-question-answering":""")
code should include("table_dict = {}")
code should include("""payload = {"inputs": {"query": prompt_value,
"table": table_dict}}""")
- code should include(
- """return body.get("answer", json.dumps(body)) if isinstance(body, dict)
else json.dumps(body)"""
- )
+ code should include("""body.get("answer", json.dumps(body))""")
+ code should include("""body["choices"][0]["message"]["content"]""")
}
it should "route zero-shot-classification with candidate labels" in {
@@ -630,6 +628,24 @@ class HuggingFaceInferenceOpDescSpec extends AnyFlatSpec
with Matchers {
}
}
+ it should
+ "reformulate structured tasks into a chat message on fallback providers
(#7195)" in {
+ // On third-party chat providers the structured pipeline_payload (context /
+ // table / labels / sentences) is inlined into the chat message via the
+ // helper, instead of sending only prompt_value and dropping the rest.
+ val code = makeDesc(task = "question-answering", contextColumn =
"context").generatePythonCode()
+ code should include("def _chat_content_for_task(")
+ // Every chat branch (zai-org, OpenAI-compatible, unknown-fallback) uses
it.
+ code should include("self._chat_content_for_task(pipeline_payload,
prompt_value)")
+ code should not include ("""messages = [{"role": "user", "content":
prompt_value}]""")
+ // Per-task reformulations are present.
+ code should include("Answer the question using only the context below.")
+ code should include("Answer the question using the table below")
+ code should include("Classify the text into exactly one of these labels:")
+ code should include("Rate how semantically similar the source sentence")
+ code should include("Rank the passages below by relevance to the query")
+ }
+
"getOutputSchemas" should "add the result column as a STRING to the
inherited schema" in {
val desc = makeDesc(resultColumn = "answer")
val inputSchema = Schema().add("prompt", AttributeType.STRING)
diff --git
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/codegen/QaRankingCodegenSpec.scala
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/codegen/QaRankingCodegenSpec.scala
index 0e21824b88..c7c3c18a91 100644
---
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/codegen/QaRankingCodegenSpec.scala
+++
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/huggingFace/codegen/QaRankingCodegenSpec.scala
@@ -110,11 +110,14 @@ class QaRankingCodegenSpec extends AnyFlatSpec with
Matchers {
out should include("""payload = {"inputs": prompt_value}""")
}
- "QaRankingCodegen.parsePython" should "extract the answer field for both QA
variants" in {
+ "QaRankingCodegen.parsePython" should "extract the answer, including from
chat-provider responses" in {
val out = QaRankingCodegen.parsePython(makeCtx())
out should include("""if task == "question-answering":""")
out should include("""elif task == "table-question-answering":""")
out should include("""body.get("answer"""")
+ // #7195: chat-completions responses (third-party providers) are read from
+ // choices[0].message.content, not the native {"answer": ...} shape.
+ out should include("""body["choices"][0]["message"]["content"]""")
}
it should "return the raw JSON body for the ranking-style tasks" in {
