refac: backend now is its own package

This commit is contained in:
ogabrielluiz 2023-03-07 15:05:39 -03:00
commit 0e79b82fea
15 changed files with 19 additions and 36 deletions

View file

@ -1,10 +0,0 @@
import uvicorn
import sys
from pathlib import Path
path = Path(__file__)
sys.path.append(str(path.parent.parent.parent))
from app import app
uvicorn.run(app, host="0.0.0.0", port=5003)

View file

@ -1,15 +1,15 @@
[tool.poetry]
name = "langflow"
name = "langflow-backend"
version = "0.0.21"
description = "Backend for Langflow"
authors = ["Ibis Prevedello <ibiscp@gmail.com>"]
# packages = [
# { include = "*.py" }
# ]
# exclude = ['dev.py']
authors = ["Ibis Prevedello <ibiscp@gmail.com>", "Gabriel Almeida <gabrielf.almeida90@gmail.com>"]
packages = [
{ include = "src" }
]
[tool.poetry.dependencies]
python = "^3.10"
python = "^3.9"
openai = "^0.26.5"
fastapi = "^0.91.0"
uvicorn = "^0.20.0"

View file

View file

View file

@ -0,0 +1,207 @@
from fastapi import APIRouter
from api import signature
from api import list_endpoints
from utils import payload
from langchain.agents.loading import load_agent_executor_from_config
from langchain.chains.loading import load_chain_from_config
from langchain.llms.loading import load_llm_from_config
from langchain.prompts.loading import load_prompt_from_config
from typing import Any
import io
import contextlib
import re
# build router
router = APIRouter()
def get_type_list():
all_types = get_all()
all_types.pop("tools")
for key, value in all_types.items():
all_types[key] = [item["template"]["_type"] for item in value.values()]
return all_types
@router.get("/all")
def get_all():
return {
"chains": {
chain: signature.get_chain(chain) for chain in list_endpoints.list_chains()
},
"agents": {
agent: signature.get_agent(agent) for agent in list_endpoints.list_agents()
},
"prompts": {
prompt: signature.get_prompt(prompt)
for prompt in list_endpoints.list_prompts()
},
"llms": {llm: signature.get_llm(llm) for llm in list_endpoints.list_llms()},
# "utilities": {
# "template": {
# # utility: templates.utility(utility) for utility in list.list_utilities()
# }
# },
# "memories": {
# memory: signature.get_memory(memory)
# for memory in list_endpoints.list_memories()
# },
# "document_loaders": {
# "template": {
# # memory: templates.document_loader(memory)
# # for memory in list.list_document_loaders()
# }
# },
# "vectorstores": {"template": {}},
# "docstores": {"template": {}},
# "tools": {
# tool: {"template": signature.tool(tool), **values}
# for tool, values in tools.items()
# },
"tools": {
tool: signature.get_tool(tool) for tool in list_endpoints.list_tools()
},
}
@router.post("/predict")
def get_load(data: dict[str, Any]):
# Get type list
type_list = get_type_list()
# Substitute ZeroShotPromt with PromptTemplate
for node in data["nodes"]:
if node["data"]["type"] == "ZeroShotPrompt":
# Build Prompt Template
tools = [
tool
for tool in data["nodes"]
if tool["type"] != "chatOutputNode"
and "Tool" in tool["data"]["node"]["base_classes"]
]
node["data"] = build_prompt_template(prompt=node["data"], tools=tools)
break
# Add input variables
data = payload.extract_input_variables(data)
# Nodes, edges and root node
message = data["message"]
nodes = data["nodes"]
edges = data["edges"]
root = payload.get_root_node(data)
extracted_json = payload.build_json(root, nodes, edges)
# Build json
if extracted_json["_type"] in type_list["agents"]:
loaded = load_agent_executor_from_config(extracted_json)
with io.StringIO() as output_buffer, contextlib.redirect_stdout(output_buffer):
result = loaded.run(message)
thought = output_buffer.getvalue()
elif extracted_json["_type"] in type_list["chains"]:
loaded = load_chain_from_config(extracted_json)
with io.StringIO() as output_buffer, contextlib.redirect_stdout(output_buffer):
result = loaded.run(message)
thought = output_buffer.getvalue()
elif extracted_json["_type"] in type_list["llms"]:
loaded = load_llm_from_config(extracted_json)
with io.StringIO() as output_buffer, contextlib.redirect_stdout(output_buffer):
result = loaded(message)
thought = output_buffer.getvalue()
else:
result = "Error: Type should be either agent, chain or llm"
thought = ""
# Remove unnecessary data from response
begin = thought.rfind(message)
thought = thought[(begin + len(message)) :]
return {
"result": result,
"thought": re.sub(
r"\x1b\[([0-9,A-Z]{1,2}(;[0-9,A-Z]{1,2})?)?[m|K]", "", thought
).strip(),
}
def build_prompt_template(prompt, tools):
prefix = prompt["node"]["template"]["prefix"]["value"]
suffix = prompt["node"]["template"]["suffix"]["value"]
format_instructions = prompt["node"]["template"]["format_instructions"]["value"]
tool_strings = "\n".join(
[
f"{tool['data']['node']['name']}: {tool['data']['node']['description']}"
for tool in tools
]
)
tool_names = ", ".join([tool["data"]["node"]["name"] for tool in tools])
format_instructions = format_instructions.format(tool_names=tool_names)
value = "\n\n".join([prefix, tool_strings, format_instructions, suffix])
prompt["type"] = "PromptTemplate"
prompt["node"] = {
"template": {
"_type": "prompt",
"input_variables": {
"type": "str",
"required": True,
"placeholder": "",
"list": True,
"show": False,
"multiline": False,
},
"output_parser": {
"type": "BaseOutputParser",
"required": False,
"placeholder": "",
"list": False,
"show": False,
"multline": False,
"value": None,
},
"template": {
"type": "str",
"required": True,
"placeholder": "",
"list": False,
"show": True,
"multiline": True,
"value": value,
},
"template_format": {
"type": "str",
"required": False,
"placeholder": "",
"list": False,
"show": False,
"multline": False,
"value": "f-string",
},
"validate_template": {
"type": "bool",
"required": False,
"placeholder": "",
"list": False,
"show": False,
"multline": False,
"value": True,
},
},
"description": "Schema to represent a prompt for an LLM.",
"base_classes": ["BasePromptTemplate"],
}
return prompt

View file

@ -0,0 +1,126 @@
from fastapi import APIRouter
from langchain import chains
from langchain import agents
from langchain import prompts
from langchain import llms
from langchain.chains.conversation import memory as memories
from langchain.agents.load_tools import get_all_tool_names
from utils import util, allowed_components
from custom import customs
# build router
router = APIRouter(
prefix="/list",
tags=["list"],
)
@router.get("/")
def read_items():
"""List all components"""
return [
"chains",
"agents",
"prompts",
"llms",
# "utilities",
# "memories",
# "document_loaders",
# "vectorstores",
# "docstores",
"tools",
]
@router.get("/chains")
def list_chains():
"""List all chain types"""
return [
chain.__annotations__["return"].__name__
for chain in chains.loading.type_to_loader_dict.values()
if chain.__annotations__["return"].__name__ in allowed_components.CHAINS
]
@router.get("/agents")
def list_agents():
"""List all agent types"""
# return list(agents.loading.AGENT_TO_CLASS.keys())
return [
agent.__name__
for agent in agents.loading.AGENT_TO_CLASS.values()
if agent.__name__ in allowed_components.AGENTS
]
@router.get("/prompts")
def list_prompts():
"""List all prompt types"""
custom_prompts = customs.get_custom_prompts()
library_prompts = [
prompt.__annotations__["return"].__name__
for prompt in prompts.loading.type_to_loader_dict.values()
if prompt.__annotations__["return"].__name__ in allowed_components.PROMPTS
]
return library_prompts + list(custom_prompts.keys())
@router.get("/llms")
def list_llms():
"""List all llm types"""
return [
llm.__name__
for llm in llms.type_to_cls_dict.values()
if llm.__name__ in allowed_components.LLMS
]
@router.get("/memories")
def list_memories():
"""List all memory types"""
return [memory.__name__ for memory in memories.type_to_cls_dict.values()]
# @router.get("/utilities")
# def list_utilities():
# """List all utility types"""
# return list(utilities.__all__)
# @router.get("/document_loaders")
# def list_document_loaders():
# """List all document loader types"""
# return list(document_loaders.__all__)
# @router.get("/vectorstores")
# def list_vectorstores():
# """List all vector store types"""
# return list(vectorstores.__all__)
# @router.get("/docstores")
# def list_docstores():
# """List all document store types"""
# return list(docstore.__all__)
@router.get("/tools")
def list_tools():
"""List all load tools"""
tools = []
for tool in get_all_tool_names():
tool_params = util.get_tool_params(util.get_tools_dict(tool))
if tool_params and tool_params["name"] in allowed_components.TOOLS:
tools.append(tool_params["name"])
return tools
# return [
# util.get_tool_params(util.get_tools_dict(tool))["name"]
# for tool in get_all_tool_names()
# ]

