Skip to content
Snippets Groups Projects
Commit 6f250c5b authored by Simonas's avatar Simonas
Browse files

small fixes

parent 5faa32c0
Branches
Tags
No related merge requests found
Expand all Hide whitespace changes
Inline Side-by-side
Showing
with 779 additions and 107 deletions
.env.example 0 → 100644
+ 1
− 0
View file @ 6f250c5b
COHERE_API_KEY=
\ No newline at end of file
.gitignore
+ 11
− 4
View file @ 6f250c5b
.env
mac.env
__pycache__
*.pyc
.venv
.DS_Store
venv/
/.vscode
**/__pycache__
**/*.py[cod]
\ No newline at end of file
**/*.py[cod]
# local env files
.env*.local
.env
mac.env
\ No newline at end of file
README.md
+ 1
− 1
View file @ 6f250c5b
# Decision Layer
\ No newline at end of file
# Semantic Router
poetry.lock
+ 673
− 12
View file @ 6f250c5b
This diff is collapsed.
pyproject.toml
+ 4
− 0
View file @ 6f250c5b
......@@ -18,6 +18,10 @@ openai = "^0.28.1"
transformers = "^4.34.1"
cohere = "^4.32"
[tool.poetry.group.dev.dependencies]
ipykernel = "^6.26.0"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
semantic_router/encoders/base.py
+ 1
− 1
View file @ 6f250c5b
......@@ -8,4 +8,4 @@ class BaseEncoder(BaseModel):
arbitrary_types_allowed = True
def __call__(self, texts: list[str]) -> list[float]:
pass
raise NotImplementedError("Subclasses must implement this method")
semantic_router/encoders/cohere.py
+ 13
− 7
View file @ 6f250c5b
import os
import cohere
from semantic_router.encoders import BaseEncoder
class CohereEncoder(BaseEncoder):
client: cohere.Client | None
def __init__(self, name: str = "embed-english-v3.0", cohere_api_key: str | None = None):
super().__init__(name=name, client=None)
def __init__(
self, name: str = "embed-english-v3.0", cohere_api_key: str | None = None
):
super().__init__(name=name)
cohere_api_key = cohere_api_key or os.getenv("COHERE_API_KEY")
if cohere_api_key is None:
raise ValueError("Cohere API key cannot be 'None'.")
self.client = cohere.Client(cohere_api_key)
def __call__(self, texts: list[str]) -> list[float]:
def __call__(self, texts: list[str]) -> list[list[float]]:
if self.client is None:
raise ValueError("Cohere client is not initialized.")
if len(texts) == 1:
input_type = "search_query"
else:
input_type = "search_document"
embeds = self.client.embed(
texts, input_type=input_type, model=self.name
)
return embeds.embeddings
\ No newline at end of file
embeds = self.client.embed(texts, input_type=input_type, model=self.name)
return embeds.embeddings
semantic_router/encoders/huggingface.py
+ 2
− 2
View file @ 6f250c5b
......@@ -3,7 +3,7 @@ from semantic_router.encoders import BaseEncoder
class HuggingFaceEncoder(BaseEncoder):
def __init__(self, name: str):
super().__init__(name)
self.name = name
def __call__(self, texts: list[str]) -> list[float]:
raise NotImplementedError
\ No newline at end of file
raise NotImplementedError
semantic_router/encoders/openai.py
+ 13
− 13
View file @ 6f250c5b
import os
from time import sleep
from semantic_router.encoders import BaseEncoder
import openai
from time import time
from openai.error import RateLimitError
from semantic_router.encoders import BaseEncoder
class OpenAIEncoder(BaseEncoder):
......@@ -16,20 +18,18 @@ class OpenAIEncoder(BaseEncoder):
"""Encode a list of texts using the OpenAI API. Returns a list of
vector embeddings.
"""
passed = False
res = None
# exponential backoff in case of RateLimitError
for j in range(5):
try:
# create embeddings
res = openai.Embedding.create(
input=texts, engine=self.name
)
passed = True
except openai.error.RateLimitError:
time.sleep(2 ** j)
if not passed:
raise openai.error.RateLimitError
res = openai.Embedding.create(input=texts, engine=self.name)
if isinstance(res, dict) and "data" in res:
break
except RateLimitError:
sleep(2**j)
if not res or not isinstance(res, dict) or "data" not in res:
raise ValueError("Failed to create embeddings.")
# get embeddings
embeds = [r["embedding"] for r in res["data"]]
return embeds
\ No newline at end of file
semantic_router/layer.py
+ 45
− 32
View file @ 6f250c5b
from semantic_router.encoders import BaseEncoder, OpenAIEncoder, CohereEncoder
from semantic_router.schema import Decision
import numpy as np
from numpy.linalg import norm
from semantic_router.encoders import BaseEncoder, CohereEncoder, OpenAIEncoder
from semantic_router.schema import Decision
class DecisionLayer:
index = None
categories = None
threshold = 0.82
similarity_threshold = 0.82
def __init__(self, encoder: BaseEncoder, decisions: list[Decision] = []):
self.encoder = encoder
# decide on default threshold based on encoder
if isinstance(encoder, OpenAIEncoder):
self.threshold = 0.82
self.similarity_threshold = 0.82
elif isinstance(encoder, CohereEncoder):
self.threshold = 0.3
self.similarity_threshold = 0.3
else:
self.threshold = 0.82
self.similarity_threshold = 0.82
# if decisions list has been passed, we initialize index now
if decisions:
# initialize index now
......@@ -26,15 +27,15 @@ class DecisionLayer:
def __call__(self, text: str) -> str | None:
results = self._query(text)
top_class, top_class_scores = self._semantic_classify(results)
passed = self._pass_threshold(top_class_scores, self.threshold)
top_class, top_class_scores = self._semantic_classify(results)
passed = self._pass_threshold(top_class_scores, self.similarity_threshold)
if passed:
return top_class
else:
return None
def add(self, decision: Decision):
self._add_decision(devision=decision)
self._add_decision(decision=decision)
def _add_decision(self, decision: Decision):
# create embeddings
......@@ -42,9 +43,9 @@ class DecisionLayer:
# create decision array
if self.categories is None:
self.categories = np.array([decision.name]*len(embeds))
self.categories = np.array([decision.name] * len(embeds))
else:
str_arr = np.array([decision.name]*len(embeds))
str_arr = np.array([decision.name] * len(embeds))
self.categories = np.concatenate([self.categories, str_arr])
# create utterance array (the index)
if self.index is None:
......@@ -53,39 +54,51 @@ class DecisionLayer:
embed_arr = np.array(embeds)
self.index = np.concatenate([self.index, embed_arr])
def _query(self, text: str, top_k: int=5):
def _query(self, text: str, top_k: int = 5):
"""Given some text, encodes and searches the index vector space to
retrieve the top_k most similar records.
"""
# create query vector
xq = np.array(self.encoder([text]))
xq = np.squeeze(xq) # Reduce to 1d array.
sim = np.dot(self.index, xq.T) / (norm(self.index, axis=1)*norm(xq.T))
# get indices of top_k records
top_k = min(top_k, sim.shape[0])
idx = np.argpartition(sim, -top_k)[-top_k:]
scores = sim[idx]
# get the utterance categories (decision names)
decisions = self.categories[idx]
return [
{"decision": d, "score": s.item()} for d, s in zip(decisions, scores)
]
xq = np.squeeze(xq) # Reduce to 1d array.
if self.index is not None:
index_norm = norm(self.index, axis=1)
xq_norm = norm(xq.T)
sim = np.dot(self.index, xq.T) / (index_norm * xq_norm)
# get indices of top_k records
top_k = min(top_k, sim.shape[0])
idx = np.argpartition(sim, -top_k)[-top_k:]
scores = sim[idx]
# get the utterance categories (decision names)
decisions = self.categories[idx] if self.categories is not None else []
return [
{"decision": d, "score": s.item()} for d, s in zip(decisions, scores)
]
else:
return []
def _semantic_classify(self, query_results: dict):
def _semantic_classify(self, query_results: list[dict]) -> tuple[str, list[float]]:
scores_by_class = {}
for result in query_results:
score = result['score']
decision = result['decision']
score = result["score"]
decision = result["decision"]
if decision in scores_by_class:
scores_by_class[decision].append(score)
else:
scores_by_class[decision] = [score]
# Calculate total score for each class
total_scores = {decision: sum(scores) for decision, scores in scores_by_class.items()}
top_class = max(total_scores, key=total_scores.get, default=None)
total_scores = {
decision: sum(scores) for decision, scores in scores_by_class.items()
}
top_class = max(total_scores, key=lambda x: total_scores[x], default=None)
# Return the top class and its associated scores
return str(top_class), scores_by_class.get(top_class, [])
def _pass_threshold(self, scores: list[float], threshold: float):
"""Returns true if the threshold has been passed."""
return max(scores) > threshold
def _pass_threshold(self, scores: list[float], threshold: float) -> bool:
if scores:
return max(scores) > threshold
else:
return False
walkthrough.ipynb
+ 15
− 35
View file @ 6f250c5b
%% Cell type:markdown id: tags:
# Semantic Router Walkthrough
%% Cell type:markdown id: tags:
The Semantic Router library can be used as a super fast decision making layer on top of LLMs. That means rather than waiting on a slow agent to decide what to do, we can use the magic of semantic vector space to make decisions. Cutting decision making time down from seconds to milliseconds.
%% Cell type:markdown id: tags:
## Getting Started
%% Cell type:markdown id: tags:
We start by installing the library:
%% Cell type:code id: tags:
``` python
!pip install -qU semantic-router==0.0.1
```
%% Cell type:markdown id: tags:
We start by defining a dictionary mapping decisions to example phrases that should trigger those decisions.
%% Cell type:code id: tags:
``` python
from semantic_router.schema import Decision
politics = Decision(
name="politics",
utterances=[
"isn't politics the best thing ever",
"why don't you tell me about your political opinions",
"don't you just love the president"
"don't you just hate the president",
"they're going to destroy this country!",
"they will save the country!"
]
)
```
%% Cell type:markdown id: tags:
Let's define another for good measure:
%% Cell type:code id: tags:
``` python
chitchat = Decision(
name="chitchat",
utterances=[
"how's the weather today?",
"how are things going?",
"lovely weather today",
"the weather is horrendous",
"let's go to the chippy"
]
)
decisions = [politics, chitchat]
```
%% Cell type:markdown id: tags:
Now we initialize our embedding model:
%% Cell type:code id: tags:
``` python
from semantic_router.encoders import CohereEncoder
from getpass import getpass
import os
os.environ["COHERE_API_KEY"] = getpass("Enter Cohere API Key: ")
# os.environ["COHERE_API_KEY"] = getpass("Enter Cohere API Key: ")
os.environ["COHERE_API_KEY"]
encoder = CohereEncoder()
```
%% Cell type:markdown id: tags:
Now we define the `DecisionLayer`. When called, the decision layer will consume text (a query) and output the category (`Decision`) it belongs to — to initialize a `DecisionLayer` we need our `encoder` model and a list of `decisions`.
%% Cell type:code id: tags:
``` python
from semantic_router import DecisionLayer
dl = DecisionLayer(encoder=encoder, decisions=decisions)
```
%% Cell type:markdown id: tags:
Now we can test it:
%% Cell type:code id: tags:
``` python
dl("don't you love politics?")
```
%% Output
'politics'
%% Cell type:code id: tags:
``` python
dl("how's the weather today?")
```
%% Output
'chitchat'
%% Cell type:markdown id: tags:
Both are classified accurately, what if we send a query that is unrelated to our existing `Decision` objects?
%% Cell type:code id: tags:
``` python
dl("I'm interested in learning about llama 2")
```
%% Cell type:markdown id: tags:
In this case, we return `None` because no matches were identified.
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment

:feet::heart: Strive to be the person your dogs believe you are :heart::feet: