🔥 refactor(conftest.py): remove unused fixtures and custom chain classes
The following changes were made: - Removed the `custom_chain` fixture and the `MyCustomChain` and `CustomChain` classes as they were not being used. - Removed the `data_processing`, `filter_docs`, `get_request`, and `post_request` fixtures as they were not being used. 🔧 fix(test_agents_template.py): set "dynamic" property to False for all template variables to ensure consistency and improve clarity 🐛 fix(test_chains_template.py): add missing "dynamic" field to template dictionaries to ensure consistency and avoid potential bugs 🔧 fix(test_custom_component.py): fix import statements and remove unused imports to improve code readability and maintainability ✨ feat(test_custom_component.py): add tests for the initialization of the CodeParser, Component, and CustomComponent classes 🔧 fix(test_custom_component.py): fix test names and add missing test cases for the Component and CustomComponent classes 🔨 refactor: refactor server.ts to use uppercase PORT variable for improved semantics ✨ feat: add support for process.env.PORT environment variable to run app on configurable port 🔨 refactor: refactor CustomComponent tests for improved readability and maintainability 🔨 refactor: refactor CodeParser tests for improved readability and maintainability 🔨 refactor: refactor Component tests for improved readability and maintainability 🐛 fix: fix CustomComponent class template validation to raise HTTPException when code is None 🔧 fix(tests): fix syntax error in custom_component._class_template_validation ✨ feat(tests): add test_custom_component_get_code_tree_syntax_error to test CustomComponent.get_code_tree method for raising CodeSyntaxError when given incorrect syntax ✨ feat(tests): add test_custom_component_get_function_entrypoint_args_no_args to test CustomComponent.get_function_entrypoint_args property with a build method with no arguments ✨ feat(tests): add test_custom_component_get_function_entrypoint_return_type_no_return_type to test CustomComponent.get_function_entrypoint_return_type property with a build method with no return type ✨ feat(tests): add test_custom_component_get_main_class_name_no_main_class to test CustomComponent.get_main_class_name property when there is no main class ✨ feat(tests): add test_custom_component_build_not_implemented to test CustomComponent.build method for raising NotImplementedError ✨ feat(tests): add fixtures for custom_chain, data_processing, filter_docs, and get_request 🔧 fix(tests): remove commented out code and unused imports to improve code readability and maintainability 🐛 fix(test_llms_template.py): set "dynamic" property to False for all template properties to ensure static values are used 🐛 fix(test_prompts_template.py): set "dynamic" property to False for all template properties to ensure consistency and improve readability
This commit is contained in:
gustavoschaedler
parent
9788ca6c2e
commit
11f7846e0a
7 changed files with 943 additions and 394 deletions
|
|
@ -116,254 +116,3 @@ def client_fixture(session: Session):
|
|||
|
||||
yield TestClient(app)
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def custom_chain():
|
||||
return '''
|
||||
from __future__ import annotations
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from pydantic import Extra
|
||||
|
||||
from langchain.schema import BaseLanguageModel, Document
|
||||
from langchain.callbacks.manager import (
|
||||
AsyncCallbackManagerForChainRun,
|
||||
CallbackManagerForChainRun,
|
||||
)
|
||||
from langchain.chains.base import Chain
|
||||
from langchain.prompts import StringPromptTemplate
|
||||
from langflow.interface.custom.base import CustomComponent
|
||||
|
||||
class MyCustomChain(Chain):
|
||||
"""
|
||||
An example of a custom chain.
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from pydantic import Extra
|
||||
|
||||
from langchain.schema import BaseLanguageModel, Document
|
||||
from langchain.callbacks.manager import (
|
||||
AsyncCallbackManagerForChainRun,
|
||||
CallbackManagerForChainRun,
|
||||
)
|
||||
from langchain.chains.base import Chain
|
||||
from langchain.prompts import StringPromptTemplate
|
||||
from langflow.interface.custom.base import CustomComponent
|
||||
|
||||
class MyCustomChain(Chain):
|
||||
"""
|
||||
An example of a custom chain.
|
||||
"""
|
||||
|
||||
prompt: StringPromptTemplate
|
||||
"""Prompt object to use."""
|
||||
llm: BaseLanguageModel
|
||||
output_key: str = "text" #: :meta private:
|
||||
|
||||
class Config:
|
||||
"""Configuration for this pydantic object."""
|
||||
|
||||
extra = Extra.forbid
|
||||
arbitrary_types_allowed = True
|
||||
|
||||
@property
|
||||
def input_keys(self) -> List[str]:
|
||||
"""Will be whatever keys the prompt expects.
|
||||
|
||||
:meta private:
|
||||
"""
|
||||
return self.prompt.input_variables
|
||||
|
||||
@property
|
||||
def output_keys(self) -> List[str]:
|
||||
"""Will always return text key.
|
||||
|
||||
:meta private:
|
||||
"""
|
||||
return [self.output_key]
|
||||
|
||||
def _call(
|
||||
self,
|
||||
inputs: Dict[str, Any],
|
||||
run_manager: Optional[CallbackManagerForChainRun] = None,
|
||||
) -> Dict[str, str]:
|
||||
# Your custom chain logic goes here
|
||||
# This is just an example that mimics LLMChain
|
||||
prompt_value = self.prompt.format_prompt(**inputs)
|
||||
|
||||
# Whenever you call a language model, or another chain, you should pass
|
||||
# a callback manager to it. This allows the inner run to be tracked by
|
||||
# any callbacks that are registered on the outer run.
|
||||
# You can always obtain a callback manager for this by calling
|
||||
# `run_manager.get_child()` as shown below.
|
||||
response = self.llm.generate_prompt(
|
||||
[prompt_value],
|
||||
callbacks=run_manager.get_child() if run_manager else None,
|
||||
)
|
||||
|
||||
# If you want to log something about this run, you can do so by calling
|
||||
# methods on the `run_manager`, as shown below. This will trigger any
|
||||
# callbacks that are registered for that event.
|
||||
if run_manager:
|
||||
run_manager.on_text("Log something about this run")
|
||||
|
||||
return {self.output_key: response.generations[0][0].text}
|
||||
|
||||
async def _acall(
|
||||
self,
|
||||
inputs: Dict[str, Any],
|
||||
run_manager: Optional[AsyncCallbackManagerForChainRun] = None,
|
||||
) -> Dict[str, str]:
|
||||
# Your custom chain logic goes here
|
||||
# This is just an example that mimics LLMChain
|
||||
prompt_value = self.prompt.format_prompt(**inputs)
|
||||
|
||||
# Whenever you call a language model, or another chain, you should pass
|
||||
# a callback manager to it. This allows the inner run to be tracked by
|
||||
# any callbacks that are registered on the outer run.
|
||||
# You can always obtain a callback manager for this by calling
|
||||
# `run_manager.get_child()` as shown below.
|
||||
response = await self.llm.agenerate_prompt(
|
||||
[prompt_value],
|
||||
callbacks=run_manager.get_child() if run_manager else None,
|
||||
)
|
||||
|
||||
# If you want to log something about this run, you can do so by calling
|
||||
# methods on the `run_manager`, as shown below. This will trigger any
|
||||
# callbacks that are registered for that event.
|
||||
if run_manager:
|
||||
await run_manager.on_text("Log something about this run")
|
||||
|
||||
return {self.output_key: response.generations[0][0].text}
|
||||
|
||||
@property
|
||||
def _chain_type(self) -> str:
|
||||
return "my_custom_chain"
|
||||
|
||||
class CustomChain(CustomComponent):
|
||||
display_name: str = "Custom Chain"
|
||||
field_config = {
|
||||
"prompt": {"field_type": "prompt"},
|
||||
"llm": {"field_type": "BaseLanguageModel"},
|
||||
}
|
||||
|
||||
def build(self, prompt, llm, input: str) -> Document:
|
||||
chain = MyCustomChain(prompt=prompt, llm=llm)
|
||||
return chain(input)
|
||||
'''
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def data_processing():
|
||||
return """
|
||||
import pandas as pd
|
||||
from langchain.schema import Document
|
||||
from langflow.interface.custom.base import CustomComponent
|
||||
|
||||
class CSVLoaderComponent(CustomComponent):
|
||||
display_name: str = "CSV Loader"
|
||||
field_config = {
|
||||
"filename": {"field_type": "str", "required": True},
|
||||
"column_name": {"field_type": "str", "required": True},
|
||||
}
|
||||
|
||||
def build(self, filename: str, column_name: str) -> Document:
|
||||
# Load the CSV file
|
||||
df = pd.read_csv(filename)
|
||||
|
||||
# Verify the column exists
|
||||
if column_name not in df.columns:
|
||||
raise ValueError(f"Column '{column_name}' not found in the CSV file")
|
||||
|
||||
# Convert each row of the specified column to a document object
|
||||
documents = []
|
||||
for content in df[column_name]:
|
||||
metadata = {"filename": filename}
|
||||
documents.append(Document(page_content=str(content), metadata=metadata))
|
||||
|
||||
return documents
|
||||
"""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def filter_docs():
|
||||
return """
|
||||
from langchain.schema import Document
|
||||
from langflow.interface.custom.base import CustomComponent
|
||||
from typing import List
|
||||
|
||||
class DocumentFilterByLengthComponent(CustomComponent):
|
||||
display_name: str = "Document Filter By Length"
|
||||
field_config = {
|
||||
"documents": {"field_type": "Document", "required": True},
|
||||
"max_length": {"field_type": "int", "required": True},
|
||||
}
|
||||
|
||||
def build(self, documents: List[Document], max_length: int) -> List[Document]:
|
||||
# Filter the documents by length
|
||||
filtered_documents = [doc for doc in documents if len(doc.page_content) <= max_length]
|
||||
|
||||
return filtered_documents
|
||||
"""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def get_request():
|
||||
return """
|
||||
import requests
|
||||
from typing import Dict, Union
|
||||
from langchain.schema import Document
|
||||
from langflow.interface.custom.base import CustomComponent
|
||||
|
||||
class GetRequestComponent(CustomComponent):
|
||||
display_name: str = "GET Request"
|
||||
field_config = {
|
||||
"url": {"field_type": "str", "required": True},
|
||||
}
|
||||
|
||||
def build(self, url: str) -> Document:
|
||||
# Send a GET request to the URL
|
||||
response = requests.get(url)
|
||||
|
||||
# Raise an exception if the request was not successful
|
||||
if response.status_code != 200:
|
||||
raise ValueError(f"GET request failed: {response.status_code} status code")
|
||||
|
||||
# Create a document with the response text and the URL as metadata
|
||||
document = Document(page_content=response.text, metadata={"url": url})
|
||||
|
||||
return document
|
||||
"""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def post_request():
|
||||
return """
|
||||
import requests
|
||||
from typing import Dict, Union
|
||||
from langchain.schema import Document
|
||||
from langflow.interface.custom.base import CustomComponent
|
||||
|
||||
class PostRequestComponent(CustomComponent):
|
||||
display_name: str = "POST Request"
|
||||
field_config = {
|
||||
"url": {"field_type": "str", "required": True},
|
||||
"data": {"field_type": "dict", "required": True},
|
||||
}
|
||||
|
||||
def build(self, url: str, data: Dict[str, Union[str, int]]) -> Document:
|
||||
# Send a POST request to the URL
|
||||
response = requests.post(url, data=data)
|
||||
|
||||
# Raise an exception if the request was not successful
|
||||
if response.status_code != 200:
|
||||
raise ValueError(f"POST request failed: {response.status_code} status code")
|
||||
|
||||
# Create a document with the response text and the URL and data as metadata
|
||||
document = Document(page_content=response.text, metadata={"url": url, "data": data})
|
||||
|
||||
return document
|
||||
"""
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue