refactor: rename variable to be more understandable
This commit is contained in:
parent
2fb06e0729
commit
2e1bfe78b1
4 changed files with 11 additions and 11 deletions
|
|
@ -41,7 +41,7 @@ async def add(
|
||||||
extraction_rules: Optional[Dict[str, Any]] = None,
|
extraction_rules: Optional[Dict[str, Any]] = None,
|
||||||
tavily_config: Optional[BaseModel] = None,
|
tavily_config: Optional[BaseModel] = None,
|
||||||
soup_crawler_config: Optional[BaseModel] = None,
|
soup_crawler_config: Optional[BaseModel] = None,
|
||||||
data_batch_size: Optional[int] = 20,
|
data_per_batch: Optional[int] = 20,
|
||||||
):
|
):
|
||||||
"""
|
"""
|
||||||
Add data to Cognee for knowledge graph processing.
|
Add data to Cognee for knowledge graph processing.
|
||||||
|
|
@ -236,7 +236,7 @@ async def add(
|
||||||
vector_db_config=vector_db_config,
|
vector_db_config=vector_db_config,
|
||||||
graph_db_config=graph_db_config,
|
graph_db_config=graph_db_config,
|
||||||
incremental_loading=incremental_loading,
|
incremental_loading=incremental_loading,
|
||||||
data_batch_size=data_batch_size,
|
data_per_batch=data_per_batch,
|
||||||
):
|
):
|
||||||
pipeline_run_info = run_info
|
pipeline_run_info = run_info
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -51,7 +51,7 @@ async def cognify(
|
||||||
incremental_loading: bool = True,
|
incremental_loading: bool = True,
|
||||||
custom_prompt: Optional[str] = None,
|
custom_prompt: Optional[str] = None,
|
||||||
temporal_cognify: bool = False,
|
temporal_cognify: bool = False,
|
||||||
data_batch_size: int = 20,
|
data_per_batch: int = 20,
|
||||||
):
|
):
|
||||||
"""
|
"""
|
||||||
Transform ingested data into a structured knowledge graph.
|
Transform ingested data into a structured knowledge graph.
|
||||||
|
|
@ -229,7 +229,7 @@ async def cognify(
|
||||||
graph_db_config=graph_db_config,
|
graph_db_config=graph_db_config,
|
||||||
incremental_loading=incremental_loading,
|
incremental_loading=incremental_loading,
|
||||||
pipeline_name="cognify_pipeline",
|
pipeline_name="cognify_pipeline",
|
||||||
data_batch_size=data_batch_size,
|
data_per_batch=data_per_batch,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -35,7 +35,7 @@ async def run_pipeline(
|
||||||
vector_db_config: dict = None,
|
vector_db_config: dict = None,
|
||||||
graph_db_config: dict = None,
|
graph_db_config: dict = None,
|
||||||
incremental_loading: bool = False,
|
incremental_loading: bool = False,
|
||||||
data_batch_size: int = 20,
|
data_per_batch: int = 20,
|
||||||
):
|
):
|
||||||
validate_pipeline_tasks(tasks)
|
validate_pipeline_tasks(tasks)
|
||||||
await setup_and_check_environment(vector_db_config, graph_db_config)
|
await setup_and_check_environment(vector_db_config, graph_db_config)
|
||||||
|
|
@ -51,7 +51,7 @@ async def run_pipeline(
|
||||||
pipeline_name=pipeline_name,
|
pipeline_name=pipeline_name,
|
||||||
context={"dataset": dataset},
|
context={"dataset": dataset},
|
||||||
incremental_loading=incremental_loading,
|
incremental_loading=incremental_loading,
|
||||||
data_batch_size=data_batch_size,
|
data_per_batch=data_per_batch,
|
||||||
):
|
):
|
||||||
yield run_info
|
yield run_info
|
||||||
|
|
||||||
|
|
@ -64,7 +64,7 @@ async def run_pipeline_per_dataset(
|
||||||
pipeline_name: str = "custom_pipeline",
|
pipeline_name: str = "custom_pipeline",
|
||||||
context: dict = None,
|
context: dict = None,
|
||||||
incremental_loading=False,
|
incremental_loading=False,
|
||||||
data_batch_size: int = 20,
|
data_per_batch: int = 20,
|
||||||
):
|
):
|
||||||
# Will only be used if ENABLE_BACKEND_ACCESS_CONTROL is set to True
|
# Will only be used if ENABLE_BACKEND_ACCESS_CONTROL is set to True
|
||||||
await set_database_global_context_variables(dataset.id, dataset.owner_id)
|
await set_database_global_context_variables(dataset.id, dataset.owner_id)
|
||||||
|
|
@ -80,7 +80,7 @@ async def run_pipeline_per_dataset(
|
||||||
return
|
return
|
||||||
|
|
||||||
pipeline_run = run_tasks(
|
pipeline_run = run_tasks(
|
||||||
tasks, dataset.id, data, user, pipeline_name, context, incremental_loading, data_batch_size
|
tasks, dataset.id, data, user, pipeline_name, context, incremental_loading, data_per_batch
|
||||||
)
|
)
|
||||||
|
|
||||||
async for pipeline_run_info in pipeline_run:
|
async for pipeline_run_info in pipeline_run:
|
||||||
|
|
|
||||||
|
|
@ -59,7 +59,7 @@ async def run_tasks(
|
||||||
pipeline_name: str = "unknown_pipeline",
|
pipeline_name: str = "unknown_pipeline",
|
||||||
context: dict = None,
|
context: dict = None,
|
||||||
incremental_loading: bool = False,
|
incremental_loading: bool = False,
|
||||||
data_batch_size: int = 20,
|
data_per_batch: int = 20,
|
||||||
):
|
):
|
||||||
if not user:
|
if not user:
|
||||||
user = await get_default_user()
|
user = await get_default_user()
|
||||||
|
|
@ -91,8 +91,8 @@ async def run_tasks(
|
||||||
|
|
||||||
# Create and gather batches of async tasks of data items that will run the pipeline for the data item
|
# Create and gather batches of async tasks of data items that will run the pipeline for the data item
|
||||||
results = []
|
results = []
|
||||||
for start in range(0, len(data), data_batch_size):
|
for start in range(0, len(data), data_per_batch):
|
||||||
data_batch = data[start : start + data_batch_size]
|
data_batch = data[start : start + data_per_batch]
|
||||||
|
|
||||||
data_item_tasks = [
|
data_item_tasks = [
|
||||||
asyncio.create_task(
|
asyncio.create_task(
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue