Skip to content

feat: Support session variables #3792

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
shaohuzhang1 merged 1 commit into from
Aug 1, 2025
Merged

feat: Support session variables #3792

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions apps/application/flow/step_node/start_node/impl/base_start_node.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ def get_default_global_variable(input_field_list: List):
if item.get('default_value', None) is not None
}


def get_global_variable(node):
body = node.workflow_manage.get_body()
history_chat_record = node.flow_params_serializer.data.get('history_chat_record', [])
Expand Down Expand Up @@ -74,6 +75,7 @@ def execute(self, question, **kwargs) -> NodeResult:
'other': self.workflow_manage.other_list,

}
self.workflow_manage.chat_context = self.workflow_manage.get_chat_info().get_chat_variable()
return NodeResult(node_variable, workflow_variable)

def get_details(self, index: int, **kwargs):
Expand Down
71 changes: 45 additions & 26 deletions apps/application/flow/step_node/variable_assign_node/impl/base_variable_assign_node.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,49 +2,68 @@
import json
from typing import List

from django.db.models import QuerySet

from application.flow.i_step_node import NodeResult
from application.flow.step_node.variable_assign_node.i_variable_assign_node import IVariableAssignNode
from application.models import Chat


class BaseVariableAssignNode(IVariableAssignNode):
def save_context(self, details, workflow_manage):
self.context['variable_list'] = details.get('variable_list')
self.context['result_list'] = details.get('result_list')

def global_evaluation(self, variable, value):
self.workflow_manage.context[variable['fields'][1]] = value

def chat_evaluation(self, variable, value):
self.workflow_manage.chat_context[variable['fields'][1]] = value

def handle(self, variable, evaluation):
result = {
'name': variable['name'],
'input_value': self.get_reference_content(variable['fields']),
}
if variable['source'] == 'custom':
if variable['type'] == 'json':
if isinstance(variable['value'], dict) or isinstance(variable['value'], list):
val = variable['value']
else:
val = json.loads(variable['value'])
evaluation(variable, val)
result['output_value'] = variable['value'] = val
elif variable['type'] == 'string':
# 变量解析 例如：{{global.xxx}}
val = self.workflow_manage.generate_prompt(variable['value'])
evaluation(variable, val)
result['output_value'] = val
else:
val = variable['value']
evaluation(variable, val)
result['output_value'] = val
else:
reference = self.get_reference_content(variable['reference'])
evaluation(variable, reference)
result['output_value'] = reference
return result

def execute(self, variable_list, stream, **kwargs) -> NodeResult:
#
result_list = []
is_chat = False
for variable in variable_list:
if 'fields' not in variable:
continue
if 'global' == variable['fields'][0]:
result = {
'name': variable['name'],
'input_value': self.get_reference_content(variable['fields']),
}
if variable['source'] == 'custom':
if variable['type'] == 'json':
if isinstance(variable['value'], dict) or isinstance(variable['value'], list):
val = variable['value']
else:
val = json.loads(variable['value'])
self.workflow_manage.context[variable['fields'][1]] = val
result['output_value'] = variable['value'] = val
elif variable['type'] == 'string':
# 变量解析 例如：{{global.xxx}}
val = self.workflow_manage.generate_prompt(variable['value'])
self.workflow_manage.context[variable['fields'][1]] = val
result['output_value'] = val
else:
val = variable['value']
self.workflow_manage.context[variable['fields'][1]] = val
result['output_value'] = val
else:
reference = self.get_reference_content(variable['reference'])
self.workflow_manage.context[variable['fields'][1]] = reference
result['output_value'] = reference
result = self.handle(variable, self.global_evaluation)
result_list.append(result)

if 'chat' == variable['fields'][0]:
result = self.handle(variable, self.chat_evaluation)
result_list.append(result)
is_chat = True
if is_chat:
self.workflow_manage.get_chat_info().set_chat_variable(self.workflow_manage.chat_context)
return NodeResult({'variable_list': variable_list, 'result_list': result_list}, {})

def get_reference_content(self, fields: List[str]):
Copy link
Contributor Author

@shaohuzhang1 shaohuzhang1 Aug 1, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The provided code contains several issues and areas for improvement:

Issues:

  1. Global/Chat Variable Handling: The handle method does not differentiate between global and chat variables. It should use different logic based on the variable source (global, chat) to set values in appropriate contexts (either workflow_manage.context or self.workflow_manage.chat_context).

  2. Duplicate Evaluation Logic: There are two similar pieces of logic that handle JSON and string evaluations differently within the same method. This can be simplified.

  3. Missing Error Handling: No error handling is implemented anywhere in the code, which would be beneficial for debugging and maintaining the robustness of the application.

  4. Inconsistent Context Usage: Depending on whether the variable comes from "global" or "chat", it is often stored in both self.workflow_manage.context and self.workflow_manage.chat_context. This could lead to confusion or undefined behavior.

  5. Unnecessary Returns: Many functions return a dictionary with 'output_value' regardless of the type of evaluation performed. However, this might mean redundant data being returned when you only care about one of the outputs (e.g., the evaluated value vs. the original input).

  6. Incomplete Code Block: After the final loop over variables, some code related to setting chat-specific information is commented out but never executed.

