Spaces:

dev-strender
/

proofread-demo

Sleeping

proofread-demo / solar_eval /pipelines /steps /completion.py

Replace v24-era demo with v34 pipeline demo (engine-vendored bundle)

9c84f9d verified about 2 months ago

3.74 kB

"""`llm_completion` 스텝 -- input_fields 로 유저 메시지를 만들어 한 번 호출.
`single_step.py`(SingleStepPipeline) 에서 로직 이동 (2026-08-10, 동작 동일).
번역/요약처럼 "입력 필드들 → 완성문 하나" 형태의 태스크가 쓴다.
`pipeline: single_step` 선언은 이 스텝 하나짜리 파이프라인으로 해석된다.
"""

import logging
from typing import Any

from solar_eval.models.sample import EvalSample
from solar_eval.pipelines.pipeline import BaseStep, StepContext
from solar_eval.pipelines.registry import register_step

logger = logging.getLogger(__name__)


@register_step("llm_completion")
class LLMCompletionStep(BaseStep):
    """Single LLM call step. Used for translation and summarization."""

    async def run(self, sample: EvalSample, ctx: StepContext) -> None:
        input_data = self._input_fields_dict(sample.input)

        if ctx.messages:
            # Structured messages mode: substitute {{field}} placeholders
            resolved = self._resolve_messages(ctx.messages, input_data)
        else:
            # Legacy mode: system prompt + auto-built user message
            user_content = self._build_user_message(input_data)
            resolved = [
                {"role": "system", "content": ctx.system_prompt},
                {"role": "user", "content": user_content},
            ]

        result = await ctx.provider.chat_completion(
            messages=resolved,
            model=self.config.get("model", ctx.model),
            temperature=self.config.get("temperature", ctx.temperature),
            max_tokens=self.config.get("max_tokens", ctx.max_tokens),
            reasoning_effort=self.config.get("reasoning_effort", ctx.reasoning_effort),
        )

        ctx.text = result["content"]
        ctx.add_usage(result.get("usage", {}))
        sample.artifacts["think_content"] = result.get("think_content", "")
        sample.artifacts["finish_reason"] = result.get("finish_reason", "stop")

    def _input_fields_dict(self, sample_input: Any) -> dict[str, Any]:
        """`sample.input` 을 {필드명: 값} dict 로 정규화한다.
        러너는 `task_config`의 `input_fields` 전체로 만든 dict 를 그대로
        `sample.input` 에 담아 넘긴다(단일 필드든 다중 필드든). 여기서는
        그 관례를 그대로 받되, 다른 shape 이 들어오면 조용히 무시하지 않고
        바로 실패시킨다.
        """
        if sample_input is None:
            return {}
        if isinstance(sample_input, dict):
            return sample_input
        raise TypeError(
            "LLMCompletionStep expects sample.input to be a dict of "
            f"{{field: value}} (runner.py convention); got {type(sample_input)}"
        )

    def _resolve_messages(
        self, messages: list[dict[str, str]], input_data: dict[str, Any]
    ) -> list[dict[str, str]]:
        """Substitute {{field_name}} placeholders in message content."""
        resolved = []
        for msg in messages:
            content = msg["content"]
            for field, value in input_data.items():
                content = content.replace(f"{{{{{field}}}}}", str(value))
            resolved.append({"role": msg["role"], "content": content})
        return resolved

    def _build_user_message(self, input_data: dict[str, Any]) -> str:
        input_fields = self.env.input_fields
        if len(input_fields) == 1:
            return str(input_data.get(input_fields[0], ""))
        # Multiple fields: format as labeled sections
        parts = []
        for field in input_fields:
            value = input_data.get(field, "")
            parts.append(f"{field}:\n{value}")
        return "\n\n".join(parts)

	"""`llm_completion` 스텝 -- input_fields 로 유저 메시지를 만들어 한 번 호출.

	`single_step.py`(SingleStepPipeline) 에서 로직 이동 (2026-08-10, 동작 동일).
	번역/요약처럼 "입력 필드들 → 완성문 하나" 형태의 태스크가 쓴다.
	`pipeline: single_step` 선언은 이 스텝 하나짜리 파이프라인으로 해석된다.
	"""

	import logging
	from typing import Any

	from solar_eval.models.sample import EvalSample
	from solar_eval.pipelines.pipeline import BaseStep, StepContext
	from solar_eval.pipelines.registry import register_step

	logger = logging.getLogger(__name__)


	@register_step("llm_completion")
	class LLMCompletionStep(BaseStep):
	"""Single LLM call step. Used for translation and summarization."""

	async def run(self, sample: EvalSample, ctx: StepContext) -> None:
	input_data = self._input_fields_dict(sample.input)

	if ctx.messages:
	# Structured messages mode: substitute {{field}} placeholders
	resolved = self._resolve_messages(ctx.messages, input_data)
	else:
	# Legacy mode: system prompt + auto-built user message
	user_content = self._build_user_message(input_data)
	resolved = [
	{"role": "system", "content": ctx.system_prompt},
	{"role": "user", "content": user_content},
	]

	result = await ctx.provider.chat_completion(
	messages=resolved,
	model=self.config.get("model", ctx.model),
	temperature=self.config.get("temperature", ctx.temperature),
	max_tokens=self.config.get("max_tokens", ctx.max_tokens),
	reasoning_effort=self.config.get("reasoning_effort", ctx.reasoning_effort),
	)

	ctx.text = result["content"]
	ctx.add_usage(result.get("usage", {}))
	sample.artifacts["think_content"] = result.get("think_content", "")
	sample.artifacts["finish_reason"] = result.get("finish_reason", "stop")

	def _input_fields_dict(self, sample_input: Any) -> dict[str, Any]:
	"""`sample.input` 을 {필드명: 값} dict 로 정규화한다.

	러너는 `task_config`의 `input_fields` 전체로 만든 dict 를 그대로
	`sample.input` 에 담아 넘긴다(단일 필드든 다중 필드든). 여기서는
	그 관례를 그대로 받되, 다른 shape 이 들어오면 조용히 무시하지 않고
	바로 실패시킨다.
	"""
	if sample_input is None:
	return {}
	if isinstance(sample_input, dict):
	return sample_input
	raise TypeError(
	"LLMCompletionStep expects sample.input to be a dict of "
	f"{{field: value}} (runner.py convention); got {type(sample_input)}"
	)

	def _resolve_messages(
	self, messages: list[dict[str, str]], input_data: dict[str, Any]
	) -> list[dict[str, str]]:
	"""Substitute {{field_name}} placeholders in message content."""
	resolved = []
	for msg in messages:
	content = msg["content"]
	for field, value in input_data.items():
	content = content.replace(f"{{{{{field}}}}}", str(value))
	resolved.append({"role": msg["role"], "content": content})
	return resolved

	def _build_user_message(self, input_data: dict[str, Any]) -> str:
	input_fields = self.env.input_fields
	if len(input_fields) == 1:
	return str(input_data.get(input_fields[0], ""))
	# Multiple fields: format as labeled sections
	parts = []
	for field in input_fields:
	value = input_data.get(field, "")
	parts.append(f"{field}:\n{value}")
	return "\n\n".join(parts)