admin
d703ab64e3
CI / docs-check (push) Canceled after 0s
CI / backend-test (push) Canceled after 0s
CI / service-test (push) Canceled after 0s
CI / frontend-test (push) Canceled after 0s
CI / rust-core (push) Canceled after 0s
CI / docs-check (pull_request) Canceled after 0s
CI / backend-test (pull_request) Canceled after 0s
CI / service-test (pull_request) Canceled after 0s
CI / frontend-test (pull_request) Canceled after 0s
CI / rust-core (pull_request) Canceled after 0s
22 lines
551 B
Python
22 lines
551 B
Python
"""Task-搜索实际使用的嵌入路径的局部观察。"""
|
|
from contextlib import contextmanager
|
|
from contextvars import ContextVar
|
|
|
|
_observation: ContextVar[dict | None] = ContextVar("embedding_observation", default=None)
|
|
|
|
|
|
@contextmanager
|
|
def capture_embedding():
|
|
result = {"source": "not_used"}
|
|
token = _observation.set(result)
|
|
try:
|
|
yield result
|
|
finally:
|
|
_observation.reset(token)
|
|
|
|
|
|
def record_embedding(**fields) -> None:
|
|
result = _observation.get()
|
|
if result is not None:
|
|
result.update(fields)
|