Skip to content

symfonic.evals.targets

targets

Drive a compiled Agent as an in-process evaluation target.

The started-host and deployed-HTTP targets live next door in :mod:symfonic.evals.host_target and :mod:symfonic.evals.http_target. The line between them is real: this target holds the compiled object and can report what it folded, which is what optional packs read.

AgentTarget

AgentTarget(agent: Any, *, close_agent: bool = True, sessions: dict[str, str] | None = None, resume: ResumeSeam | None = None, evidence: TargetEvidenceAdapter | None = None, evidence_scope: Any = 'default', resources: Any = None)

Drive a public Agent and retain one transcript per conversation.

Parameters:

Name Type Description Default
resume ResumeSeam | None

the deployment's own public redemption operation, taking the person's answer. Optional and never invented: a compiled Agent pauses but does not redeem, so a target that was given no seam publishes no resume operation and every pack that needs one resolves to not-applicable by name.

None
Source code in src/symfonic/evals/targets.py
def __init__(
    self,
    agent: Any,
    *,
    close_agent: bool = True,
    sessions: dict[str, str] | None = None,
    resume: ResumeSeam | None = None,
    evidence: TargetEvidenceAdapter | None = None,
    evidence_scope: Any = "default",
    resources: Any = None,
) -> None:
    """
    Args:
        resume: the deployment's own public redemption operation, taking
            the person's answer. Optional and never invented: a compiled
            ``Agent`` pauses but does not redeem, so a target that was
            given no seam publishes no ``resume`` operation and every pack
            that needs one resolves to not-applicable by name.
    """
    if not callable(getattr(agent, "stream", None)):
        raise TypeError("AgentTarget requires an object with stream()")
    if resume is not None and not callable(resume):
        raise TypeError("the resume seam must be callable")
    self._agent = agent
    self._close_agent = close_agent
    self._histories: dict[str, tuple[Any, ...]] = {}
    self._sessions = sessions if sessions is not None else {}
    self._resume = resume
    self._evidence_adapter = evidence
    self._evidence_scope = evidence_scope
    self._resources = resources

capabilities property

capabilities: tuple[str, ...]

The names the wrapped agent folded, never a declared feature list.

operations property

operations: tuple[str, ...]

The non-turn operations this target can actually perform.

capability_evidence async

capability_evidence() -> CapabilityEvidence

Read optional-pack evidence from the compiled agent itself.

Source code in src/symfonic/evals/targets.py
async def capability_evidence(self) -> CapabilityEvidence:
    """Read optional-pack evidence from the compiled agent itself."""
    channels = () if self._evidence_adapter is None else self._evidence_adapter.channels
    traits = ()
    if self._evidence_adapter is not None:
        attest = getattr(self._evidence_adapter, "traits", None)
        if attest is not None:
            traits = attest(self._agent, self._resources, self._evidence_scope)
            if inspect.isawaitable(traits):
                traits = await traits
    return CapabilityEvidence.from_agent(
        self._agent,
        operations=self.operations,
        traits=traits,
        evidence_channels=channels,
    )

resume async

resume(*, conversation: str, scope: str, answer: Mapping[str, Any]) -> Observation

Answer one outstanding pause through the deployment's own seam.

Source code in src/symfonic/evals/targets.py
async def resume(
    self, *, conversation: str, scope: str, answer: Mapping[str, Any]
) -> Observation:
    """Answer one outstanding pause through the deployment's own seam."""
    if scope != "default":
        raise ValueError("AgentTarget cannot switch scope; use HostTarget")
    if self._resume is None:
        raise TypeError(
            "this AgentTarget was built with no resume seam, so the pause "
            "it took cannot be answered here"
        )
    del conversation  # the token identifies the paused turn, not the alias
    return resume_observation(await self._resume(answer), self._evidence())

new_session_id

new_session_id() -> str

Stay below the shipped telemetry schema's 36-character limit.

Source code in src/symfonic/evals/targets.py
def new_session_id() -> str:
    """Stay below the shipped telemetry schema's 36-character limit."""
    return f"eval-{uuid4().hex[:24]}"