View file

@ -0,0 +1,148 @@
from typing import Any, Dict
from fastapi import APIRouter, HTTPException
from langchain import agents, chains, llms, prompts
from langchain.agents.load_tools import (
_BASE_TOOLS,
_EXTRA_LLM_TOOLS,
_EXTRA_OPTIONAL_TOOLS,
_LLM_TOOLS,
get_all_tool_names,
)
from langchain.chains.conversation import memory as memories
from utils import util
from custom import customs
# build router
router = APIRouter(
prefix="/signatures",
tags=["signatures"],
)
@router.get("/chain")
def get_chain(name: str):
"""Get the signature of a chain."""
try:
return util.build_template_from_function(
name, chains.loading.type_to_loader_dict
)
except ValueError as exc:
raise HTTPException(status_code=404, detail="Chain not found") from exc
@router.get("/agent")
def get_agent(name: str):
"""Get the signature of an agent."""
try:
return util.build_template_from_class(name, agents.loading.AGENT_TO_CLASS)
except ValueError as exc:
raise HTTPException(status_code=404, detail="Agent not found") from exc
@router.get("/prompt")
def get_prompt(name: str):
"""Get the signature of a prompt."""
try:
if name in customs.get_custom_prompts().keys():
return customs.get_custom_prompts()[name]
return util.build_template_from_function(
name, prompts.loading.type_to_loader_dict
)
except ValueError as exc:
raise HTTPException(status_code=404, detail="Prompt not found") from exc
@router.get("/llm")
def get_llm(name: str):
"""Get the signature of an llm."""
try:
return util.build_template_from_class(name, llms.type_to_cls_dict)
except ValueError as exc:
raise HTTPException(status_code=404, detail="LLM not found") from exc
# @router.get("/utility")
# def utility(name: str):
# # Raise error if name is not in utilities
# if name not in utilities.__all__:
# raise Exception(f"Prompt {name} not found.")
# _class = getattr(utilities, name)
# return {
# name: {name: value for (name, value) in value.__repr_args__() if name != "name"}
# for name, value in _class.__fields__.items()
# }
@router.get("/memory")
def get_memory(name: str):
"""Get the signature of a memory."""
try:
return util.build_template_from_class(name, memories.type_to_cls_dict)
except ValueError as exc:
raise HTTPException(status_code=404, detail="Memory not found") from exc
# @router.get("/document_loader")
# def document_loader(name: str):
# # Raise error if name is not in document_loader
# if name not in document_loaders.__all__:
# raise Exception(f"Prompt {name} not found.")
# _class = getattr(document_loaders, name)
# return {
# name: {name: value for (name, value) in value.__repr_args__() if name != "name"}
# for name, value in _class.__fields__.items()
# }
@router.get("/tool")
def get_tool(name: str):
"""Get the signature of a tool."""
all_tools = {}
for tool in get_all_tool_names():
if tool_params := util.get_tool_params(util.get_tools_dict(tool)):
all_tools[tool_params["name"]] = tool
# Raise error if name is not in tools
if name not in all_tools.keys():
raise HTTPException(status_code=404, detail=f"Tool {name} not found.")
type_dict = {
"str": {
"type": "str",
"required": False,
"list": False,
"show": True,
"placeholder": "",
"value": "",
},
"llm": {"type": "BaseLLM", "required": True, "list": False, "show": True},
}
tool_type = all_tools[name]
if tool_type in _BASE_TOOLS:
params = []
elif tool_type in _LLM_TOOLS:
params = ["llm"]
elif tool_type in _EXTRA_LLM_TOOLS:
_, extra_keys = _EXTRA_LLM_TOOLS[tool_type]
params = ["llm"] + extra_keys
elif tool_type in _EXTRA_OPTIONAL_TOOLS:
_, extra_keys = _EXTRA_OPTIONAL_TOOLS[tool_type]
params = extra_keys
else:
params = []
template = {
param: (type_dict[param] if param == "llm" else type_dict["str"])
for param in params
} # type: Dict[str, Any]
template["_type"] = tool_type
return {
"template": template,
**util.get_tool_params(util.get_tools_dict(tool_type)),
"base_classes": ["Tool"],
}

