feat: llm ops tracing

api/.env.example

@@ -224,3 +224,4 @@ WORKFLOW_CALL_MAX_DEPTH=5
 # App configuration
 APP_MAX_EXECUTION_TIME=1200
 
+

api/controllers/console/__init__.py

@@ -20,6 +20,7 @@
     generator,
     message,
     model_config,
+    ops_trace,
     site,
     statistic,
     workflow,

api/controllers/console/app/app.py

@@ -1,4 +1,3 @@
-import json
 import uuid
 
 from flask_login import current_user
@@ -9,17 +8,14 @@
 from controllers.console.app.wraps import get_app_model
 from controllers.console.setup import setup_required
 from controllers.console.wraps import account_initialization_required, cloud_edition_billing_resource_check
-from core.tools.tool_manager import ToolManager
-from core.tools.utils.configuration import ToolParameterConfigurationManager
 from fields.app_fields import (
     app_detail_fields,
     app_detail_fields_with_site,
     app_pagination_fields,
 )
 from libs.login import login_required
-from models.model import App, AppMode, AppModelConfig
 from services.app_service import AppService
-from services.tag_service import TagService
+from services.ops_trace.ops_trace_service import OpsTraceService
 
 ALLOW_CREATE_APP_MODES = ['chat', 'agent-chat', 'advanced-chat', 'workflow', 'completion']
 
@@ -286,6 +282,39 @@ def post(self, app_model):
         return app_model
 
 
+class AppTraceApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, app_id):
+        """Get app trace"""
+        app_trace_config = OpsTraceService.get_app_tracing_config(
+            app_id=app_id
+        )
+
+        return app_trace_config
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, app_id):
+        # add app trace
+        if not current_user.is_admin_or_owner:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument('enabled', type=bool, required=True, location='json')
+        parser.add_argument('tracing_provider', type=str, required=True, location='json')
+        args = parser.parse_args()
+
+        OpsTraceService.update_app_tracing_config(
+            app_id=app_id,
+            enabled=args['enabled'],
+            tracing_provider=args['tracing_provider'],
+        )
+
+        return {"result": "success"}
+
+
 api.add_resource(AppListApi, '/apps')
 api.add_resource(AppImportApi, '/apps/import')
 api.add_resource(AppApi, '/apps/<uuid:app_id>')
@@ -295,3 +324,4 @@ def post(self, app_model):
 api.add_resource(AppIconApi, '/apps/<uuid:app_id>/icon')
 api.add_resource(AppSiteStatus, '/apps/<uuid:app_id>/site-enable')
 api.add_resource(AppApiStatus, '/apps/<uuid:app_id>/api-enable')
+api.add_resource(AppTraceApi, '/apps/<uuid:app_id>/trace')

api/controllers/console/app/error.py

@@ -97,3 +97,15 @@ class DraftWorkflowNotSync(BaseHTTPException):
     error_code = 'draft_workflow_not_sync'
     description = "Workflow graph might have been modified, please refresh and resubmit."
     code = 400
+
+
+class TracingConfigNotExist(BaseHTTPException):
+    error_code = 'trace_config_not_exist'
+    description = "Trace config not exist."
+    code = 400
+
+
+class TracingConfigIsExist(BaseHTTPException):
+    error_code = 'trace_config_is_exist'
+    description = "Trace config is exist."
+    code = 400

api/controllers/console/app/ops_trace.py

