File size: 4,975 Bytes
b987573 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 |
from typing import Optional, List
import dspy
import copy
import dspy.evaluate
from pydantic import BaseModel
from dotenv import load_dotenv
import os
from dspy.teleprompt import BootstrapFewShotWithRandomSearch
load_dotenv()
class Agent(dspy.Module):
"""
Base Agent Module
"""
def __init__(
self,
model: Optional[str] | None = "llama3",
client: Optional[str] | None = "ollama",
max_tokens: Optional[int] | None = 4096,
temperature: Optional[float] | None = 0.5,
) -> None:
"""
Initialising Agent Module
Args:
model: str -> default = llama3
client: str -> default = ollama
max_tokens: int -> default = 4096
temperature: float -> default = 0.5
tools: List[Tool] -> default = None
"""
self.model = dspy.GROQ(
model="llama3-8b-8192",
temperature=temperature,
api_key=os.getenv("GROQ_API_KEY"),
max_tokens=max_tokens,
frequency_penalty=1.5,
presence_penalty=1.5,
)
dspy.settings.configure(
lm=self.model,
max_tokens = max_tokens,
temperature = temperature
)
def __deepcopy__(self, memo):
new_instance = self.__class__.__new__(self.__class__)
memo[id(self)] = new_instance
for k, v in self.__dict__.items():
if k != 'model':
setattr(new_instance, k, copy.deepcopy(v, memo))
new_instance.model = self.model
return new_instance
class OutputFormat(BaseModel):
expand: Optional[str]
topic: str
class Conversation(BaseModel):
role: str
content: str
class Memory(BaseModel):
conversations: List[Conversation]
class BaseSignature(dspy.Signature):
"""
You are an expert in expanding the user question and generating suitable tags for the question.
Follow the exact instructions given:
1. Expand with only single question.
2. Try to keep the actual content in the expand question. Example: User question: What is math ?, expand: What is mathematics ?
3. Tags should be 2-level hierarchy topics. Eg - India - Politics, Sports- Football. Tags should be as specific as possible. If it is a general question topic: GENERAL
4. Do not give the reference of the previous question in the expanded question.
5. If there is no expanded version of the user question, then give it as expand = "None"
6. If there is a general question asked, do not expand the question, just give it as expand="None"
7. topic can not be "None"
8. Use the provided memory to understand context and provide more relevant expansions and topics.
"""
query: str = dspy.InputField(prefix = "Question: ")
memory: Memory = dspy.InputField(prefix = "Previous conversations: ", desc="This is a list of previous conversations.")
output: OutputFormat = dspy.OutputField(desc='''Expanded user question and tags are generated as output. Respond with a single JSON object. JSON Schema: {"properties": {"expand": {"anyOf": [{"type": "string"}, {"type": "null"}], "title": "Expand"}, "topic": {"title": "Topic", "type": "string"}}, "required": ["expand", "topic"], "title": "OutputFormat", "type": "object"}''')
class OutputAgent(Agent):
"""
Multi-output Agent Module. Inherited from Agent Module
"""
def __init__(self, model: str | None = "llama3", client: str | None = "ollama", max_tokens: int | None = 8192) -> None:
super().__init__(
model = model,
client = client,
max_tokens = max_tokens
)
def __call__(self, query: str, memory: List[dict]) -> dspy.Prediction:
"""
This function expands the user question and generates the tags for the user question.
Args:
query: str -> The current user query
memory: List[dict] -> List of previous conversations
Returns:
dspy.Prediction: Expanded question and topic
"""
# Convert the memory list to the Memory model
conversations = [Conversation(role=m["role"], content=m["content"]) for m in memory]
memory_model = Memory(conversations=conversations)
# modules
outputGenerator = dspy.TypedPredictor(BaseSignature)
# infer
try:
output = outputGenerator(query=query, memory=memory_model)
return output
except Exception as e:
print("Retrying...", e)
return self.__call__(query=query, memory=memory)
# This function can be called from app.py to get the expanded question and topic
def get_expanded_query_and_topic(query: str, conversation_context: List[dict]):
agent = OutputAgent()
result = agent(query, conversation_context)
return result.output |