refac: listing and signature for tools moved

This commit is contained in:
Gabriel Almeida 2023-03-30 18:13:35 -03:00
commit 18b3fa6c34
14 changed files with 169 additions and 586 deletions

View file

@ -1,58 +0,0 @@
from fastapi import APIRouter
from langflow.interface.listing import list_type
# build router
router = APIRouter(
prefix="/list",
tags=["list"],
)
@router.get("/")
def read_items():
"""List all components"""
return [
"chains",
"agents",
"prompts",
"llms",
"tools",
]
@router.get("/chains")
def list_chains():
"""List all chain types"""
return list_type("chains")
@router.get("/agents")
def list_agents():
"""List all agent types"""
# return list(agents.loading.AGENT_TO_CLASS.keys())
return list_type("agents")
@router.get("/prompts")
def list_prompts():
"""List all prompt types"""
return list_type("prompts")
@router.get("/llms")
def list_llms():
"""List all llm types"""
return list_type("llms")
@router.get("/memories")
def list_memories():
"""List all memory types"""
return list_type("memories")
@router.get("/tools")
def list_tools():
"""List all load tools"""
return list_type("tools")

View file

@ -1,63 +0,0 @@
from fastapi import APIRouter, HTTPException
from langflow.interface.signature import get_signature
# build router
router = APIRouter(
prefix="/signatures",
tags=["signatures"],
)
@router.get("/chain")
def get_chain(name: str):
"""Get the signature of a chain."""
try:
return get_signature(name, "chains")
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 get_signature(name, "agents")
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:
return get_signature(name, "prompts")
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 get_signature(name, "llms")
except ValueError as exc:
raise HTTPException(status_code=404, detail="LLM not found") from exc
@router.get("/memory")
def get_memory(name: str):
"""Get the signature of a memory."""
try:
return get_signature(name, "memories")
except ValueError as exc:
raise HTTPException(status_code=404, detail="Memory not found") from exc
@router.get("/tool")
def get_tool(name: str):
"""Get the signature of a tool."""
try:
return get_signature(name, "tools")
except ValueError as exc:
raise HTTPException(status_code=404, detail="Tool not found") from exc

View file

@ -24,9 +24,4 @@ tools:
- PythonFunction - PythonFunction
- JsonSpec - JsonSpec
memories:
# - ConversationBufferMemory
dev: false dev: false

View file

@ -6,8 +6,9 @@
from copy import deepcopy from copy import deepcopy
import types import types
from typing import Any, Dict, List from typing import Any, Dict, List
from langflow.interface.listing import ALL_TYPES_DICT, TOOLS_DICT from langflow.interface.listing import ALL_TYPES_DICT
from langflow.interface import loading from langflow.interface import loading
from langflow.interface.tools.base import tool_creator
class Node: class Node:
@ -139,7 +140,7 @@ class Node:
# and return the instance # and return the instance
for base_type, value in ALL_TYPES_DICT.items(): for base_type, value in ALL_TYPES_DICT.items():
if base_type == "tools": if base_type == "tools":
value = TOOLS_DICT value = tool_creator.type_to_loader_dict
if self.node_type in value: if self.node_type in value:
self._built_object = loading.instantiate_class( self._built_object = loading.instantiate_class(
@ -208,5 +209,3 @@ class Edge:
f"Edge(source={self.source.id}, target={self.target.id}, valid={self.valid}" f"Edge(source={self.source.id}, target={self.target.id}, valid={self.valid}"
f", matched_type={self.matched_type})" f", matched_type={self.matched_type})"
) )

View file

