File size: 3,950 Bytes
ca7e9c6 f2f63e5 ca7e9c6 f2f63e5 ca7e9c6 a653647 ca7e9c6 7c6adb7 a653647 ca7e9c6 7c6adb7 ca7e9c6 |
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 |
##############################################################################
# Agent interfaces that bridges private capability agents (pandas,
# sql, ...), 3rd party plugin agents (search, weather, movie, ...),
# and 3rd party LLMs
#
# @philmui
# Mon May 1 18:34:45 PDT 2023
##############################################################################
from langchain.schema import HumanMessage
from langchain.prompts import PromptTemplate, ChatPromptTemplate, \
HumanMessagePromptTemplate
from models import load_chat_agent, load_chained_agent, load_sales_agent, \
load_sqlite_agent
import logging
logger = logging.getLogger(__name__)
# To parse outputs and get structured data back
from langchain.output_parsers import StructuredOutputParser, ResponseSchema
instruct_template = """
Please answer this question clearly with easy to follow reasoning:
{query}
If you don't know the answer, just reply: not available.
"""
instruct_prompt = PromptTemplate(
input_variables=["query"],
template=instruct_template
)
response_schemas = [
ResponseSchema(name="artist",
description="The name of the musical artist"),
ResponseSchema(name="song",
description="The name of the song that the artist plays")
]
output_parser = StructuredOutputParser.from_response_schemas(response_schemas)
format_instructions = output_parser.get_format_instructions()
LOCAL_MAGIC_TOKENS = ["my company", "for us", "our company", "our sales"]
DIGITAL_MAGIC_TOKENS = ["digital media", "our database", "our digital"]
def is_magic(sentence, magic_tokens):
return any([t in sentence.lower() for t in magic_tokens])
chat_prompt = ChatPromptTemplate(
messages=[
HumanMessagePromptTemplate.from_template(
"Given a command from the user, extract the artist and \
song names \n{format_instructions}\n{user_prompt}")
],
input_variables=["user_prompt"],
partial_variables={"format_instructions": format_instructions}
)
def chatAgent(chat_message):
try:
agent = load_chat_agent(verbose=True)
output = agent([HumanMessage(content=chat_message)])
except:
output = "Please rephrase and try chat again."
return output
def instructAgent(question_text, model_name):
output = ""
if is_magic(question_text, LOCAL_MAGIC_TOKENS):
output = salesAgent(question_text)
print(f"🔹 salesAgent: {output}")
elif is_magic(question_text, DIGITAL_MAGIC_TOKENS):
output = chinookAgent(question_text, model_name)
print(f"🔹 chinookAgent: {output}")
else:
try:
instruction = instruct_prompt.format(query=question_text)
logger.info(f"instruction: {instruction}")
agent = load_chained_agent(verbose=True, model_name=model_name)
response = agent([instruction])
if response is None or "not available" in response["output"]:
response = ""
else:
output = response['output']
logger.info(f"🔹 Steps: {response['intermediate_steps']}")
except Exception as e:
output = "Please rephrase and try again ..."
logger.error(e)
return output
def salesAgent(instruction):
output = ""
try:
agent = load_sales_agent(verbose=True)
output = agent.run(instruction)
print("panda> " + output)
except Exception as e:
logger.error(e)
output = f"Please rephrase and try again for company sales data {e}"
return output
def chinookAgent(instruction, model_name):
output = ""
try:
agent = load_sqlite_agent(model_name)
output = agent.run(instruction)
print("chinook> " + output)
except Exception as e:
logger.error(e)
output = "Please rephrase and try again for digital media data"
return output |