@@ -0,0 +1,99 @@
+from flask_restful import Resource, reqparse
+
+from controllers.console import api
+from controllers.console.app.error import TracingConfigIsExist, TracingConfigNotExist
+from controllers.console.setup import setup_required
+from controllers.console.wraps import account_initialization_required
+from libs.login import login_required
+from services.ops_trace.ops_trace_service import OpsTraceService
+
+
+class TraceAppConfigApi(Resource):
+    """
+    Manage trace app configurations
+    """
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, app_id):
+        parser = reqparse.RequestParser()
+        parser.add_argument('tracing_provider', type=str, required=True, location='args')
+        args = parser.parse_args()
+
+        try:
+            trace_config = OpsTraceService.get_tracing_app_config(
+                app_id=app_id, tracing_provider=args['tracing_provider']
+                )
+            if not trace_config:
+                return {"has_not_configured": True}
+            return trace_config
+        except Exception as e:
+            raise e
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, app_id):
+        """Create a new trace app configuration"""
+        parser = reqparse.RequestParser()
+        parser.add_argument('tracing_provider', type=str, required=True, location='json')
+        parser.add_argument('tracing_config', type=dict, required=True, location='json')
+        args = parser.parse_args()
+
+        try:
+            result = OpsTraceService.create_tracing_app_config(
+                app_id=app_id,
+                tracing_provider=args['tracing_provider'],
+                tracing_config=args['tracing_config']
+            )
+            if not result:
+                raise TracingConfigIsExist()
+            return {"result": "success"}
+        except Exception as e:
+            raise e
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def patch(self, app_id):
+        """Update an existing trace app configuration"""
+        parser = reqparse.RequestParser()
+        parser.add_argument('tracing_provider', type=str, required=True, location='json')
+        parser.add_argument('tracing_config', type=dict, required=True, location='json')
+        args = parser.parse_args()
+
+        try:
+            result = OpsTraceService.update_tracing_app_config(
+                app_id=app_id,
+                tracing_provider=args['tracing_provider'],
+                tracing_config=args['tracing_config']
+            )
+            if not result:
+                raise TracingConfigNotExist()
+            return {"result": "success"}
+        except Exception as e:
+            raise e
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def delete(self, app_id):
+        """Delete an existing trace app configuration"""
+        parser = reqparse.RequestParser()
+        parser.add_argument('tracing_provider', type=str, required=True, location='args')
+        args = parser.parse_args()
+
+        try:
+            result = OpsTraceService.delete_tracing_app_config(
+                app_id=app_id,
+                tracing_provider=args['tracing_provider']
+            )
+            if not result:
+                raise TracingConfigNotExist()
+            return {"result": "success"}
+        except Exception as e:
+            raise e
+
+
+api.add_resource(TraceAppConfigApi, '/apps/<uuid:app_id>/trace-config')

api/controllers/console/datasets/error.py

@@ -83,3 +83,4 @@ class DatasetInUseError(BaseHTTPException):
     error_code = 'dataset_in_use'
     description = "The dataset is being used by some apps. Please remove the dataset from the apps before deleting it."
     code = 409
+

api/core/agent/cot_agent_runner.py

@@ -1,7 +1,7 @@
 import json
 from abc import ABC, abstractmethod
 from collections.abc import Generator
-from typing import Union
+from typing import Optional, Union
 
 from core.agent.base_agent_runner import BaseAgentRunner
 from core.agent.entities import AgentScratchpadUnit
@@ -20,6 +20,7 @@
 from core.tools.tool.tool import Tool
 from core.tools.tool_engine import ToolEngine
 from models.model import Message
+from services.ops_trace.base_trace_instance import BaseTraceInstance
 
 
 class CotAgentRunner(BaseAgentRunner, ABC):
@@ -31,17 +32,21 @@ class CotAgentRunner(BaseAgentRunner, ABC):
     _query: str = None
     _prompt_messages_tools: list[PromptMessage] = None
 
-    def run(self, message: Message,
-            query: str,
-            inputs: dict[str, str],
-            ) -> Union[Generator, LLMResult]:
+    def run(
+        self, message: Message,
+        query: str,
+        inputs: dict[str, str],
+    ) -> Union[Generator, LLMResult]:
         """
         Run Cot agent application
         """
         app_generate_entity = self.application_generate_entity
         self._repack_app_generate_entity(app_generate_entity)
         self._init_react_state(query)
 
+        # get tracing instance
+        tracing_instance = app_generate_entity.tracing_instance
+
         # check model mode
         if 'Observation' not in app_generate_entity.model_conf.stop:
             if app_generate_entity.model_conf.provider not in self._ignore_observation_providers:
@@ -183,7 +188,7 @@ def increase_usage(final_llm_usage_dict: dict[str, LLMUsage], usage: LLMUsage):
                 messages_ids=[],
                 llm_usage=usage_dict['usage']
             )
-
+            
             if not scratchpad.is_final():
                 self.queue_manager.publish(QueueAgentThoughtEvent(
                     agent_thought_id=agent_thought.id
@@ -209,9 +214,10 @@ def increase_usage(final_llm_usage_dict: dict[str, LLMUsage], usage: LLMUsage):
                     function_call_state = True
                     # action is tool call, invoke tool
                     tool_invoke_response, tool_invoke_meta = self._handle_invoke_action(
-                        action=scratchpad.action,
+                        action=scratchpad.action, 
                         tool_instances=tool_instances,
-                        message_file_ids=message_file_ids
+                        message_file_ids=message_file_ids,
+                        tracing_instance=tracing_instance,
                     )
                     scratchpad.observation = tool_invoke_response
                     scratchpad.agent_response = tool_invoke_response
@@ -237,8 +243,7 @@ def increase_usage(final_llm_usage_dict: dict[str, LLMUsage], usage: LLMUsage):
 
                 # update prompt tool message
                 for prompt_tool in self._prompt_messages_tools:
-                    self.update_prompt_message_tool(
-                        tool_instances[prompt_tool.name], prompt_tool)
+                    self.update_prompt_message_tool(tool_instances[prompt_tool.name], prompt_tool)
 
             iteration_step += 1
 
@@ -257,12 +262,12 @@ def increase_usage(final_llm_usage_dict: dict[str, LLMUsage], usage: LLMUsage):
 
         # save agent thought
         self.save_agent_thought(
-            agent_thought=agent_thought,
+            agent_thought=agent_thought, 
             tool_name='',
             tool_input={},
             tool_invoke_meta={},
             thought=final_answer,
-            observation={},
+            observation={}, 
             answer=final_answer,
             messages_ids=[]
         )
@@ -275,14 +280,15 @@ def increase_usage(final_llm_usage_dict: dict[str, LLMUsage], usage: LLMUsage):
             message=AssistantPromptMessage(
                 content=final_answer
             ),
-            usage=llm_usage['usage'] if llm_usage['usage'] else LLMUsage.empty_usage(
-            ),
+            usage=llm_usage['usage'] if llm_usage['usage'] else LLMUsage.empty_usage(),
             system_fingerprint=''
         )), PublishFrom.APPLICATION_MANAGER)
 
