github/DB-GPT
1
0
Fork 0
mirror of https://github.com/csunny/DB-GPT.git synced 2026-07-17 01:58:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: AWEL flow support no-streaming chat build from UI

Browse Source
This commit is contained in:
Fangyin Cheng
2024-02-01 09:37:56 +08:00
parent 5918c2139d
commit fbe4e83f3f
10 changed files with 252 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.pre-commit-config.yaml
View File

@@ -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]

18
dbgpt/core/awel/flow/base.py
View File

@@ -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:

72
dbgpt/core/awel/flow/flow_factory.py
View File

@@ -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())

4
dbgpt/core/awel/trigger/ext_http_trigger.py
View File

@@ -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,

61
dbgpt/core/awel/trigger/http_trigger.py
View File

@@ -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}

61
dbgpt/core/interface/operators/llm_operator.py
View File

@@ -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,

2
dbgpt/core/interface/operators/prompt_operator.py
View File

@@ -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):

2
dbgpt/core/interface/prompt.py
View File

@@ -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", [])

34
dbgpt/serve/flow/models/models.py
View File

@@ -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)

24
dbgpt/serve/flow/service/service.py
View File

@@ -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