从 HTTP 触发器到 DAG 编排DB-GPT AWEL 工作流快速上手指南【免费下载链接】DB-GPTopen-source agentic AI data assistant for the next generation of AI Data products.项目地址: https://gitcode.com/GitHub_Trending/db/DB-GPT本文基于 DB-GPT 仓库中的 AWELAgentic Workflow Expression Language入门文档完整讲解如何用 HTTP 触发器 自定义算子搭建一条最小可运行的工作流你将学会定义请求体模型、编写自定义 MapOperator、用语法组装 DAG并掌握“生产模式挂服务”与“开发模式本地调试”两种验证方式最终能把任意 AWEL 编排以 REST 接口形式暴露出来。AWEL 是什么为 LLM 应用设计的智能体工作流语言AWELAgentic Workflow Expression Language智能体工作流表达语言是 DB-GPT 专门为大模型应用开发设计的编排框架。其官方包文档给出了清晰的定位描述AWEL is a set of intelligent agent workflow expression language specially designed for large model application development. It provides great functionality and flexibility. Through the AWEL API, you can focus on the development of business logic for LLMs applications without paying attention to cumbersome model and environment details.从源码包 awel/init.py 的导出列表可以确认AWEL 对外暴露的核心构件分为四类DAG 编排层DAG、DAGContext、DAGVar定义任务间的依赖关系算子层OperatorBaseOperator、MapOperator、JoinOperator、BranchOperator、ReduceStreamOperator、InputOperator等对应不同数据流转模式一对一映射、分支、归并、流式聚合触发器层TriggerHttpTrigger、IteratorTrigger以及可选依赖的RequestHttpTriggerawel/init.py 中通过 try/except 做可选导入缺少 starlette 依赖时自动降级执行层RunnerDefaultWorkflowRunner及TaskContext、TaskOutput、InputSource等任务执行抽象。理解这四层后下面这个入门示例就顺理成章了一个 AWEL 应用 触发器接收输入 若干算子处理数据 DAG描述流转顺序 Runner负责调度执行。示例目标HTTP 请求 输出改写入门示例的核心功能是“处理一个 HTTP 请求的输入并改写其输出”因此整条编排只包含两步HTTP Request接收请求Processing HTTP Response Result处理请求体返回改写后的结果。DB-GPT 已将上述依赖的基础算子封装好可直接引用from dbgpt._private.pydantic import BaseModel, Field from dbgpt.core.awel import DAG, HttpTrigger, MapOperator注意BaseModel与Field的导入路径是dbgpt._private.pydantic而非直接pydantic——这是 DB-GPT 对 pydantic v1/v2 做版本兼容适配的做法写 AWEL 算子时应统一使用这个路径以保证行为一致。第一步定义请求体模型定义一个接受name和age两个参数的 HTTP 请求体class TriggerReqBody(BaseModel): name: str Field(..., descriptionUser name) age: int Field(18, descriptionUser age)这个模型将作为HttpTrigger的request_body参数传入框架会用它做请求体解析与校验name为必填字段...age缺省时默认取18。第二步自定义算子 RequestHandleOperator定义一个请求处理算子RequestHandleOperator它继承基础MapOperator并做泛型特化MapOperator[TriggerReqBody, str]声明输入类型为TriggerReqBody、输出类型为str。算子动作非常直接——解析请求体取出 name 与 age 字段拼接成一句话例如Hello, zhangsan, your age is 18.class RequestHandleOperator(MapOperator[TriggerReqBody, str]): def __init__(self, **kwargs): super().__init__(**kwargs) async def map(self, input_value: TriggerReqBody) - str: print(fReceive input value: {input_value}) return fHello, {input_value.name}, your age is {input_value.age}这里有两个值得注意的设计点map是异步方法AWEL 的执行链路整体基于 asyncio算子的核心处理方法声明为asyncRunner 会在事件循环中调度它这使得同一 DAG 内可以并行处理多个请求泛型参数即接口契约MapOperator[I, O]的输入/输出类型在类定义处就确定了DAG 组装时上下游节点的数据类型匹配关系在编写期就能被表达出来而不是等到运行期才发现类型不匹配。第三步用 DAG 上下文组装管道写完算子后用 DAG 上下文把它们装配成编排图。这条 DAG 共两个节点第一个是内置的HttpTrigger负责处理 HTTP 请求第二个是自定义的RequestHandleOperator处理请求体with DAG(simple_dag_example) as dag: trigger HttpTrigger(/examples/hello, request_bodyTriggerReqBody) map_node RequestHandleOperator() trigger map_nodewith DAG(...)是 DAG 的上下文管理器写法dag变量保存了整个编排图实例后文开发模式中会用到。HttpTrigger的第一个参数是相对端点路径/examples/hellorequest_body指定请求体解析模型。关于trigger map_node这条依赖边源码层面有明确的实现依据dag/base.py 中的DependencyMixin定义了set_upstream/set_downstream接口并重载了移位运算符——node next_node实际调用node.set_downstream(next_node)node input_node则调用set_upstream且都支持传入节点列表以一次声明多条依赖。也就是说只是“设置下游节点”的语法糖DAG 内部维护的是显式的上下游依赖关系图Runner 据此决定执行顺序与并行度。访问验证两种运行模式示例提供了两种验证路径生产模式挂到 DB-GPT 服务端与开发模式本地独立调试分别对应不同的端口与启动方式。生产模式随 dbgpt_server 启动进行访问验证前需要先启动项目服务python dbgpt/app/dbgpt_server.py文档写作时的入口路径为dbgpt/app/dbgpt_server.py在当前仓库的 monorepo 布局下服务端代码位于packages/dbgpt-app/src/dbgpt_app/目录其中 dbgpt_app/_cli.py 中的from dbgpt_app.dbgpt_server import run_webserver表明dbgpt_server模块仍然存在可通过 CLI 的 start 子命令启动。服务启动后DAG 定义文件会被自动加载AWEL 的DAGManager启动时扫描配置的 DAG 目录并注册触发器节点然后即可用 curl 验证curl -X GET http://127.0.0.1:5670/api/v1/awel/trigger/examples/hello\?name\zhangsan Hello, zhangsan, your age is 18这个 URL 的三段式结构是 AWEL 触发器路由的通用形态各段含义在源码中均有对应http://127.0.0.1:5670DB-GPT 服务端默认监听地址/api/v1/awel/trigger触发器路由前缀。在 trigger_manager.py 中HttpTriggerManager的构造参数router_prefix默认值就是/api/v1/awel/trigger注册时会把前缀与触发器端点拼接成完整路径examples/helloHttpTrigger(/examples/hello, ...)中声明的相对端点。参数namezhangsan通过查询字符串传入age缺省由TriggerReqBody的默认值补全为 18最终响应即为算子map方法拼接出的字符串。开发模式不启动服务直接调试为了让用户更方便地测试AWEL 提供了无需启动完整 dbgpt_server 的开发环境。在 DAG 定义之后追加如下代码if __name__ __main__: if dag.leaf_nodes[0].dev_mode: # Development mode, you can run the dag locally for debugging. from dbgpt.core.awel import setup_dev_environment setup_dev_environment([dag], port5555) else: # Production mode, DB-GPT will automatically load and execute the current file after startup. pass然后直接运行python examples/awel/simple_dag_example.py在不启动项目的情况下测试curl -X GET http://127.0.0.1:5555/api/v1/awel/trigger/examples/hello\?name\zhangsan Hello, zhangsan, your age is 18从 awel/init.py 的setup_dev_environment实现看它的签名与行为如下def setup_dev_environment( dags: List[DAG], host: str 127.0.0.1, port: int 5555, logging_level: Optional[str] None, logger_filename: Optional[str] None, show_dag_graph: Optional[bool] True, ) - Nonedags待运行的 DAG 列表示例传的是[dag]host/port开发服务器的绑定地址默认127.0.0.1:5555所以示例中显式传port5555其实与默认值一致logging_level/logger_filename日志级别与日志文件名未指定时默认写入dbgpt_awel_dev.logshow_dag_graph默认True会调用dag.visualize_dag()把 DAG 图保存为文件并尝试自动打开若系统未安装 graphviz会降级为 warning 提示安装pip install graphviz或sudo apt install graphviz不会中断运行。其内部执行流程是创建含 HTTP 触发器时一个 FastAPI 应用 → 构建SystemApp并设置DAGVar上下文 → 创建DefaultTriggerManager并把每个 DAG 的trigger_nodes逐个register_trigger→ 最后用 uvicorn 启动 HTTP 服务。也就是说开发模式本质上是在本地起一个“精简版 DB-GPT 服务”只挂载你传入的这些 DAG路由前缀同样是/api/v1/awel/trigger因此两个模式下的 curl 命令结构完全相同只是端口从 5670 换成了 5555。另外注意if dag.leaf_nodes[0].dev_mode这个分支判断同一个 DAG 定义文件同时兼容两种模式——由脚本直接运行开发调试时走setup_dev_environment由 DB-GPT 服务端加载执行时生产模式则什么都不做交给服务端的 DAG 加载机制接管。仓库中对应的完整示例文件是 examples/awel/simple_dag_example.py其文件头 docstring 中给出的调用示例与上文 curl 命令一致且仓库中已存在可直接查看的 examples/awel/simple_dag_example.py 文件与文档代码完全对应。生产模式下 DAG 是如何被自动加载的文档中提到“DB-GPT 启动后会自动加载执行当前文件”这一行为背后的调用链在源码中可以完整确认服务初始化时component_configs.py 中的_initialize_awel(system_app, web_config.awel_dirs)被调用它先取内置的_DAG_DEFINITION_DIR作为基础 DAG 目录再追加配置文件awel_dirs中指定的目录逗号分隔最后调用initialize_awel(system_app, dag_dirs)initialize_awel 做了三件事绑定DAGVar的 SystemApp 上下文、注册DefaultTriggerManager组件、创建DAGManager(system_app, dag_dirs)并注册为系统实例最后initialize_runner(DefaultWorkflowRunner())安装默认执行器DAGManager 内部使用LocalFileDAGLoader(dag_dirs)扫描指定目录下的 DAG 定义文件loader.py加载后注册其中声明的触发器节点触发器注册时HttpTriggerManager.register_trigger 会把router_prefix/api/v1/awel/trigger与触发器的真实端点拼接成完整路由挂载到 FastAPI 应用/路由上并维护路由表防止路径冲突。因此“把 DAG 文件放进配置的awel_dirs目录 重启服务”就是在生产环境上线一条 AWEL 工作流的标准操作而从HttpTrigger支持methods、http_response_body、streaming_response等参数见 http_trigger.py 中HttpTriggerMetadata与请求体体系BaseHttpBody/DictHttpBody/StringHttpBody可以看出该触发器机制同样支撑 POST、流式响应等更复杂的 API 形态。小结与延伸本文沿 AWEL 入门文档的脉络走完了最小闭环用BaseModel定义请求体TriggerReqBodyname 必填、age 默认 18用MapOperator[TriggerReqBody, str]派生自定义算子RequestHandleOperator核心逻辑写在async def map中用with DAG(...) as dag:上下文把HttpTrigger与算子用连成两节点 DAG生产模式下随 dbgpt_server 启动后访问http://127.0.0.1:5670/api/v1/awel/trigger/examples/hello?namezhangsan开发模式下用setup_dev_environment([dag], port5555)在本地 5555 端口独立验证二者响应一致。掌握这个模式后可以沿着以下仓库入口继续深入 AWEL 的能力边界完整示例脚本 examples/awel/simple_dag_example.py、AWEL 核心包 packages/dbgpt-core/src/dbgpt/core/awel/、触发器实现 packages/dbgpt-core/src/dbgpt/core/awel/trigger/、DAG 基类与依赖关系实现 packages/dbgpt-core/src/dbgpt/core/awel/dag/base.py以及官方文档中的进阶章节 docs/docs/awel/awel.md 与 docs/docs/awel/why_use_awel.md其中涵盖了分支算子BranchOperator、流式算子StreamifyAbsOperator等和多轮会话场景的编排方法。【免费下载链接】DB-GPTopen-source agentic AI data assistant for the next generation of AI Data products.项目地址: https://gitcode.com/GitHub_Trending/db/DB-GPT创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考