From fbe4e83f3fcac5e9cd88a498fd719c785b1a5361 Mon Sep 17 00:00:00 2001 From: Fangyin Cheng Date: Thu, 1 Feb 2024 09:37:56 +0800 Subject: [PATCH] feat: AWEL flow support no-streaming chat build from UI --- .pre-commit-config.yaml | 2 +- dbgpt/core/awel/flow/base.py | 18 +++++ dbgpt/core/awel/flow/flow_factory.py | 72 ++++++++++++++++++- dbgpt/core/awel/trigger/ext_http_trigger.py | 4 +- dbgpt/core/awel/trigger/http_trigger.py | 61 ++++++++++++++-- .../core/interface/operators/llm_operator.py | 61 ++++++++++++++-- .../interface/operators/prompt_operator.py | 2 +- dbgpt/core/interface/prompt.py | 2 +- dbgpt/serve/flow/models/models.py | 34 +++++++-- dbgpt/serve/flow/service/service.py | 24 ++++++- 10 files changed, 252 insertions(+), 28 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 2dd7e535d..eb82f70f8 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -4,7 +4,7 @@ repos: hooks: - id: python-fmt name: Python Format - entry: make fmt + entry: make fmt-check language: system exclude: '^dbgpt/app/static/|^web/' types: [python] diff --git a/dbgpt/core/awel/flow/base.py b/dbgpt/core/awel/flow/base.py index 552a2a86f..f4cf892f8 100644 --- a/dbgpt/core/awel/flow/base.py +++ b/dbgpt/core/awel/flow/base.py @@ -746,6 +746,24 @@ class ViewMetadata(BaseMetadata): values["name"], values["category"], values.get("version", "v1") ) values["id"] = values["flow_type"] + "_" + key + inputs = values.get("inputs") + outputs = values.get("outputs") + if inputs: + new_inputs = [] + for field in inputs: + if isinstance(field, dict): + new_inputs.append(IOField(**field)) + elif not isinstance(field, IOField): + raise ValueError("Inputs should be IOField.") + values["inputs"] = new_inputs + if outputs: + new_outputs = [] + for field in outputs: + if isinstance(field, dict): + new_outputs.append(IOField(**field)) + elif not isinstance(field, IOField): + raise ValueError("Outputs should be IOField.") + values["outputs"] = new_outputs return values def get_operator_key(self) -> str: diff --git a/dbgpt/core/awel/flow/flow_factory.py b/dbgpt/core/awel/flow/flow_factory.py index 951ecb7a7..4e6ef84ef 100644 --- a/dbgpt/core/awel/flow/flow_factory.py +++ b/dbgpt/core/awel/flow/flow_factory.py @@ -3,6 +3,7 @@ import logging import uuid from contextlib import suppress +from enum import Enum from typing import Any, Dict, List, Optional, Type, Union, cast from dbgpt._private.pydantic import BaseModel, Field, root_validator, validator @@ -122,6 +123,30 @@ class FlowData(BaseModel): viewport: FlowPositionData = Field(..., description="Viewport of the flow") +class State(str, Enum): + """State of a flow panel.""" + + INITIALIZING = "initializing" + DEVELOPING = "developing" + TESTING = "testing" + READY_TO_DEPLOY = "ready_to_deploy" + DEPLOYED = "deployed" + RUNNING = "running" + PAUSED = "paused" + DISABLED = "disabled" + ENABLED = "enabled" + + @classmethod + def value_of(cls, value: Optional[str]) -> "State": + """Get the state by value.""" + if not value: + return cls.INITIALIZING + for state in State: + if state.value == value: + return state + raise ValueError(f"Invalid state value: {value}") + + class FlowPanel(BaseModel): """Flow panel.""" @@ -142,6 +167,21 @@ class FlowPanel(BaseModel): description="Flow panel description", examples=["My first AWEL flow"], ) + state: State = Field( + default=State.INITIALIZING, description="Current state of the flow panel" + ) + source: Optional[str] = Field( + "DBGPT-WEB", + description="Source of the flow panel", + examples=["DB-GPT-WEB", "DBGPT-GITHUB"], + ) + source_url: Optional[str] = Field( + None, + description="Source url of the flow panel", + ) + version: Optional[str] = Field( + "0.1.0", description="Version of the flow panel", examples=["0.1.0", "0.2.0"] + ) user_name: Optional[str] = Field(None, description="User name") sys_code: Optional[str] = Field(None, description="System code") dag_id: Optional[str] = Field(None, description="DAG id, Created by AWEL") @@ -157,6 +197,26 @@ class FlowPanel(BaseModel): examples=["2021-08-01 12:00:00", "2021-08-01 12:00:01", "2021-08-01 12:00:02"], ) + def change_state(self, new_state: State) -> bool: + """Change the state of the flow panel.""" + allowed_transitions: Dict[State, List[State]] = { + State.INITIALIZING: [State.DEVELOPING], + State.DEVELOPING: [State.TESTING, State.DISABLED], + State.TESTING: [State.READY_TO_DEPLOY, State.DEVELOPING, State.DISABLED], + State.READY_TO_DEPLOY: [State.DEPLOYED, State.DEVELOPING], + State.DEPLOYED: [State.RUNNING, State.DISABLED], + State.RUNNING: [State.PAUSED, State.DISABLED, State.DEPLOYED], + State.PAUSED: [State.RUNNING, State.DISABLED], + State.DISABLED: [State.ENABLED], + State.ENABLED: [s for s in State if s != State.INITIALIZING], + } + if new_state in allowed_transitions[self.state]: + self.state = new_state + return True + else: + logger.error(f"Invalid state transition from {self.state} to {new_state}") + return False + class FlowFactory: """Flow factory.""" @@ -255,7 +315,11 @@ class FlowFactory: key_to_tasks[operator_key] = operator_task return self.build_dag( - flow_panel, key_to_tasks, key_to_downstream, key_to_upstream + flow_panel, + key_to_tasks, + key_to_downstream, + key_to_upstream, + dag_id=flow_panel.dag_id, ) def build_dag( @@ -264,11 +328,13 @@ class FlowFactory: key_to_tasks: Dict[str, DAGNode], key_to_downstream: Dict[str, List[str]], key_to_upstream: Dict[str, List[str]], + dag_id: Optional[str] = None, ) -> DAG: """Build the DAG.""" formatted_name = flow_panel.name.replace(" ", "_") - dag_name = f"{self._dag_prefix}_{formatted_name}_{flow_panel.uid}" - with DAG(dag_name) as dag: + if not dag_id: + dag_id = f"{self._dag_prefix}_{formatted_name}_{flow_panel.uid}" + with DAG(dag_id) as dag: for key, task in key_to_tasks.items(): if not task._node_id: task.set_node_id(dag._new_node_id()) diff --git a/dbgpt/core/awel/trigger/ext_http_trigger.py b/dbgpt/core/awel/trigger/ext_http_trigger.py index d1d40bde2..2465f25f6 100644 --- a/dbgpt/core/awel/trigger/ext_http_trigger.py +++ b/dbgpt/core/awel/trigger/ext_http_trigger.py @@ -7,7 +7,7 @@ from typing import List, Optional, Type, Union from starlette.requests import Request -from ..flow import OperatorCategory, OperatorType, Parameter, ViewMetadata +from ..flow import IOField, OperatorCategory, OperatorType, ViewMetadata from .http_trigger import ( _PARAMETER_ENDPOINT, _PARAMETER_MEDIA_TYPE, @@ -32,7 +32,7 @@ class RequestHttpTrigger(HttpTrigger): " as a starlette Request", inputs=[], outputs=[ - Parameter.build_from( + IOField.build_from( "Request Body", "request_body", Request, diff --git a/dbgpt/core/awel/trigger/http_trigger.py b/dbgpt/core/awel/trigger/http_trigger.py index bb89a791b..f995eb384 100644 --- a/dbgpt/core/awel/trigger/http_trigger.py +++ b/dbgpt/core/awel/trigger/http_trigger.py @@ -19,6 +19,7 @@ from dbgpt._private.pydantic import BaseModel, Field from ..dag.base import DAG from ..flow import ( + IOField, OperatorCategory, OperatorType, OptionValue, @@ -705,7 +706,7 @@ class DictHttpTrigger(HttpTrigger): " as a dict", inputs=[], outputs=[ - Parameter.build_from( + IOField.build_from( "Request Body", "request_body", dict, @@ -762,7 +763,7 @@ class StringHttpTrigger(HttpTrigger): " as a string", inputs=[], outputs=[ - Parameter.build_from( + IOField.build_from( "Request Body", "request_body", str, @@ -820,7 +821,7 @@ class CommonLLMHttpTrigger(HttpTrigger): "as a common LLM http body", inputs=[], outputs=[ - Parameter.build_from( + IOField.build_from( "Request Body", "request_body", CommonLLMHttpRequestBody, @@ -898,7 +899,7 @@ class ExampleHttpHelloOperator(MapOperator[dict, ExampleHttpResponse]): category=OperatorCategory.COMMON, parameters=[], inputs=[ - Parameter.build_from( + IOField.build_from( "Http Request Body", "request_body", dict, @@ -906,7 +907,7 @@ class ExampleHttpHelloOperator(MapOperator[dict, ExampleHttpResponse]): ) ], outputs=[ - Parameter.build_from( + IOField.build_from( "Response Body", "response_body", ExampleHttpResponse, @@ -947,7 +948,7 @@ class RequestBodyToDictOperator(MapOperator[CommonLLMHttpRequestBody, Dict[str, ) ], inputs=[ - Parameter.build_from( + IOField.build_from( "Request Body", "request_body", CommonLLMHttpRequestBody, @@ -955,7 +956,7 @@ class RequestBodyToDictOperator(MapOperator[CommonLLMHttpRequestBody, Dict[str, ) ], outputs=[ - Parameter.build_from( + IOField.build_from( "Response Body", "response_body", dict, @@ -984,3 +985,49 @@ class RequestBodyToDictOperator(MapOperator[CommonLLMHttpRequestBody, Dict[str, f"Prefix key {self._key} is not a valid key of the request body" ) return dict_value + + +class UserInputParsedOperator(MapOperator[CommonLLMHttpRequestBody, Dict[str, Any]]): + """User input parsed operator.""" + + metadata = ViewMetadata( + label="User Input Parsed Operator", + name="user_input_parsed_operator", + category=OperatorCategory.COMMON, + parameters=[ + Parameter.build_from( + "Key", + "key", + str, + optional=True, + default="user_input", + description="The key of the dict, link 'user_input'", + ) + ], + inputs=[ + IOField.build_from( + "Request Body", + "request_body", + CommonLLMHttpRequestBody, + description="The request body of the API endpoint", + ) + ], + outputs=[ + IOField.build_from( + "User Input Dict", + "user_input_dict", + dict, + description="The user input dict of the API endpoint", + ) + ], + description="User input parsed operator", + ) + + def __init__(self, key: str = "user_input", **kwargs): + """Initialize a UserInputParsedOperator.""" + self._key = key + super().__init__(**kwargs) + + async def map(self, request_body: CommonLLMHttpRequestBody) -> Dict[str, Any]: + """Map the request body to response body.""" + return {self._key: request_body.messages} diff --git a/dbgpt/core/interface/operators/llm_operator.py b/dbgpt/core/interface/operators/llm_operator.py index 141522bea..8c1f9bb34 100644 --- a/dbgpt/core/interface/operators/llm_operator.py +++ b/dbgpt/core/interface/operators/llm_operator.py @@ -12,10 +12,11 @@ from dbgpt.core.awel import ( CommonLLMHttpRequestBody, CommonLLMHttpResponseBody, DAGContext, + JoinOperator, MapOperator, StreamifyAbsOperator, ) -from dbgpt.core.awel.flow import OperatorCategory, Parameter, ViewMetadata +from dbgpt.core.awel.flow import IOField, OperatorCategory, Parameter, ViewMetadata from dbgpt.core.interface.llm import ( LLMClient, ModelOutput, @@ -23,6 +24,7 @@ from dbgpt.core.interface.llm import ( ModelRequestContext, ) from dbgpt.core.interface.message import ModelMessage +from dbgpt.util.function_utils import rearrange_args_by_type RequestInput = Union[ ModelRequest, @@ -34,7 +36,7 @@ RequestInput = Union[ ] -class RequestBuilderOperator(MapOperator[RequestInput, ModelRequest], ABC): +class RequestBuilderOperator(MapOperator[RequestInput, ModelRequest]): """Build the model request from the input value.""" metadata = ViewMetadata( @@ -53,7 +55,7 @@ class RequestBuilderOperator(MapOperator[RequestInput, ModelRequest], ABC): ), ], inputs=[ - Parameter.build_from( + IOField.build_from( "Request Body", "input_value", CommonLLMHttpRequestBody, @@ -61,7 +63,7 @@ class RequestBuilderOperator(MapOperator[RequestInput, ModelRequest], ABC): ), ], outputs=[ - Parameter.build_from( + IOField.build_from( "Model Request", "output_value", ModelRequest, @@ -126,6 +128,53 @@ class RequestBuilderOperator(MapOperator[RequestInput, ModelRequest], ABC): return ModelRequest(**req_dict) +class MergedRequestBuilderOperator(JoinOperator[ModelRequest]): + """Build the model request from the input value.""" + + metadata = ViewMetadata( + label="Merge Model Request Messages", + name="merged_request_builder_operator", + category=OperatorCategory.COMMON, + description="Merge the model request from the input value.", + parameters=[], + inputs=[ + IOField.build_from( + "Model Request", + "model_request", + ModelRequest, + description="The model request of upstream.", + ), + IOField.build_from( + "Model messages", + "messages", + ModelMessage, + description="The model messages of upstream.", + is_list=True, + ), + ], + outputs=[ + IOField.build_from( + "Model Request", + "output_value", + ModelRequest, + description="The output value of the operator.", + ), + ], + ) + + def __init__(self, **kwargs): + """Create a new request builder operator.""" + super().__init__(combine_function=self.merge_func, **kwargs) + + @rearrange_args_by_type + def merge_func( + self, model_request: ModelRequest, messages: List[ModelMessage] + ) -> ModelRequest: + """Merge the model request with the messages.""" + model_request.messages = messages + return model_request + + class BaseLLM: """The abstract operator for a LLM.""" @@ -276,7 +325,7 @@ class ModelOutput2CommonResponseOperator( description="Map the model output to the common response body.", parameters=[], inputs=[ - Parameter.build_from( + IOField.build_from( "Model Output", "input_value", ModelOutput, @@ -284,7 +333,7 @@ class ModelOutput2CommonResponseOperator( ), ], outputs=[ - Parameter.build_from( + IOField.build_from( "Common Response Body", "output_value", CommonLLMHttpResponseBody, diff --git a/dbgpt/core/interface/operators/prompt_operator.py b/dbgpt/core/interface/operators/prompt_operator.py index 8505975f6..c4e05fbb1 100644 --- a/dbgpt/core/interface/operators/prompt_operator.py +++ b/dbgpt/core/interface/operators/prompt_operator.py @@ -71,7 +71,7 @@ class CommonChatPromptTemplate(ChatPromptTemplate): SystemPromptTemplate.from_template(system_message), HumanPromptTemplate.from_template(human_message), ] - return values + return cls.base_pre_fill(values) class BasePromptBuilderOperator(BaseConversationOperator, ABC): diff --git a/dbgpt/core/interface/prompt.py b/dbgpt/core/interface/prompt.py index bc536aab0..1351eb09a 100644 --- a/dbgpt/core/interface/prompt.py +++ b/dbgpt/core/interface/prompt.py @@ -240,7 +240,7 @@ class ChatPromptTemplate(BasePromptTemplate): return result_messages @root_validator(pre=True) - def pre_fill(cls, values: Dict[str, Any]) -> Dict[str, Any]: + def base_pre_fill(cls, values: Dict[str, Any]) -> Dict[str, Any]: """Pre-fill the messages.""" input_variables = values.get("input_variables", {}) messages = values.get("messages", []) diff --git a/dbgpt/serve/flow/models/models.py b/dbgpt/serve/flow/models/models.py index 47abd6581..22ec1c472 100644 --- a/dbgpt/serve/flow/models/models.py +++ b/dbgpt/serve/flow/models/models.py @@ -7,6 +7,7 @@ from typing import Any, Dict, Union from sqlalchemy import Column, DateTime, Integer, String, Text, UniqueConstraint +from dbgpt.core.awel.flow.flow_factory import State from dbgpt.storage.metadata import BaseDao, Model from dbgpt.storage.metadata._base_dao import QUERY_SPEC @@ -23,7 +24,11 @@ class ServeEntity(Model): dag_id = Column(String(128), index=True, nullable=True, comment="DAG id") name = Column(String(128), index=True, nullable=True, comment="Flow name") flow_data = Column(Text, nullable=True, comment="Flow data, JSON format") - description = Column(String(128), nullable=True, comment="Flow description") + description = Column(String(512), nullable=True, comment="Flow description") + state = Column(String(32), nullable=True, comment="Flow state") + source = Column(String(64), nullable=True, comment="Flow source") + source_url = Column(String(512), nullable=True, comment="Flow source url") + version = Column(String(32), nullable=True, comment="Flow version") user_name = Column(String(128), index=True, nullable=True, comment="User name") sys_code = Column(String(128), index=True, nullable=True, comment="System code") gmt_created = Column(DateTime, default=datetime.now, comment="Record creation time") @@ -56,11 +61,16 @@ class ServeDao(BaseDao[ServeEntity, ServeRequest, ServerResponse]): """ request_dict = request.dict() if isinstance(request, ServeRequest) else request flow_data = json.dumps(request_dict.get("flow_data"), ensure_ascii=False) + state = request_dict.get("state", State.INITIALIZING.value) new_dict = { "uid": request_dict.get("uid"), "dag_id": request_dict.get("dag_id"), "name": request_dict.get("name"), "flow_data": flow_data, + "state": state, + "source": request_dict.get("source"), + "source_url": request_dict.get("source_url"), + "version": request_dict.get("version"), "description": request_dict.get("description"), "user_name": request_dict.get("user_name"), "sys_code": request_dict.get("sys_code"), @@ -83,6 +93,10 @@ class ServeDao(BaseDao[ServeEntity, ServeRequest, ServerResponse]): dag_id=entity.dag_id, name=entity.name, flow_data=flow_data, + state=State.value_of(entity.state), + source=entity.source, + source_url=entity.source_url, + version=entity.version, description=entity.description, user_name=entity.user_name, sys_code=entity.sys_code, @@ -106,6 +120,10 @@ class ServeDao(BaseDao[ServeEntity, ServeRequest, ServerResponse]): name=entity.name, flow_data=flow_data, description=entity.description, + state=State.value_of(entity.state), + source=entity.source, + source_url=entity.source_url, + version=entity.version, user_name=entity.user_name, sys_code=entity.sys_code, gmt_created=gmt_created_str, @@ -115,7 +133,7 @@ class ServeDao(BaseDao[ServeEntity, ServeRequest, ServerResponse]): def update( self, query_request: QUERY_SPEC, update_request: ServeRequest ) -> ServerResponse: - with self.session() as session: + with self.session(commit=False) as session: query = self._create_query_object(session, query_request) entry: ServeEntity = query.first() if entry is None: @@ -128,10 +146,18 @@ class ServeDao(BaseDao[ServeEntity, ServeRequest, ServerResponse]): ) if update_request.description: entry.description = update_request.description - + if update_request.state: + entry.state = update_request.state.value + if update_request.source: + entry.source = update_request.source + if update_request.source_url: + entry.source_url = update_request.source_url + if update_request.version: + entry.version = update_request.version if update_request.user_name: entry.user_name = update_request.user_name if update_request.sys_code: entry.sys_code = update_request.sys_code session.merge(entry) - return self.get_one(self.to_request(entry)) + session.commit() + return self.get_one(query_request) diff --git a/dbgpt/serve/flow/service/service.py b/dbgpt/serve/flow/service/service.py index dfe5ab35b..c167f8f64 100644 --- a/dbgpt/serve/flow/service/service.py +++ b/dbgpt/serve/flow/service/service.py @@ -99,10 +99,24 @@ class Service(BaseService[ServeEntity, ServeRequest, ServerResponse]): Returns: ServerResponse: The response """ - # TODO: implement your own logic here # Build the query request from the request query_request = {"uid": request.uid} - return self.dao.update(query_request, update_request=request) + inst = self.get(query_request) + if not inst: + raise HTTPException(status_code=404, detail=f"Flow {request.uid} not found") + old_data: Optional[ServerResponse] = None + try: + update_obj = self.dao.update(query_request, update_request=request) + old_data = self.delete(request.uid) + if not old_data: + raise HTTPException( + status_code=404, detail=f"Flow detail {request.uid} not found" + ) + return self.create(update_obj) + except Exception as e: + if old_data: + self.create(old_data) + raise e def get(self, request: QUERY_SPEC) -> Optional[ServerResponse]: """Get a Flow entity @@ -118,11 +132,14 @@ class Service(BaseService[ServeEntity, ServeRequest, ServerResponse]): query_request = request return self.dao.get_one(query_request) - def delete(self, uid: str) -> None: + def delete(self, uid: str) -> Optional[ServerResponse]: """Delete a Flow entity Args: uid (str): The uid + + Returns: + ServerResponse: The data after deletion """ # TODO: implement your own logic here @@ -137,6 +154,7 @@ class Service(BaseService[ServeEntity, ServeRequest, ServerResponse]): ) self.dag_manager.unregister_dag(inst.dag_id) self.dao.delete(query_request) + return inst def get_list(self, request: ServeRequest) -> List[ServerResponse]: """Get a list of Flow entities