@ -1,6 +1,6 @@
from typing import Dict, List, Union from typing import Dict, List, Union
from langflow.utils import payload from langflow.utils import payload
from langflow.interface.listing import ALL_TOOLS_NAMES from langflow.interface.tools.constants import ALL_TOOLS_NAMES
from langflow.graph.base import Node, Edge from langflow.graph.base import Node, Edge
from langflow.graph.nodes import ( from langflow.graph.nodes import (

View file

@ -8,7 +8,7 @@ from langchain.agents import Agent
from langchain.chains.base import Chain from langchain.chains.base import Chain
from langchain.llms.base import BaseLLM from langchain.llms.base import BaseLLM
from langchain.tools import BaseTool from langchain.tools import BaseTool
from langflow.utils.util import get_tool_by_name from langflow.interface.tools.util import get_tool_by_name
def import_module(module_path: str) -> Any: def import_module(module_path: str) -> Any:

View file

@ -1,126 +1,23 @@
from langchain import agents, chains, prompts from langflow.interface.agents.base import agent_creator
from langchain.agents import agent_toolkits from langflow.interface.chains.base import chain_creator
from langchain import requests from langflow.interface.llms.base import llm_creator
from langflow.custom import customs from langflow.interface.memories.base import memory_creator
from langflow.interface.custom_lists import ( from langflow.interface.prompts.base import prompt_creator
llm_type_to_cls_dict, from langflow.interface.tools.base import tool_creator
memory_type_to_cls_dict,
)
from langflow.settings import settings
from langflow.utils import util
from langchain.agents.load_tools import get_all_tool_names
from langchain.agents import Tool
from langflow.interface.custom_types import JsonAgent, PythonFunction
from langchain.tools.json.tool import JsonSpec
OTHER_TOOLS = {"JsonSpec": JsonSpec}
CUSTOM_TOOLS = {"Tool": Tool, "PythonFunction": PythonFunction}
TOOLS_DICT = util.get_tools_dict()
ALL_TOOLS_NAMES = set(
get_all_tool_names() + list(CUSTOM_TOOLS.keys()) + list(OTHER_TOOLS.keys())
)
def get_type_dict(): def get_type_dict():
return { return {
"chains": list_chain_types, "agents": agent_creator.to_list(),
"agents": list_agents, "prompts": prompt_creator.to_list(),
"prompts": list_prompts, "llms": llm_creator.to_list(),
"llms": list_llms, "tools": tool_creator.to_list(),
"tools": list_tools, "chains": chain_creator.to_list(),
"memories": list_memories, "memory": memory_creator.to_list(),
"toolkits": list_toolkis,
"wrappers": list_wrappers,
} }
def list_type(object_type: str): LANGCHAIN_TYPES_DICT = get_type_dict()
"""List all components"""
return get_type_dict().get(object_type, lambda: None)()
def list_wrappers():
"""List all wrapper types"""
return [requests.RequestsWrapper.__name__]
def list_agents():
"""List all agent types"""
return [
agent.__name__
for agent in agents.loading.AGENT_TO_CLASS.values()
if agent.__name__ in settings.agents or settings.dev
] + [JsonAgent.__name__]
def list_toolkis():
"""List all toolkit types"""
return agent_toolkits.__all__
def list_prompts():
"""List all prompt types"""
custom_prompts = customs.get_custom_nodes("prompts")
library_prompts = [
prompt.__annotations__["return"].__name__
for prompt in prompts.loading.type_to_loader_dict.values()
if prompt.__annotations__["return"].__name__ in settings.prompts or settings.dev
]
return library_prompts + list(custom_prompts.keys())
def list_tools():
"""List all load tools"""
tools = []
for tool in ALL_TOOLS_NAMES:
tool_params = util.get_tool_params(util.get_tool_by_name(tool))
if "name" not in tool_params:
tool_params["name"] = tool
if tool_params and (
tool_params.get("name") in settings.tools
or (tool_params.get("name") and settings.dev)
):
tools.append(tool_params["name"])
# Add Tool
custom_tools = customs.get_custom_nodes("tools")
return tools + list(custom_tools.keys())
def list_llms():
"""List all llm types"""
return [
llm.__name__
for llm in llm_type_to_cls_dict.values()
if llm.__name__ in settings.llms or settings.dev
]
def list_chain_types():
"""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 settings.chains or settings.dev
]
def list_memories():
"""List all memory types"""
return [
memory.__name__
for memory in memory_type_to_cls_dict.values()
if memory.__name__ in settings.memories or settings.dev
]
LANGCHAIN_TYPES_DICT = {
k: list_function() for k, list_function in get_type_dict().items()
}
# Now we'll build a dict with Langchain types and ours # Now we'll build a dict with Langchain types and ours

View file

@ -1,204 +0,0 @@
from typing import Any, Dict # noqa: F401
from langchain import agents, chains, prompts
from langchain.agents.load_tools import (
_BASE_TOOLS,
_EXTRA_LLM_TOOLS,
_EXTRA_OPTIONAL_TOOLS,
_LLM_TOOLS,
)
from langflow.custom import customs
from langflow.interface.custom_lists import (
llm_type_to_cls_dict,
memory_type_to_cls_dict,
toolkit_type_to_cls_dict,
wrapper_type_to_cls_dict,
)
from langflow.interface.listing import CUSTOM_TOOLS, ALL_TOOLS_NAMES
from langflow.template.template import Field, Template
from langflow.utils import util
def get_signature(name: str, object_type: str):
"""Get the signature of an object."""
return {
"toolkits": get_toolkit_signature,
"chains": get_chain_signature,
"agents": get_agent_signature,
"prompts": get_prompt_signature,
"llms": get_llm_signature,
# "memories": get_memory_signature,
"tools": get_tool_signature,
"wrappers": get_wrapper_signature,
}.get(object_type, lambda name: f"Invalid type: {name}")(name)
def get_toolkit_signature(name: str):
"""Get the signature of a toolkit."""
try:
if name.islower():
...
# return util.build_template_from_function(
# name, toolkit_type_to_loader_dict, add_function=True
# )
else:
return util.build_template_from_class(
name, toolkit_type_to_cls_dict, add_function=True
)
except ValueError as exc:
raise ValueError("Toolkit not found") from exc
def get_wrapper_signature(name: str):
"""Get the signature of a wrapper."""
try:
return util.build_template_from_class(
name,
wrapper_type_to_cls_dict,
)
except ValueError as exc:
raise ValueError("Wrapper not found") from exc
def get_chain_signature(name: str):
"""Get the chain type by signature."""
try:
return util.build_template_from_function(
name, chains.loading.type_to_loader_dict, add_function=True
)
except ValueError as exc:
raise ValueError("Chain not found") from exc
def get_agent_signature(name: str):
"""Get the signature of an agent."""
try:
if name in customs.get_custom_nodes("agents").keys():
return customs.get_custom_nodes("agents")[name]
return util.build_template_from_class(
name, agents.loading.AGENT_TO_CLASS, add_function=True
)
except ValueError as exc:
raise ValueError("Agent not found") from exc
def get_prompt_signature(name: str):
"""Get the signature of a prompt."""
try:
if name in customs.get_custom_nodes("prompts").keys():
return customs.get_custom_nodes("prompts")[name]
return util.build_template_from_function(
name, prompts.loading.type_to_loader_dict
)
except ValueError as exc:
raise ValueError("Prompt not found") from exc
def get_llm_signature(name: str):
"""Get the signature of an llm."""
try:
return util.build_template_from_class(name, llm_type_to_cls_dict)
except ValueError as exc:
raise ValueError("LLM not found") from exc
def get_memory_signature(name: str):
"""Get the signature of a memory."""
try:
return util.build_template_from_class(name, memory_type_to_cls_dict)
except ValueError as exc:
raise ValueError("Memory not found") from exc
def get_tool_signature(name: str):
"""Get the signature of a tool."""
NODE_INPUTS = ["llm", "func"]
base_classes = ["Tool"]
all_tools = {}
for tool in ALL_TOOLS_NAMES:
if tool_params := util.get_tool_params(util.get_tool_by_name(tool)):
tool_name = tool_params.get("name") or str(tool)
all_tools[tool_name] = {"type": tool, "params": tool_params}
# Raise error if name is not in tools
if name not in all_tools.keys():
raise ValueError("Tool not found")
type_dict = {
"str": Field(
field_type="str",
required=True,
is_list=False,
show=True,
placeholder="",
value="",
),
"llm": Field(field_type="BaseLLM", required=True, is_list=False, show=True),
"func": Field(
field_type="function",
required=True,
is_list=False,
show=True,
multiline=True,
),
"code": Field(
field_type="str",
required=True,
is_list=False,
show=True,
value="",
multiline=True,
),
}
tool_type: str = all_tools[name]["type"] # type: ignore
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
elif tool_type == "Tool":
params = ["name", "description", "func"]
elif tool_type in CUSTOM_TOOLS:
# Get custom tool params
params = all_tools[name]["params"] # type: ignore
base_classes = ["function"]
if node := customs.get_custom_nodes("tools").get(tool_type):
return node
else:
params = []
# Copy the field and add the name
fields = []
for param in params:
if param in NODE_INPUTS:
field = type_dict[param].copy()
else:
field = type_dict["str"].copy()
field.name = param
if param == "aiosession":
field.show = False
field.required = False
fields.append(field)
template = Template(fields=fields, type_name=tool_type)
tool_params = util.get_tool_params(util.get_tool_by_name(tool_type))
if tool_params is None:
tool_params = {}
return {
"template": util.format_dict(template.to_dict()),
**tool_params,
"base_classes": base_classes,
}

View file

@ -1,5 +1,6 @@
from langflow.custom import customs from langflow.custom import customs
from langflow.interface.listing import ALL_TOOLS_NAMES, CUSTOM_TOOLS from langflow.interface.tools.constants import ALL_TOOLS_NAMES, CUSTOM_TOOLS
import langflow.interface.tools.util
from langflow.template.template import Field, Template from langflow.template.template import Field, Template
from langflow.utils import util from langflow.utils import util
from langflow.settings import settings from langflow.settings import settings
@ -11,14 +12,18 @@ from langchain.agents.load_tools import (
_EXTRA_OPTIONAL_TOOLS, _EXTRA_OPTIONAL_TOOLS,
_LLM_TOOLS, _LLM_TOOLS,
) )
from langflow.interface.tools.util import get_tools_dict
class ToolCreator(LangChainTypeCreator): class ToolCreator(LangChainTypeCreator):
type_name: str = "tools" type_name: str = "tools"
tools_dict: Dict | None = None
@property @property
def type_to_loader_dict(self) -> Dict: def type_to_loader_dict(self) -> Dict:
return util.get_tools_dict() if self.tools_dict is None:
self.tools_dict = get_tools_dict()
return self.tools_dict
def get_signature(self, name: str) -> Dict | None: def get_signature(self, name: str) -> Dict | None:
"""Get the signature of a tool.""" """Get the signature of a tool."""
@ -27,7 +32,9 @@ class ToolCreator(LangChainTypeCreator):
base_classes = ["Tool"] base_classes = ["Tool"]
all_tools = {} all_tools = {}
for tool in self.type_to_loader_dict.keys(): for tool in self.type_to_loader_dict.keys():
if tool_params := util.get_tool_params(util.get_tool_by_name(tool)): if tool_params := langflow.interface.tools.util.get_tool_params(
langflow.interface.tools.util.get_tool_by_name(tool)
):
tool_name = tool_params.get("name") or str(tool) tool_name = tool_params.get("name") or str(tool)
all_tools[tool_name] = {"type": tool, "params": tool_params} all_tools[tool_name] = {"type": tool, "params": tool_params}
@ -101,7 +108,9 @@ class ToolCreator(LangChainTypeCreator):
template = Template(fields=fields, type_name=tool_type) template = Template(fields=fields, type_name=tool_type)
tool_params = util.get_tool_params(util.get_tool_by_name(tool_type)) tool_params = langflow.interface.tools.util.get_tool_params(
langflow.interface.tools.util.get_tool_by_name(tool_type)
)
if tool_params is None: if tool_params is None:
tool_params = {} tool_params = {}
return { return {
@ -116,7 +125,9 @@ class ToolCreator(LangChainTypeCreator):
tools = [] tools = []
for tool in ALL_TOOLS_NAMES: for tool in ALL_TOOLS_NAMES:
tool_params = util.get_tool_params(util.get_tool_by_name(tool)) tool_params = langflow.interface.tools.util.get_tool_params(
langflow.interface.tools.util.get_tool_by_name(tool)
)
if tool_params and ( if tool_params and (
tool_params.get("name") in settings.tools tool_params.get("name") in settings.tools
or (tool_params.get("name") and settings.dev) or (tool_params.get("name") and settings.dev)

View file

@ -0,0 +1,11 @@
from langchain.agents.load_tools import get_all_tool_names
from langchain.agents import Tool
from langflow.interface.custom_types import PythonFunction
from langchain.tools.json.tool import JsonSpec
OTHER_TOOLS = {"JsonSpec": JsonSpec}
CUSTOM_TOOLS = {"Tool": Tool, "PythonFunction": PythonFunction}
ALL_TOOLS_NAMES = set(
get_all_tool_names() + list(CUSTOM_TOOLS.keys()) + list(OTHER_TOOLS.keys())
)

View file

@ -0,0 +1,122 @@
import ast
import inspect
from typing import Dict, Union
from langchain.agents.load_tools import (
_BASE_TOOLS,
_EXTRA_LLM_TOOLS,
_EXTRA_OPTIONAL_TOOLS,
_LLM_TOOLS,
)
from langchain.agents.tools import Tool
from langflow.interface.tools.constants import CUSTOM_TOOLS, OTHER_TOOLS
def get_tools_dict():
"""Get the tools dictionary."""
return {
**_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()},
**CUSTOM_TOOLS,
**OTHER_TOOLS,
}
def get_tool_by_name(name: str):
"""Get a tool from the tools dictionary."""
tools = get_tools_dict()
if name not in tools:
raise ValueError(f"{name} not found.")
return tools[name]
def get_func_tool_params(func, **kwargs) -> Union[Dict, None]:
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 isinstance(tool.func, ast.Name) and 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_tool_params(cls, **kwargs) -> Union[Dict, None]:
tree = ast.parse(inspect.getsource(cls))
tool_params = {}
# Iterate over the statements in the abstract syntax tree
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
# Find the class definition and look for methods
for stmt in node.body:
if isinstance(stmt, ast.FunctionDef) and stmt.name == "__init__":
# There is no assignment statements in the __init__ method
# So we need to get the params from the function definition
for arg in stmt.args.args:
if arg.arg == "name":
# It should be the name of the class
tool_params[arg.arg] = cls.__name__
elif arg.arg == "self":
continue
# If there is not default value, set it to an empty string
else:
try:
annotation = ast.literal_eval(arg.annotation) # type: ignore
tool_params[arg.arg] = annotation
except ValueError:
tool_params[arg.arg] = ""
# Get the attribute name and the annotation
elif cls != Tool and isinstance(stmt, ast.AnnAssign):
# Get the attribute name and the annotation
tool_params[stmt.target.id] = "" # type: ignore
return tool_params
def get_tool_params(tool, **kwargs) -> Dict:
# Parse the function code into an abstract syntax tree
# Define if it is a function or a class
if inspect.isfunction(tool):
return get_func_tool_params(tool, **kwargs) or {}
elif inspect.isclass(tool):
# Get the parameters necessary to
# instantiate the class
return get_class_tool_params(tool, **kwargs) or {}
else:
raise ValueError("Tool must be a function or class.")

View file

@ -2,8 +2,6 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from langflow.api.endpoints import router as endpoints_router from langflow.api.endpoints import router as endpoints_router
from langflow.api.list_endpoints import router as list_router
from langflow.api.signature import router as signatures_router
def create_app(): def create_app():
@ -23,8 +21,6 @@ def create_app():
) )
app.include_router(endpoints_router) app.include_router(endpoints_router)
app.include_router(list_router)
app.include_router(signatures_router)
return app return app

View file

@ -1,17 +1,7 @@
import ast
import importlib import importlib
import inspect import inspect
import re import re
from typing import Dict, Optional, Union from typing import Dict, Optional
from langchain.agents.load_tools import (
_BASE_TOOLS,
_EXTRA_LLM_TOOLS,
_EXTRA_OPTIONAL_TOOLS,
_LLM_TOOLS,
)
from langchain.agents.tools import Tool
from langflow.utils import constants from langflow.utils import constants
@ -190,119 +180,6 @@ def get_default_factory(module: str, function: str):
return None return None
def get_tools_dict():
"""Get the tools dictionary."""
from langflow.interface.listing import CUSTOM_TOOLS, OTHER_TOOLS
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()},
**CUSTOM_TOOLS,
**OTHER_TOOLS,
}
return tools
def get_tool_by_name(name: str):
"""Get a tool from the tools dictionary."""
tools = get_tools_dict()
if name not in tools:
raise ValueError(f"{name} not found.")
return tools[name]
def get_tool_params(tool, **kwargs) -> Dict:
# Parse the function code into an abstract syntax tree
# Define if it is a function or a class
if inspect.isfunction(tool):
return get_func_tool_params(tool, **kwargs) or {}
elif inspect.isclass(tool):
# Get the parameters necessary to
# instantiate the class
return get_class_tool_params(tool, **kwargs) or {}
else:
raise ValueError("Tool must be a function or class.")
def get_func_tool_params(func, **kwargs) -> Union[Dict, None]:
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 isinstance(tool.func, ast.Name) and 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_tool_params(cls, **kwargs) -> Union[Dict, None]:
tree = ast.parse(inspect.getsource(cls))
tool_params = {}
# Iterate over the statements in the abstract syntax tree
for node in ast.walk(tree):
if isinstance(node, ast.ClassDef):
# Find the class definition and look for methods
for stmt in node.body:
if isinstance(stmt, ast.FunctionDef) and stmt.name == "__init__":
# There is no assignment statements in the __init__ method
# So we need to get the params from the function definition
for arg in stmt.args.args:
if arg.arg == "name":
# It should be the name of the class
tool_params[arg.arg] = cls.__name__
elif arg.arg == "self":
continue
# If there is not default value, set it to an empty string
else:
try:
annotation = ast.literal_eval(arg.annotation) # type: ignore
tool_params[arg.arg] = annotation
except ValueError:
tool_params[arg.arg] = ""
# Get the attribute name and the annotation
elif cls != Tool and isinstance(stmt, ast.AnnAssign):
# Get the attribute name and the annotation
tool_params[stmt.target.id] = "" # type: ignore
return tool_params
def get_class_doc(class_name): def get_class_doc(class_name):
""" """
Extracts information from the docstring of a given class. Extracts information from the docstring of a given class.

View file

@ -1,4 +1,4 @@
from langflow.interface.listing import CUSTOM_TOOLS from langflow.interface.tools.constants import CUSTOM_TOOLS
from fastapi.testclient import TestClient from fastapi.testclient import TestClient