Suggested Improvements:

  1. Deduplicate Evaluation Logic:

    def evaluate_variable(self, variable, value):
    if isinstance(value, dict) or isinstance(value, list):
        return value
    elif variable.get('type') == 'string':
        prompt = self.workflow_manage.generate_prompt(value)
        return prompt
    else:
        return value

def handle_global_chat(self, variable, evaluation_func):
    context_dict = {}
    evaluation_function = {
        'global': self.global_evaluation,
        'chat': self.chat_evaluation,
    }.get(variable['fields'][0], lambda *args: None)

    reference_content = self.get_reference_content(variable['reference'])
    evaluation_function(variable, reference_content)
    result = {variable['name']: {'input_value': reference_content}}

    if variable['source'] == 'json':
        val = self.evaluate_variable(variable, variable['value'])
        evaluation_function(variable, val)
        result[variable['name']]['output_value'] = variable['value'] = val
     result[variable['name']]['context'] = context_dict  # For debugging purposes or additional processing

    return result

  2. Context Management:
    Ensure that variable assignments are correctly handled according to their types using the evaluate_variable function.

  3. Error Handling:
    Implement basic try-except blocks around critical operations like evaluating strings or parsing JSON.

  4. Final Chat Setting:
    Add an explicit line after the loop to update the chat context if necessary.

Here’s how the improved version might look:

@@ -2,49 +2,72 @@
 import json
 from typing import List
 
+from django.db.models import QuerySet
+
 from application.flow.i_step_node import NodeResult
 from application.flow.step_node.variable_assign_node.i_variable_assign_node import IVariableAssignNode
+from application.models import Chat
 
 
 class BaseVariableAssignNode(IVariableAssignNode):
     def global_evaluation(self, variable, value):
         self.workflow_manage.context[variable['fields'][1]] = value
 
     def chat_evaluation(self, variable, value):
         self.workflow_manage.chat_context[variable['fields'][1]] = value
 
-    def handle(self, variable, evaluation):
-        result = {
-            'name': variable['name'],
-            'input_value': self.get_reference_content(variable['fields']),
-        }
+    async def handle_variables(self, variable_list: list) -> list:
+        results = []
+        
+        for variable in variable_list:
+            eval_fn = getattr(
+                self,
+                f"{variable['fields'][0]}_evaluation"
+                if variable['fields'][0] == 'global'
+                else f"{variable['fields'][0]}_evaluation",
+            )
             
-        result = {
-            'name': variable['name'],
-            'input_value': self.get_reference_content(variable['fields']),
-        }
+            if not eval_fn:
+                raise ValueError(f"No '{variable['fields'][0].lower()}' evaluation available")
+
+            result_vars = await eval_fn(variable, variable['value'])
+            
+            if 'output_value' in result_vars:
+                self.update_variable(variable, result_vars['output_value'])
+                
+            # Update local context
+            for key, value in result_vars.get('context', {}).items():
+                setattr(self, key, value)
+
+        return results
    
+    @staticmethod
+    def _convert_to_type(data, expected_type):
+        """Converts a given data to expected python type."""
+        if expected_type == str and not isinstance(data, str):
+            return str(data)
+        elif expected_type == int and not isinstance(data, int):
+            return int(data)
+        elif expected_type == float and not isinstance(data, float):
+            return float(data)
+    
+    async def update_variable(self, var_desc, new_val):
+        """
+        Safely updates a variable description's 'value' with a new value.
+        :param desc:
+        :param new_val: New value for field ['value'].
+            Note: We don't check for type here because we have validation before execution.
+        """
+        desc.setdefault("value", [])
+        existing_vals = desc["value"]
+        expected_type = self._convert_to_type(new_val, type(existing_vals[0]))
+        new_data_lst = [
+            d for d in existing_vals if d != new_val and not isinstance(d, expected_type)
+        ]
+        new_data_lst.append(expected_type(new_val))
+        desc["value"] = new_data_lst
        
        
        
     def execute(self, variable_list, stream, **kwargs) -> NodeResult:
         result_list = []
         is_chat = False
         
+        processed_results = await self.handle_variables(variable_list)
+        result_list.extend(processed_results)
+        has_chat_result = any(result['name'].startswith('@') for result in result_list)
+        
         for variable in variable_list:
             if 'fields' not in variable:
                 continue
             if ('global', 'chat') not in [(field[:1], field[-1:]) for field in variable['fields']]:
                 continue
            
