wehub-resource-sync
100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
import json
|
|
|
|
from nanobot.utils import helpers
|
|
from nanobot.utils.helpers import estimate_prompt_tokens, estimate_prompt_tokens_chain
|
|
|
|
|
|
class _NoCounterProvider:
|
|
pass
|
|
|
|
|
|
class _BrokenCounterProvider:
|
|
def estimate_prompt_tokens(self, messages, tools=None, model=None):
|
|
raise RuntimeError("counter unavailable")
|
|
|
|
|
|
def test_estimate_prompt_tokens_chain_falls_back_without_provider_counter() -> None:
|
|
tokens, source = estimate_prompt_tokens_chain(
|
|
_NoCounterProvider(),
|
|
"test-model",
|
|
[{"role": "user", "content": "hello"}],
|
|
)
|
|
|
|
assert tokens > 0
|
|
assert source == "tiktoken"
|
|
|
|
|
|
def test_estimate_prompt_tokens_chain_falls_back_when_provider_counter_fails() -> None:
|
|
tokens, source = estimate_prompt_tokens_chain(
|
|
_BrokenCounterProvider(),
|
|
"test-model",
|
|
[{"role": "user", "content": "hello"}],
|
|
)
|
|
|
|
assert tokens > 0
|
|
assert source == "tiktoken"
|
|
|
|
|
|
def test_estimate_prompt_tokens_caches_tools_encoding(monkeypatch) -> None:
|
|
helpers._get_token_encoding.cache_clear()
|
|
helpers._TOOLS_TOKEN_CACHE.clear()
|
|
|
|
class FakeEncoding:
|
|
def __init__(self) -> None:
|
|
self.encoded: list[str] = []
|
|
|
|
def encode(self, text: str) -> list[int]:
|
|
self.encoded.append(text)
|
|
return list(range(max(1, len(text) // 4)))
|
|
|
|
fake_encoding = FakeEncoding()
|
|
get_encoding_calls = 0
|
|
|
|
def fake_get_encoding(name: str) -> FakeEncoding:
|
|
nonlocal get_encoding_calls
|
|
assert name == "cl100k_base"
|
|
get_encoding_calls += 1
|
|
return fake_encoding
|
|
|
|
monkeypatch.setattr(helpers.tiktoken, "get_encoding", fake_get_encoding)
|
|
tools = [{"type": "function", "function": {"name": "demo", "description": "cached"}}]
|
|
messages = [{"role": "user", "content": "hello"}]
|
|
|
|
first = estimate_prompt_tokens(messages, tools)
|
|
second = estimate_prompt_tokens(messages, tools)
|
|
|
|
assert first == second
|
|
assert get_encoding_calls == 1
|
|
rendered_tools = "\n" + json.dumps(tools, ensure_ascii=False)
|
|
assert fake_encoding.encoded.count(rendered_tools) == 1
|
|
|
|
|
|
def test_estimate_prompt_tokens_recomputes_when_tool_items_change(monkeypatch) -> None:
|
|
helpers._get_token_encoding.cache_clear()
|
|
helpers._TOOLS_TOKEN_CACHE.clear()
|
|
|
|
class FakeEncoding:
|
|
def __init__(self) -> None:
|
|
self.encoded: list[str] = []
|
|
|
|
def encode(self, text: str) -> list[int]:
|
|
self.encoded.append(text)
|
|
return list(range(max(1, len(text) // 4)))
|
|
|
|
fake_encoding = FakeEncoding()
|
|
monkeypatch.setattr(helpers.tiktoken, "get_encoding", lambda _name: fake_encoding)
|
|
|
|
tools = [{"type": "function", "function": {"name": "before"}}]
|
|
messages = [{"role": "user", "content": "hello"}]
|
|
estimate_prompt_tokens(messages, tools)
|
|
|
|
tools[0] = {"type": "function", "function": {"name": "after"}}
|
|
estimate_prompt_tokens(messages, tools)
|
|
|
|
before_tools = "\n" + json.dumps(
|
|
[{"type": "function", "function": {"name": "before"}}], ensure_ascii=False
|
|
)
|
|
after_tools = "\n" + json.dumps(tools, ensure_ascii=False)
|
|
assert before_tools in fake_encoding.encoded
|
|
assert after_tools in fake_encoding.encoded
|