How do I get the contents of the "usage" in the json string output under the LLM node?

[Alickx]

Self Checks

• I have searched for existing issues search for existing issues, including closed ones.
• I confirm that I am using English to submit this report (我已阅读并同意 Language Policy).
• [FOR CHINESE USERS] 请务必使用英文提交 Issue，否则会被关闭。谢谢！:）
• Please do not modify this template :) and fill in all the required fields.

Content

{
  "text": "xxxxx",
  "usage": {
    "prompt_tokens": 1974,
    "prompt_unit_price": "5.00",
    "prompt_price_unit": "0.000001",
    "prompt_price": "0.0098700",
    "completion_tokens": 60,
    "completion_unit_price": "15.00",
    "completion_price_unit": "0.000001",
    "completion_price": "0.0009000",
    "total_tokens": 2034,
    "total_price": "0.0107700",
    "currency": "USD",
    "latency": 2.0280010253190994
  }
}

I want to extract the total_price of usage to the next node for processing, but I found that there seems to be no way to get it?

[GCLYTK]

Is this problem solved? I am also confused about this problem

[Alickx]

We modified the source code to query the Node list and return the usage quota when returning the result

[GCLYTK]

Can you give me some guidance?

[Alickx]

1. Modify the returned entity class WorkflowAppBlockingResponse api/core/app/entities/task_entities.py and add the returned amount field to it
2. Modify the workflow code in api/core/app/apps/workflow/generate_task_pipeline.py, the method is _to_blocking_response

    def _to_blocking_response(self, generator: Generator[StreamResponse, None, None]) -> WorkflowAppBlockingResponse:
        """
        To blocking response.
        :return:
        """
        for stream_response in generator:
            if isinstance(stream_response, ErrorStreamResponse):
                raise stream_response.err
            elif isinstance(stream_response, WorkflowFinishStreamResponse):
+                node_executions = (
+                    db.session.query(WorkflowNodeExecution)
+                    .filter(
+                        WorkflowNodeExecution.workflow_id == stream_response.data.workflow_id,
+                        WorkflowNodeExecution.workflow_run_id == stream_response.data.id,
+                        WorkflowNodeExecution.node_type == NodeType.LLM.value,
+                    )
+                    .order_by(WorkflowNodeExecution.index.desc())
+                    .all()
+                )
+
+                total_price_usd = 0.0
+                total_price_rmb = 0.0
+                for workflow_node_execution in node_executions:
+                    execution_metadata = workflow_node_execution.execution_metadata_dict
+                    total_price = float(
+                        execution_metadata.get('total_price', '0.0'))
+                    currency = execution_metadata.get('currency')
+                    if currency == 'USD':
+                        total_price_usd += total_price
+                    elif currency == 'RMB':
+                        total_price_rmb += total_price

                response = WorkflowAppBlockingResponse(
                    task_id=self._application_generate_entity.task_id,
                    workflow_run_id=stream_response.data.id,
                    data=WorkflowAppBlockingResponse.Data(
                        id=stream_response.data.id,
                        workflow_id=stream_response.data.workflow_id,
                        status=stream_response.data.status,
                        outputs=stream_response.data.outputs,
                        error=stream_response.data.error,
                        elapsed_time=stream_response.data.elapsed_time,
                        total_tokens=stream_response.data.total_tokens,
                        total_steps=stream_response.data.total_steps,
+                        total_price_usd='{:f}'.format(total_price_usd),
+                        total_price_rmb='{:f}'.format(total_price_rmb),
                        created_at=int(stream_response.data.created_at),
                        finished_at=int(stream_response.data.finished_at),
                    ),
                )

                return response
            else:
                continue

        raise Exception("Queue listening stopped unexpectedly.")