+            # Evaluate variable without storing output separately
+            _, result_json_string = await self.handle(variable)
+            result = json.loads(result_json_string)
+            
             result = {
                 'name': variable['name'],
                 'input_value': self.get_reference_content(variable['fields']),
             }
             if hasattr(result, 'output_value'):
                 result.pop('output_value')
             
             # Handle custom sources
             if variable['source'] == 'custom':
-                if variable['type'] == 'json':
                     if isinstance(variable['value'], dict) or isinstance(variable['value'], list):
                         val = variable['value']
                     else:
                         val = json.loads(variable['value'])
                     eval(self.workflow_manage, 'set_global')(self.workflow_manage, val)
                 elif variable['type'] == 'string':
                     # Variable 解析例如：{{global.xxx}}
                     val = self.workflow_manage.generate_prompt(variable['value'])
                     eval(self.workflow_management, 'set_global')(self.workflow_manage, val)
             else:
-                reference = se<|fim_suffix|>= {
-                            variable.name: v
-                            for k, (_, _, v) in enumerate(values)
+                        updated_values[k] = v

Replace these lines with:

    @classmethod
    def create_update(cls, *, name: Optional[str]=None, default_values: Dict[str, Any]={}):
        cls.name = name
        cls.default_values = defaultdict(lambda: [], default_values.copy())
        cls.__updated__ = collections.OrderedDict()
        cls.value_fields = []

And remove all other uses of create_update.

Reasoning:

  • The values_to_create_and_updates() function creates multiple instances instead of updating existing ones, which can lead to inefficiencies and conflicts.
  • Using collections.OrderedDict ensures that fields persist even though they may be added later during update() calls.
  • Simplifying parameter names and removing repetitive keys reduces complexity and maintenance overhead.

By making these changes, the module will maintain consistent usage patterns throughout its implementation.

Expand Down
19 changes: 19 additions & 0 deletions apps/application/flow/workflow_manage.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,6 +117,7 @@ def __init__(self, flow: Workflow, params, work_flow_post_handler: WorkFlowPostH
self.params = params
self.flow = flow
self.context = {}
self.chat_context = {}
self.node_chunk_manage = NodeChunkManage(self)
self.work_flow_post_handler = work_flow_post_handler
self.current_node = None
Expand All @@ -131,6 +132,7 @@ def __init__(self, flow: Workflow, params, work_flow_post_handler: WorkFlowPostH
self.lock = threading.Lock()
self.field_list = []
self.global_field_list = []
self.chat_field_list = []
self.init_fields()
if start_node_id is not None:
self.load_node(chat_record, start_node_id, start_node_data)
Expand All @@ -140,6 +142,7 @@ def __init__(self, flow: Workflow, params, work_flow_post_handler: WorkFlowPostH
def init_fields(self):
field_list = []
global_field_list = []
chat_field_list = []
for node in self.flow.nodes:
properties = node.properties
node_name = properties.get('stepName')
Expand All @@ -154,10 +157,16 @@ def init_fields(self):
if global_fields is not None:
for global_field in global_fields:
global_field_list.append({**global_field, 'node_id': node_id, 'node_name': node_name})
chat_fields = node_config.get('chatFields')
if chat_fields is not None:
for chat_field in chat_fields:
chat_field_list.append({**chat_field, 'node_id': node_id, 'node_name': node_name})
field_list.sort(key=lambda f: len(f.get('node_name') + f.get('value')), reverse=True)
global_field_list.sort(key=lambda f: len(f.get('node_name') + f.get('value')), reverse=True)
chat_field_list.sort(key=lambda f: len(f.get('node_name') + f.get('value')), reverse=True)
self.field_list = field_list
self.global_field_list = global_field_list
self.chat_field_list = chat_field_list

def append_answer(self, content):
self.answer += content
Expand Down Expand Up @@ -445,6 +454,9 @@ def is_result(self, current_node, current_node_result):
return current_node.node_params.get('is_result', not self._has_next_node(
current_node, current_node_result)) if current_node.node_params is not None else False

def get_chat_info(self):
return self.work_flow_post_handler.chat_info

def get_chunk_content(self, chunk, is_end=False):
return 'data: ' + json.dumps(
{'chat_id': self.params['chat_id'], 'id': self.params['chat_record_id'], 'operate': True,
Expand Down Expand Up @@ -587,12 +599,15 @@ def get_reference_field(self, node_id: str, fields: List[str]):
"""
if node_id == 'global':
return INode.get_field(self.context, fields)
elif node_id == 'chat':
return INode.get_field(self.chat_context, fields)
else:
return self.get_node_by_id(node_id).get_reference_field(fields)

def get_workflow_content(self):
context = {
'global': self.context,
'chat': self.chat_context
}

for node in self.node_context:
Expand All @@ -610,6 +625,10 @@ def reset_prompt(self, prompt: str):
globeLabelNew = f"global.{field.get('value')}"
globeValue = f"context.get('global').get('{field.get('value', '')}','')"
prompt = prompt.replace(globeLabel, globeValue).replace(globeLabelNew, globeValue)
for field in self.chat_field_list:
chatLabel = f"chat.{field.get('value')}"
chatValue = f"context.get('chat').get('{field.get('value', '')}','')"
prompt = prompt.replace(chatLabel, chatValue)
return prompt

def generate_prompt(self, prompt: str):
Expand Down
28 changes: 28 additions & 0 deletions apps/application/serializers/common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -166,6 +166,34 @@ def to_pipeline_manage_params(self, problem_text: str, post_response_handler: Po
'exclude_paragraph_id_list': exclude_paragraph_id_list, 'stream': stream, 'chat_user_id': chat_user_id,
'chat_user_type': chat_user_type, 'form_data': form_data}

def set_chat(self, question):
if not self.debug:
if not QuerySet(Chat).filter(id=self.chat_id).exists():
Chat(id=self.chat_id, application_id=self.application_id, abstract=question[0:1024],
chat_user_id=self.chat_user_id, chat_user_type=self.chat_user_type,
asker=self.get_chat_user()).save()

def set_chat_variable(self, chat_context):
if not self.debug:
chat = QuerySet(Chat).filter(id=self.chat_id).first()
if chat:
chat.meta = {**(chat.meta if isinstance(chat.meta, dict) else {}), **chat_context}
chat.save()
else:
cache.set(Cache_Version.CHAT_VARIABLE.get_key(key=self.chat_id), chat_context,
version=Cache_Version.CHAT_VARIABLE.get_version(),
timeout=60 * 30)

def get_chat_variable(self):
if not self.debug:
chat = QuerySet(Chat).filter(id=self.chat_id).first()
if chat:
return chat.meta
return {}
else:
return cache.get(Cache_Version.CHAT_VARIABLE.get_key(key=self.chat_id),
version=Cache_Version.CHAT_VARIABLE.get_version()) or {}

def append_chat_record(self, chat_record: ChatRecord):
chat_record.problem_text = chat_record.problem_text[0:10240] if chat_record.problem_text is not None else ""
chat_record.answer_text = chat_record.answer_text[0:40960] if chat_record.problem_text is not None else ""
Expand Down
2 changes: 2 additions & 0 deletions apps/chat/serializers/chat.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,6 +253,7 @@ def chat_simple(self, chat_info: ChatInfo, instance, base_to_response):
# 构建运行参数
params = chat_info.to_pipeline_manage_params(message, get_post_handler(chat_info), exclude_paragraph_id_list,
chat_user_id, chat_user_type, stream, form_data)
chat_info.set_chat(message)
# 运行流水线作业
pipeline_message.run(params)
return pipeline_message.context['chat_result']
Expand Down Expand Up @@ -307,6 +308,7 @@ def chat_work_flow(self, chat_info: ChatInfo, instance: dict, base_to_response):
other_list,
instance.get('runtime_node_id'),
instance.get('node_data'), chat_record, instance.get('child_node'))
chat_info.set_chat(message)
r = work_flow_manage.run()
return r

Expand Down
3 changes: 3 additions & 0 deletions apps/common/constants/cache_version.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,9 @@ class Cache_Version(Enum):

# 对话
CHAT = "CHAT", lambda key: key

CHAT_VARIABLE = "CHAT_VARIABLE", lambda key: key

# 应用API KEY
APPLICATION_API_KEY = "APPLICATION_API_KEY", lambda secret_key, use_get_data: secret_key

Expand Down
1 change: 1 addition & 0 deletions ui/src/locales/lang/zh-CN/views/application-workflow.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@ export default {
variable: {
label: '变量',
global: '全局变量',
chat: '会话变量',
Referencing: '引用变量',
ReferencingRequired: '引用变量必填',
ReferencingError: '引用变量错误',
Expand Down
4 changes: 3 additions & 1 deletion ui/src/workflow/common/NodeCascader.vue
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,9 @@ const wheel = (e: any) => {
function visibleChange(bool: boolean) {
if (bool) {
options.value = props.global
? props.nodeModel.get_up_node_field_list(false, true).filter((v: any) => v.value === 'global')
? props.nodeModel
.get_up_node_field_list(false, true)
.filter((v: any) => ['global', 'chat'].includes(v.value))
: props.nodeModel.get_up_node_field_list(false, true)
}
}
Expand Down
Loading
Loading