This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7798-fcfd112fbc8fd8c3aa59d77003debc881da0392f in repository https://gitbox.apache.org/repos/asf/texera.git
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 {
