|
|
|
|
|
from flask import Flask,request |
|
from dotenv import load_dotenv |
|
|
|
from langchain.agents import tool |
|
|
|
|
|
|
|
app = Flask(__name__) |
|
load_dotenv() |
|
|
|
|
|
@tool |
|
def FAQ(question: str): |
|
"""Answers the question 1+1""" |
|
return 23 |
|
|
|
tools=[FAQ] |
|
|
|
|
|
@app.route('/', methods=['GET','POST']) |
|
def index(): |
|
|
|
from langchain_openai import ChatOpenAI |
|
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder |
|
from langchain.agents import AgentExecutor |
|
from typing import List |
|
from pydantic import BaseModel, Field |
|
import json |
|
|
|
from langchain.utils.openai_functions import convert_pydantic_to_openai_function |
|
from langchain.agents.format_scratchpad import format_to_openai_function_messages |
|
|
|
from langchain_core.agents import AgentActionMessageLog, AgentFinish |
|
|
|
|
|
class Response(BaseModel): |
|
"""Final response to the question being asked. This is consumed by a frontend chatbot engine that has the ability to execute suggested actions""" |
|
|
|
message: str = Field(description="The final answer to be displayed to the user") |
|
tokens: int = Field(description="Count the number of tokens used to produce the response") |
|
actions: List[int] = Field( |
|
description="List of actions to be executed. Only include an action if it contains relevant information" |
|
) |
|
|
|
|
|
def parse(output): |
|
if "function_call" not in output.additional_kwargs: return AgentFinish(return_values={"output": output.content}, log=output.content) |
|
function_call = output.additional_kwargs["function_call"] |
|
name = function_call["name"] |
|
inputs = json.loads(function_call["arguments"]) |
|
if name == "Response": |
|
return AgentFinish(return_values=inputs, log=str(function_call)) |
|
else: |
|
return AgentActionMessageLog( |
|
tool=name, tool_input=inputs, log="", message_log=[output] |
|
) |
|
|
|
prompt = ChatPromptTemplate.from_messages( |
|
[ |
|
("system", "You are a helpful assistant"), |
|
|
|
("user", "{input}"), |
|
MessagesPlaceholder(variable_name="agent_scratchpad"), |
|
] |
|
) |
|
|
|
llm = ChatOpenAI(temperature=2) |
|
|
|
llm_with_tools = llm.bind( |
|
functions=[ |
|
convert_pydantic_to_openai_function(Response), |
|
] |
|
) |
|
|
|
agent = ( |
|
{ |
|
"input": lambda x: x["input"], |
|
"agent_scratchpad": lambda x: format_to_openai_function_messages( |
|
x["intermediate_steps"] |
|
), |
|
} |
|
| prompt |
|
| llm_with_tools |
|
| parse |
|
) |
|
|
|
agent_executor = AgentExecutor(tools=[], agent=agent, verbose=True, handle_parsing_errors="Check your output and make sure it conforms, use the Action/Action Input syntax") |
|
|
|
return agent_executor.invoke( |
|
{"input": "what did the president say about kentaji brown jackson"}, |
|
return_only_outputs=True, |
|
) |