View file

View file

@ -0,0 +1,40 @@
from langchain.agents.mrkl import prompt
def get_custom_prompts():
return {
"ZeroShotPrompt": {
"template": {
"_type": "zero_shot",
"prefix": {
"type": "str",
"required": False,
"placeholder": "",
"list": False,
"show": True,
"multiline": True,
"value": prompt.PREFIX,
},
"suffix": {
"type": "str",
"required": True,
"placeholder": "",
"list": False,
"show": True,
"multiline": True,
"value": prompt.SUFFIX,
},
"format_instructions": {
"type": "str",
"required": False,
"placeholder": "",
"list": False,
"show": True,
"multiline": True,
"value": prompt.FORMAT_INSTRUCTIONS,
},
},
"description": "Prompt template for Zero Shot Agent.",
"base_classes": ["BasePromptTemplate"],
}
}

View file

@ -0,0 +1,36 @@
from fastapi import FastAPI
from api.endpoints import router as endpoints_router
from api.list_endpoints import router as list_router
from api.signature import router as signatures_router
from fastapi.middleware.cors import CORSMiddleware
def create_app():
"""Create the FastAPI app and include the router."""
app = FastAPI()
origins = [
"*",
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(endpoints_router)
app.include_router(list_router)
app.include_router(signatures_router)
return app
app = create_app()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=5003)

View file

View file

@ -0,0 +1,9 @@
CHAINS = ["LLMChain", "LLMMathChain", "LLMRequestsChain", "LLMBashChain"]
AGENTS = ["ZeroShotAgent"]
PROMPTS = ["PromptTemplate", "FewShotPromptTemplate"]
LLMS = ["OpenAI"]
TOOLS = ["Search", "requests_get", "PAL-MATH", "Calculator"]

View file

@ -0,0 +1,92 @@
import re
def extract_input_variables(data):
"""
Extracts input variables from the template and adds them to the input_variables field.
"""
for node in data["nodes"]:
try:
if "input_variables" in node["data"]["node"]["template"]:
if node["data"]["node"]["template"]["_type"] == "prompt":
variables = re.findall(
r"\{(.*?)\}",
node["data"]["node"]["template"]["template"]["value"],
)
elif node["data"]["node"]["template"]["_type"] == "few_shot":
variables = re.findall(
r"\{(.*?)\}",
node["data"]["node"]["template"]["prefix"]["value"]
+ node["data"]["node"]["template"]["suffix"]["value"],
)
else:
variables = []
node["data"]["node"]["template"]["input_variables"]["value"] = variables
except:
pass
return data
def get_root_node(data):
"""
Returns the root node of the template.
"""
root = None
incoming_edges = {edge["source"] for edge in data["edges"]}
for node in data["nodes"]:
if node["id"] not in incoming_edges:
root = node
break
return root
def build_json(root, nodes, edges):
edge_ids = [edge["source"] for edge in edges if edge["target"] == root["id"]]
local_nodes = [node for node in nodes if node["id"] in edge_ids]
if "node" not in root["data"]:
return build_json(local_nodes[0], nodes, edges)
final_dict = root["data"]["node"]["template"].copy()
for key, value in final_dict.items():
if key == "_type":
continue
# elif key == "prompt":
# pass
module_type = value["type"]
# if module_type == "Tool":
# pass
# if module_type in ["str", "bool", "int", "float", "Any"] or value["value"]:
# value = value["value"]
if "value" in value and value["value"] is not None:
value = value["value"]
elif "dict" in module_type:
value = {}
else:
# if value['list']:
# print(key)
children = []
for c in local_nodes:
module_types = [c["data"]["type"]]
if "node" in c["data"]:
module_types += c["data"]["node"]["base_classes"]
if module_type in module_types:
children.append(c)
# children = [
# c
# for c in local_nodes
# if module_type
# in [c["data"]["type"]] + c["data"]["node"]["base_classes"]
# ]
# else:
# children = next((c for c in local_nodes if type in [c['data']['type']] + c['data']['node']['base_classes']), None)
if value["required"] and not children:
raise ValueError(f"No child with type {module_type} found")
values = [
build_json(child, nodes, edges) for child in children
] # if children else None
value = list(values) if value["list"] else next(iter(values), None)
final_dict[key] = value
return final_dict

