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 官方 AWELAgentic Workflow Expression LanguageCookbook 中的 QuickStart 文档带你完整跑通 AWEL 的最小工作流使用DAG编排 Prompt 构建、请求构建与 LLM 调用三类算子通过 OpenAI 兼容接口让模型根据方言与表名生成 SQL 查询。读完本篇你将掌握 AWEL 工作流的搭建范式with DAG声明算子、连线、task.call()触发执行并理解每个算子底层的参数含义与数据处理逻辑为后续编写多轮对话、RAG、数据分析等更复杂的工作流打下基础。什么是 AWELAWEL 是 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(modelgpt-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_KEYsk-xx export OPENAI_API_BASEhttps://xx:80/v1最后执行脚本python simple_sdk_llm_example_dag.py预期输出形如文档给出的示例结果ModelOutput(textSELECT * FROM users;, error_code0, model_contextNone, finish_reasonNone, usage{completion_tokens: 5, prompt_tokens: 19, total_tokens: 24}, metricsNone)逐行解析示例代码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 的输入”。这条链式写法等价于一个三节点的线性有向无环图prompt_task把提示词模板 输入字典渲染成模型消息列表model_pre_handle_task把消息包装成标准ModelRequest附带模型名等推理参数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_promptL106-L126核心逻辑是取出模板的input_variables只保留输入字典中与之匹配的键即{dialect}、{table_name}调用prompt.format_messages(**pass_kwargs)完成字符串插值通过ModelMessage.from_base_messages转换为内部消息结构。所以示例中当call传入dialectMySQL、table_nameusers时上游产出的就是单条 human 消息Write a SQL of MySQL to query all data of users.。RequestBuilderOperator组装标准 ModelRequestRequestBuilderOperator实现于 llm_operator.py类型签名为MapOperator[RequestInput, ModelRequest]。它的构造参数在源码中定义得非常完整参数类型默认值说明modelstrNone模型请求默认使用的模型名示例中为gpt-3.5-turbotemperaturefloatNone采样温度UI 滑块范围 0.0~2.0步长 0.1max_new_tokensintNone最大生成 token 数context_lenintNone上下文长度其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执行推理并产出 ModelOutputLLMOperator实现于 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透传的模型上下文默认Nonefinish_reason结束原因usagetoken 用量统计如示例中的{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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考