企业官网建设流程全解析

DB-GPT AWEL 入门实战：三步搭建 LLM 工作流 DAG 并调用模型生成 SQL

【免费下载链接】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 官方 AWEL（Agentic Workflow Expression Language）Cookbook 中的 QuickStart 文档，带你完整跑通 AWEL 的最小工作流：使用DAG编排 Prompt 构建、请求构建与 LLM 调用三类算子，通过 OpenAI 兼容接口让模型根据方言与表名生成 SQL 查询。读完本篇，你将掌握 AWEL 工作流的搭建范式（with DAG声明算子、>>连线、task.call()触发执行），并理解每个算子底层的参数含义与数据处理逻辑，为后续编写多轮对话、RAG、数据分析等更复杂的工作流打下基础。

什么是 AWEL

AWEL 是 DB-GPT 专为大模型应用开发设计的智能体工作流表达语言。其官方定义可见 AWEL 模块入口：

Agentic Workflow Expression Language(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 让你用 Python 代码以“图”的方式声明工作流节点（算子/Task）及其依赖关系，框架负责异步调度执行。本文的示例即是最典型的 AWEL 工作流：三条 Task 串联成一条线性 DAG。

环境准备与安装

按照官方文档，先安装dbgpt及示例所需的openai依赖：

pip install dbgpt --upgrade pip install openai

创建 Python 文件simple_sdk_llm_example_dag.py，写入如下内容（与文档示例完全一致）：

import asyncio from dbgpt.core.awel import DAG from dbgpt.core.operators import ( PromptBuilderOperator, RequestBuilderOperator, ) from dbgpt.model.proxy import OpenAILLMClient from dbgpt.model.operators import LLMOperator with DAG("simple_sdk_llm_example_dag") as dag: prompt_task = PromptBuilderOperator( "Write a SQL of {dialect} to query all data of {table_name}." ) model_pre_handle_task = RequestBuilderOperator(model="gpt-3.5-turbo") llm_task = LLMOperator(OpenAILLMClient()) prompt_task >> model_pre_handle_task >> llm_task output = asyncio.run( llm_task.call({ "dialect": "MySQL", "table_name": "users" } )) print(output)

运行前配置 OpenAI API 的环境变量：

export OPENAI_API_KEY=sk-xx export OPENAI_API_BASE=https://xx:80/v1

最后执行脚本：

python simple_sdk_llm_example_dag.py

预期输出形如（文档给出的示例结果）：

ModelOutput(text='SELECT * FROM users;', error_code=0, model_context=None, finish_reason=None, usage={'completion_tokens': 5, 'prompt_tokens': 19, 'total_tokens': 24}, metrics=None)

逐行解析示例代码

with DAG("simple_sdk_llm_example_dag") as dag:

DAG定义在 DAG 基类，它以上下文管理器的形式开启一个工作流定义作用域：块内实例化的每个算子（Operator）都会自动注册为一张 Task 节点，DAG对象同时携带执行上下文（DAGContext）。示例中dag即当前工作流实例，用于后续查询与执行。

prompt_task >> model_pre_handle_task >> llm_task

>>是 AWEL 中声明数据流向的操作符，含义是“前一个 Task 的输出作为后一个 Task 的输入”。这条链式写法等价于一个三节点的线性有向无环图：

1. prompt_task：把提示词模板 + 输入字典渲染成模型消息列表；
2. model_pre_handle_task：把消息包装成标准ModelRequest（附带模型名等推理参数）；
3. llm_task：把ModelRequest交给 LLM 客户端执行推理，产出ModelOutput。

llm_task.call({...})

调用末端 Task 的call方法并传入业务输入字典，框架会从该 Task 回溯上游依赖并驱动整条链路异步执行，asyncio.run负责在同步脚本中等待结果。这里传入的{"dialect": "MySQL", "table_name": "users"}会被最上游的PromptBuilderOperator用作模板变量。

三大算子的源码级解析

PromptBuilderOperator：模板变量渲染为消息列表

PromptBuilderOperator实现于 prompt_operator.py，其签名类型是MapOperator[Dict[str, Any], List[ModelMessage]]，即输入一个字典、输出一组消息。

从源码结构看，它的构造逻辑（__init__，L288-L305）会自动归一化各种 prompt 形式：传入普通字符串时，会包装为只含一条 Human 消息的ChatPromptTemplate；传入PromptTemplate、HumanPromptTemplate或整个ChatPromptTemplate（可含 system/human 多角色消息）也能直接使用。因此示例中直接传字符串"Write a SQL of {dialect} to query all data of {table_name}."是最简用法。

执行时，merge_prompt调用format_prompt（L106-L126），核心逻辑是：

• 取出模板的input_variables，只保留输入字典中与之匹配的键（即{dialect}、{table_name}）；
• 调用prompt.format_messages(**pass_kwargs)完成字符串插值；
• 通过ModelMessage.from_base_messages转换为内部消息结构。

所以示例中当call传入dialect=MySQL、table_name=users时，上游产出的就是单条 human 消息：Write a SQL of MySQL to query all data of users.。

RequestBuilderOperator：组装标准 ModelRequest

RequestBuilderOperator实现于 llm_operator.py，类型签名为MapOperator[RequestInput, ModelRequest]。它的构造参数在源码中定义得非常完整：

其map方法（L128-L194）展示了完整的请求归一化过程：上游传来的List[ModelMessage]会被装入messages字段；若请求字典中缺少model，则回填构造时的self._model（也就是gpt-3.5-turbo），若最终仍为空则抛出ValueError("model is not set")；temperature、max_new_tokens、context_len同理按需回填。此外它还自动补全context字段（ModelRequestContext，可携带stream、conv_uid、span_id、chat_mode等上下文信息），最后只保留ModelRequest数据类中声明的字段构造最终请求。这个算子本质上是“模型无关的适配层”——让上游业务数据与下游任意 LLM 客户端解耦。

LLMOperator：执行推理并产出 ModelOutput

LLMOperator实现于 model/operators/llm_operator.py，输入为ModelRequest、输出为ModelOutput（见其ViewMetadata中声明的 IO 字段）。其文档字符串说明了客户端的自动降级策略：若llm_client传None，会先尝试连接 DB-GPT 部署的模型服务集群，连接失败则回退使用OpenAILLMClient。示例中我们显式传入OpenAILLMClient()，即直连 OpenAI 兼容 API。

示例中另一个可回退的相邻算子是StreamingLLMOperator（同文件 L115 起），用于流式场景，基本结构相同。

OpenAILLMClient：环境变量从哪来

OpenAILLMClient定义于 chatgpt.py，构造参数包括api_key、api_base、api_type、api_version、model、proxies、timeout（默认 240 秒）、model_alias（默认gpt-4o-mini）、context_length（默认 8192）等。

文档要求设置OPENAI_API_KEY与OPENAI_API_BASE环境变量的依据在源码中可以找到：同文件的 OpenAICompatibleDeployModelParameters 中，api_base的默认值为${env:OPENAI_API_BASE:-https://api.openai.com/v1}（未设置时回落到官方 OpenAI 地址），api_key的默认值为${env:OPENAI_API_KEY}。也就是说：

• 设置OPENAI_API_BASE后，该客户端可对接任何 OpenAI 兼容服务（本地 vLLM、Ollama 代理、第三方网关等），这正是文档示例中https://xx:80/v1这种自定义地址的用途；
• 两个环境变量不设置时，客户端也会尝试通过参数或系统配置获取密钥。

最终输出 ModelOutput 的字段含义

示例打印的ModelOutput定义于 interface/llm.py，各字段含义如下：

• content/text：模型生成的文本（本例为SELECT * FROM users;），支持文本与多模态内容；
• error_code：推理错误码，0 表示成功；
• model_context：透传的模型上下文，默认None；
• finish_reason：结束原因；
• usage：token 用量统计，如示例中的{'completion_tokens': 5, 'prompt_tokens': 19, 'total_tokens': 24}；
• metrics：推理性能指标，可选。

从示例到完整应用

上述最小工作流验证了 AWEL 的三个核心要素：声明（with DAG）— 连线（>>）— 触发（call）。在此基础上，你可以按以下路径继续深入当前仓库：

• 更多工作流实战示例（含多轮对话、RAG、SQL + RAG Schema Linking、数据分析等）见 AWEL Cookbook 目录，其中 first_rag_with_awel.md 与 sql_awel_use_rag_and_schema_linking.md 展示了与本文相同的三算子模式如何扩展为带检索的工作流；
• 系统化的算子教程（Map/Join/Branch/Streamify 等算子、HTTP 触发器、自定义算子）见 AWEL Tutorial 目录；
• 仓库内可直接运行的 SDK 级参考脚本位于 examples/sdk/，包括 simple_sdk_llm_example.py（无 DAG 的直接 SDK 调用对照）与 chat_data_with_awel.py；
• 若想理解 AWEL 的设计动机与适用场景，可阅读 why_use_awel.md 与 awel.md。

小结

本文以 DB-GPT AWEL Cookbook 的 QuickStart 为基础，跑通了一个“提示词构建 → 请求构建 → LLM 调用”的最小 DAG，并结合源码说明了PromptBuilderOperator的模板变量过滤机制、RequestBuilderOperator的请求归一化与参数默认值（model/temperature/max_new_tokens/context_len）、LLMOperator的客户端策略，以及OpenAILLMClient通过OPENAI_API_KEY/OPENAI_API_BASE环境变量对接任意 OpenAI 兼容服务的方式。掌握这条链路后，你就可以把检索、分支、流式输出等算子组合进来，构建完整的 LLM 数据应用工作流。

【免费下载链接】DB-GPTopen-source agentic AI data assistant for the next generation of AI + Data products.项目地址: https://gitcode.com/GitHub_Trending/db/DB-GPT