Bind a memory capability to one scope and produce its turn records.
as_memory_scope
as_memory_scope(scope: Any) -> MemoryScope
Accept a memory or platform scope without trusting a partial conversion.
Source code in src/symfonic/capabilities/memory/binding.py
| def as_memory_scope(scope: Any) -> MemoryScope:
"""Accept a memory or platform scope without trusting a partial conversion."""
if isinstance(scope, MemoryScope):
return scope
converted = getattr(scope, "to_memory_scope", None)
if callable(converted):
result = converted()
if isinstance(result, MemoryScope):
return result
segments = getattr(scope, "segments", None)
if segments:
ids = [str(segment) for segment in segments if segment]
if ids:
return MemoryScope(*ids[:3])
levels = getattr(scope, "path", None)
if levels:
ids = [str(level.id) for level in levels if getattr(level, "id", None)]
if ids:
return MemoryScope(*ids[:3])
raise TypeError(
f"{type(scope).__name__} is neither a MemoryScope nor convertible to "
"one, so the capability cannot be bound to a scope"
)
|
producer_for
producer_for(scope: MemoryScope) -> Any
Produce the exchange records for the capability's bound scope.
Source code in src/symfonic/capabilities/memory/binding.py
| def producer_for(scope: MemoryScope) -> Any:
"""Produce the exchange records for the capability's bound scope."""
def records_from(context: Any) -> tuple[Any, ...]:
request = getattr(context, "request", None)
if request is None: # pragma: no cover - defensive
return ()
turn_scope = getattr(request, "scope", None)
return turn_records(
query=str(getattr(request, "prompt", "") or ""),
answer=str(getattr(getattr(context, "turn", None), "text", "") or ""),
scope=as_memory_scope(turn_scope) if turn_scope is not None else scope,
# The write stage runs once per model round; deriving the id from
# the run keeps a tool-using turn from filing one exchange twice.
turn_id=str(getattr(request, "run_id", "") or "") or None,
extra_metadata={"session_id": str(getattr(request, "session_id", "") or "")},
)
return records_from
|