View file

@ -0,0 +1,291 @@
import ast
import inspect
import re
import importlib
from langchain.agents.load_tools import *
from langchain.agents.load_tools import (
_BASE_TOOLS,
_LLM_TOOLS,
_EXTRA_LLM_TOOLS,
_EXTRA_OPTIONAL_TOOLS,
)
from typing import Optional
def build_template_from_function(name: str, type_to_loader_dict: dict):
classes = [
item.__annotations__["return"].__name__ for item in type_to_loader_dict.values()
]
# Raise error if name is not in chains
if name not in classes:
raise ValueError(f"{name} not found")
for _type, v in type_to_loader_dict.items():
if v.__annotations__["return"].__name__ == name:
_class = v.__annotations__["return"]
docs = get_class_doc(_class)
variables = {"_type": _type}
for name, value in _class.__fields__.items():
if name in ["callback_manager", "requests_wrapper"]:
continue
variables[name] = {}
for name_, value_ in value.__repr_args__():
if name_ == "default_factory":
try:
variables[name]["default"] = get_default_factory(
module=_class.__base__.__module__, function=value_
)
except Exception:
variables[name]["default"] = None
elif name_ not in ["name"]:
variables[name][name_] = value_
variables[name]["placeholder"] = (
docs["Attributes"][name] if name in docs["Attributes"] else ""
)
return {
"template": format_dict(variables),
"description": docs["Description"],
"base_classes": get_base_classes(_class),
}
def build_template_from_class(name: str, type_to_cls_dict: dict):
classes = [item.__name__ for item in type_to_cls_dict.values()]
# Raise error if name is not in chains
if name not in classes:
raise ValueError(f"{name} not found.")
for _type, v in type_to_cls_dict.items():
if v.__name__ == name:
_class = v
docs = get_class_doc(_class)
variables = {"_type": _type}
for name, value in _class.__fields__.items():
if name in ["callback_manager"]:
continue
variables[name] = {}
for name_, value_ in value.__repr_args__():
if name_ == "default_factory":
try:
variables[name]["default"] = get_default_factory(
module=_class.__base__.__module__, function=value_
)
except Exception:
variables[name]["default"] = None
elif name_ not in ["name"]:
variables[name][name_] = value_
variables[name]["placeholder"] = (
docs["Attributes"][name] if name in docs["Attributes"] else ""
)
return {
"template": format_dict(variables),
"description": docs["Description"],
"base_classes": get_base_classes(_class),
}
def get_base_classes(cls):
bases = cls.__bases__
if not bases:
return []
else:
result = []
for base in bases:
if any(type in base.__module__ for type in ["pydantic", "abc"]):
continue
result.append(base.__name__)
result.extend(get_base_classes(base))
return result
def get_default_factory(module: str, function: str):
pattern = r"<function (\w+)>"
if match := re.search(pattern, function):
imported_module = importlib.import_module(module)
return getattr(imported_module, match[1])()
return None
def get_tools_dict(name: Optional[str] = None):
"""Get the tools dictionary."""
tools = {
**_BASE_TOOLS,
**_LLM_TOOLS,
**{k: v[0] for k, v in _EXTRA_LLM_TOOLS.items()},
**{k: v[0] for k, v in _EXTRA_OPTIONAL_TOOLS.items()},
}
return tools[name] if name else tools
def get_tool_params(func, **kwargs):
# Parse the function code into an abstract syntax tree
tree = ast.parse(inspect.getsource(func))
# Iterate over the statements in the abstract syntax tree
for node in ast.walk(tree):
# Find the first return statement
if isinstance(node, ast.Return):
tool = node.value
if isinstance(tool, ast.Call):
if tool.func.id == "Tool":
if tool.keywords:
tool_params = {}
for keyword in tool.keywords:
if keyword.arg == "name":
tool_params["name"] = ast.literal_eval(keyword.value)
elif keyword.arg == "description":
tool_params["description"] = ast.literal_eval(
keyword.value
)
return tool_params
return {
"name": ast.literal_eval(tool.args[0]),
"description": ast.literal_eval(tool.args[2]),
}
else:
# get the class object from the return statement
try:
class_obj = eval(
compile(ast.Expression(tool), "<string>", "eval")
)
except Exception:
return None
return {
"name": getattr(class_obj, "name"),
"description": getattr(class_obj, "description"),
}
# Return None if no return statement was found
return None
def get_class_doc(class_name):
"""
Extracts information from the docstring of a given class.
Args:
class_name: the class to extract information from
Returns:
A dictionary containing the extracted information, with keys
for 'Description', 'Parameters', 'Attributes', and 'Returns'.
"""
# Get the class docstring
docstring = class_name.__doc__
# Parse the docstring to extract information
lines = docstring.split("\n")
data = {
"Description": "",
"Parameters": {},
"Attributes": {},
"Example": [],
"Returns": {},
}
current_section = "Description"
for line in lines:
line = line.strip()
if not line:
continue
if (
line.startswith(tuple(data.keys()))
and len(line.split()) == 1
and line.endswith(":")
):
current_section = line[:-1]
continue
if current_section in ["Description", "Example"]:
data[current_section] += line
else:
param, desc = line.split(":")
data[current_section][param.strip()] = desc.strip()
return data
def format_dict(d):
"""
Formats a dictionary by removing certain keys and modifying the
values of other keys.
Args:
d: the dictionary to format
Returns:
A new dictionary with the desired modifications applied.
"""
# Process remaining keys
for key, value in d.items():
if key == "_type":
continue
# Set verbose to True
if key == "verbose":
value["default"] = True
_type = value["type"]
# Remove 'Optional' wrapper
if "Optional" in _type:
_type = _type.replace("Optional[", "")[:-1]
# Check for list type
if "List" in _type:
_type = _type.replace("List[", "")[:-1]
value["list"] = True
else:
value["list"] = False
# Replace 'Mapping' with 'dict'
if "Mapping" in _type:
_type = _type.replace("Mapping", "dict")
value["type"] = "Tool" if key == "allowed_tools" else _type
# Show if required
value["show"] = bool(
(value["required"] and key not in ["input_variables"])
or key
in [
"allowed_tools",
# "Memory",
"memory",
"prefix",
"examples",
"temperature",
]
or "api_key" in key
)
# Add multline
value["multiline"] = key in ["suffix", "prefix", "template", "examples"]
# Replace default value with actual value
# if _type in ["str", "bool"]:
# value["value"] = value.get("default", "")
# if "default" in value:
# value.pop("default")
if "default" in value:
value["value"] = value["default"]
value.pop("default")
# Filter out keys that should not be shown
return d

View file

@ -2,7 +2,7 @@
name = "langflow"
version = "0.0.1"
description = "A Python package with a built-in web application"
authors = ["Ibis Prevedello <ibiscp@gmail.com>"]
authors = ["Ibis Prevedello <ibiscp@gmail.com>", "Gabriel Almeida <gabrielf.almeida90@gmail.com>"]
packages = [
{ include = "langflow"}
]
@ -23,11 +23,5 @@ build-backend = "poetry.core.masonry.api"
[tool.poetry.dependencies]
python = "^3.9"
openai = "^0.26.5"
fastapi = "^0.91.0"
uvicorn = "^0.20.0"
beautifulsoup4 = "^4.11.2"
google-search-results = "^2.4.1"
google-api-python-client = "^2.79.0"
langchain = {git = "https://github.com/ibiscp/langchain.git", rev = "ibis"}
langflow_backend = {path = "langflow/backend"}
typer = "^0.7.0"