Skip to content

Commit

Permalink
Merge branch 'feat/workflow-continue-on-error' into deploy/dev
Browse files Browse the repository at this point in the history
  • Loading branch information
zxhlyh committed Dec 5, 2024
2 parents 8517b62 + ec507a1 commit 946375c
Show file tree
Hide file tree
Showing 15 changed files with 36 additions and 22 deletions.
8 changes: 4 additions & 4 deletions api/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -259,15 +259,15 @@ def migrate_knowledge_vector_database():
skipped_count = 0
total_count = 0
vector_type = dify_config.VECTOR_STORE
upper_colletion_vector_types = {
upper_collection_vector_types = {
VectorType.MILVUS,
VectorType.PGVECTOR,
VectorType.RELYT,
VectorType.WEAVIATE,
VectorType.ORACLE,
VectorType.ELASTICSEARCH,
}
lower_colletion_vector_types = {
lower_collection_vector_types = {
VectorType.ANALYTICDB,
VectorType.CHROMA,
VectorType.MYSCALE,
Expand Down Expand Up @@ -307,7 +307,7 @@ def migrate_knowledge_vector_database():
continue
collection_name = ""
dataset_id = dataset.id
if vector_type in upper_colletion_vector_types:
if vector_type in upper_collection_vector_types:
collection_name = Dataset.gen_collection_name_by_id(dataset_id)
elif vector_type == VectorType.QDRANT:
if dataset.collection_binding_id:
Expand All @@ -323,7 +323,7 @@ def migrate_knowledge_vector_database():
else:
collection_name = Dataset.gen_collection_name_by_id(dataset_id)

elif vector_type in lower_colletion_vector_types:
elif vector_type in lower_collection_vector_types:
collection_name = Dataset.gen_collection_name_by_id(dataset_id).lower()
else:
raise ValueError(f"Vector store {vector_type} is not supported.")
Expand Down
2 changes: 1 addition & 1 deletion api/core/app/apps/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

Due to the presence of tasks in App Runner that require long execution times, such as LLM generation and external requests, Flask-Sqlalchemy's strategy for database connection pooling is to allocate one connection (transaction) per request. This approach keeps a connection occupied even during non-DB tasks, leading to the inability to acquire new connections during high concurrency requests due to multiple long-running tasks.

Therefore, the database operations in App Runner and Task Pipeline must ensure connections are closed immediately after use, and it's better to pass IDs rather than Model objects to avoid deattach errors.
Therefore, the database operations in App Runner and Task Pipeline must ensure connections are closed immediately after use, and it's better to pass IDs rather than Model objects to avoid detach errors.

Examples:

Expand Down
2 changes: 1 addition & 1 deletion api/core/model_runtime/docs/zh_Hans/interfaces.md
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ class XinferenceProvider(Provider):
"""
```

也可以直接抛出对应Erros,并做如下定义,这样在之后的调用中可以直接抛出`InvokeConnectionError`等异常。
也可以直接抛出对应 Errors,并做如下定义,这样在之后的调用中可以直接抛出`InvokeConnectionError`等异常。

```python
@property
Expand Down
Original file line number Diff line number Diff line change
@@ -1,2 +1,2 @@
- amazon.rerank-v1
- cohere.rerank-v3-5:0
- cohere.rerank-v3-5
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ def _invoke(
params["input"] = query
else:
finished = True
if "souces" in response_data["queryresult"]:
if "sources" in response_data["queryresult"]:
return self.create_link_message(response_data["queryresult"]["sources"]["url"])
elif "pods" in response_data["queryresult"]:
result = response_data["queryresult"]["pods"][0]["subpods"][0]["plaintext"]
Expand Down
10 changes: 5 additions & 5 deletions api/core/workflow/nodes/if_else/if_else_node.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ def _run(self) -> NodeRunResult:
"""
node_inputs: dict[str, list] = {"conditions": []}

process_datas: dict[str, list] = {"condition_results": []}
process_data: dict[str, list] = {"condition_results": []}

input_conditions = []
final_result = False
Expand All @@ -40,7 +40,7 @@ def _run(self) -> NodeRunResult:
operator=case.logical_operator,
)

process_datas["condition_results"].append(
process_data["condition_results"].append(
{
"group": case.model_dump(),
"results": group_result,
Expand All @@ -65,23 +65,23 @@ def _run(self) -> NodeRunResult:

selected_case_id = "true" if final_result else "false"

process_datas["condition_results"].append(
process_data["condition_results"].append(
{"group": "default", "results": group_result, "final_result": final_result}
)

node_inputs["conditions"] = input_conditions

except Exception as e:
return NodeRunResult(
status=WorkflowNodeExecutionStatus.FAILED, inputs=node_inputs, process_data=process_datas, error=str(e)
status=WorkflowNodeExecutionStatus.FAILED, inputs=node_inputs, process_data=process_data, error=str(e)
)

outputs = {"result": final_result, "selected_case_id": selected_case_id}

data = NodeRunResult(
status=WorkflowNodeExecutionStatus.SUCCEEDED,
inputs=node_inputs,
process_data=process_datas,
process_data=process_data,
edge_source_handle=selected_case_id or "false", # Use case ID or 'default'
outputs=outputs,
)
Expand Down
4 changes: 2 additions & 2 deletions api/core/workflow/nodes/variable_assigner/v2/exc.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,8 @@


class OperationNotSupportedError(VariableOperatorNodeError):
def __init__(self, *, operation: Operation, varialbe_type: str):
super().__init__(f"Operation {operation} is not supported for type {varialbe_type}")
def __init__(self, *, operation: Operation, variable_type: str):
super().__init__(f"Operation {operation} is not supported for type {variable_type}")


class InputTypeNotSupportedError(VariableOperatorNodeError):
Expand Down
4 changes: 2 additions & 2 deletions api/core/workflow/nodes/variable_assigner/v2/node.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ def _run(self) -> NodeRunResult:

# Check if operation is supported
if not helpers.is_operation_supported(variable_type=variable.value_type, operation=item.operation):
raise OperationNotSupportedError(operation=item.operation, varialbe_type=variable.value_type)
raise OperationNotSupportedError(operation=item.operation, variable_type=variable.value_type)

# Check if variable input is supported
if item.input_type == InputType.VARIABLE and not helpers.is_variable_input_supported(
Expand Down Expand Up @@ -156,4 +156,4 @@ def _handle_item(
case Operation.DIVIDE:
return variable.value / value
case _:
raise OperationNotSupportedError(operation=operation, varialbe_type=variable.value_type)
raise OperationNotSupportedError(operation=operation, variable_type=variable.value_type)
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ import type {
Node,
} from '@/app/components/workflow/types'
import Split from '@/app/components/workflow/nodes/_base/components/split'
import Tooltip from '@/app/components/base/tooltip'

type ErrorHandleProps = Pick<Node, 'id' | 'data'>
const ErrorHandle = ({
Expand All @@ -27,7 +28,7 @@ const ErrorHandle = ({
collapsed,
setCollapsed,
handleErrorHandleTypeChange,
} = useErrorHandle(id)
} = useErrorHandle(id, data)
const { handleFormChange } = useDefaultValue(id)

const getHandleErrorHandleTypeChange = useCallback((data: CommonNodeType) => {
Expand All @@ -52,8 +53,11 @@ const ErrorHandle = ({
onCollapse={setCollapsed}
trigger={
<div className='grow flex items-center justify-between pr-4'>
<div className='system-sm-semibold-uppercase text-text-secondary'>
{t('workflow.nodes.common.errorHandle.title')}
<div className='flex items-center'>
<div className='mr-0.5 system-sm-semibold-uppercase text-text-secondary'>
{t('workflow.nodes.common.errorHandle.title')}
</div>
<Tooltip popupContent={t('workflow.nodes.common.errorHandle.tip')} />
</div>
<ErrorHandleTypeSelector
value={error_strategy || ErrorHandleTypeEnum.none}
Expand Down
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import {
useCallback,
useMemo,
useState,
} from 'react'
import { ErrorHandleTypeEnum } from './types'
Expand Down Expand Up @@ -62,8 +63,15 @@ export const useDefaultValue = (

export const useErrorHandle = (
id: string,
data: CommonNodeType,
) => {
const [collapsed, setCollapsed] = useState(true)
const initCollapsed = useMemo(() => {
if (data.error_strategy === ErrorHandleTypeEnum.none)
return true

return false
}, [data.error_strategy])
const [collapsed, setCollapsed] = useState(initCollapsed)
const { handleNodeDataUpdateWithSyncDraft } = useNodeDataUpdate()
const { handleEdgeDeleteByDeleteBranch } = useEdgesInteractions()

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ const NodeVariableItem = ({
const { t } = useTranslation()
return (
<div className={cn(
'relative flex items-center p-[3px] pl-[5px] gap-1 self-stretch rounded-md bg-workflow-block-param-bg',
'relative flex items-center p-[3px] pl-[5px] gap-1 self-stretch rounded-md bg-workflow-block-parma-bg',
showBorder && '!bg-black/[0.02]',
className,
)}>
Expand Down
1 change: 1 addition & 0 deletions web/i18n/en-US/workflow.ts
Original file line number Diff line number Diff line change
Expand Up @@ -305,6 +305,7 @@ const translation = {
},
errorHandle: {
title: 'Error Handling',
tip: 'Exception handling strategy, triggered when a node encounters an exception.',
none: {
title: 'None',
desc: 'The node will stop running if an exception occurs and is not handled',
Expand Down
1 change: 1 addition & 0 deletions web/i18n/zh-Hans/workflow.ts
Original file line number Diff line number Diff line change
Expand Up @@ -305,6 +305,7 @@ const translation = {
},
errorHandle: {
title: '异常处理',
tip: '配置异常处理策略,当节点发生异常时触发。',
none: {
title: '无',
desc: '当发生异常且未处理时,节点将停止运行',
Expand Down

0 comments on commit 946375c

Please sign in to comment.