-    def _handle_invoke_action(self, action: AgentScratchpadUnit.Action,
+    def _handle_invoke_action(self, action: AgentScratchpadUnit.Action, 
                               tool_instances: dict[str, Tool],
-                              message_file_ids: list[str]) -> tuple[str, ToolInvokeMeta]:
+                              message_file_ids: list[str],
+                              tracing_instance: Optional[BaseTraceInstance] = None
+                              ) -> tuple[str, ToolInvokeMeta]:
         """
         handle invoke action
         :param action: action
@@ -297,7 +303,7 @@ def _handle_invoke_action(self, action: AgentScratchpadUnit.Action,
         if not tool_instance:
             answer = f"there is not a tool named {tool_call_name}"
             return answer, ToolInvokeMeta.error_instance(answer)
-
+        
         if isinstance(tool_call_args, str):
             try:
                 tool_call_args = json.loads(tool_call_args)
@@ -312,14 +318,14 @@ def _handle_invoke_action(self, action: AgentScratchpadUnit.Action,
             tenant_id=self.tenant_id,
             message=self.message,
             invoke_from=self.application_generate_entity.invoke_from,
-            agent_tool_callback=self.agent_callback
+            agent_tool_callback=self.agent_callback,
+            tracing_instance=tracing_instance,
         )
 
         # publish files
         for message_file, save_as in message_files:
             if save_as:
-                self.variables_pool.set_file(
-                    tool_name=tool_call_name, value=message_file.id, name=save_as)
+                self.variables_pool.set_file(tool_name=tool_call_name, value=message_file.id, name=save_as)
 
             # publish message file
             self.queue_manager.publish(QueueMessageFileEvent(
@@ -350,15 +356,15 @@ def _fill_in_inputs_from_external_data_tools(self, instruction: str, inputs: dic
                 continue
 
         return instruction
-
+    
     def _init_react_state(self, query) -> None:
         """
         init agent scratchpad
         """
         self._query = query
         self._agent_scratchpad = []
         self._historic_prompt_messages = self._organize_historic_prompt_messages()
-
+    
     @abstractmethod
     def _organize_prompt_messages(self) -> list[PromptMessage]:
         """
@@ -390,6 +396,13 @@ def _organize_historic_prompt_messages(self, current_session_messages: list[Prom
         scratchpads: list[AgentScratchpadUnit] = []
         current_scratchpad: AgentScratchpadUnit = None
 
+        self.history_prompt_messages = AgentHistoryPromptTransform(
+            model_config=self.model_config,
+            prompt_messages=current_session_messages or [],
+            history_messages=self.history_prompt_messages,
+            memory=self.memory
+        ).get_prompt()
+
         for message in self.history_prompt_messages:
             if isinstance(message, AssistantPromptMessage):
                 if not current_scratchpad:

api/core/agent/fc_agent_runner.py

@@ -50,6 +50,9 @@ def run(self,
         }
         final_answer = ''
 
+        # get tracing instance
+        tracing_instance = app_generate_entity.tracing_instance
+
         def increase_usage(final_llm_usage_dict: dict[str, LLMUsage], usage: LLMUsage):
             if not final_llm_usage_dict['usage']:
                 final_llm_usage_dict['usage'] = usage
@@ -243,6 +246,7 @@ def increase_usage(final_llm_usage_dict: dict[str, LLMUsage], usage: LLMUsage):
                         message=self.message,
                         invoke_from=self.application_generate_entity.invoke_from,
                         agent_tool_callback=self.agent_callback,
+                        tracing_instance=tracing_instance
                     )
                     # publish files
                     for message_file, save_as in message_files: