🐱

LangChain, OpenAI API, DuckDBを使って、SQL Agentを実装するチュートリアル

Naoto Shibata

2025/03/11に公開

Langchain, OpenAI API, DuckDBを使用してSQL Agentを実装し、Morphを使用してアプリ化する方法を解説します。LLMで生成したSQLをDuckDBに実行し、その結果を用いて質問に回答するアプリを作成します。

LangChainは、LLMを活用したアプリケーションを開発するためのコンポーネント、サードパーティの統合を備えた非常に優れたフレームワークです。

この記事では、Langchainを用いてAIワークフローの構築、DuckDBを用いてCSVデータへのSQL発行をすることでAIエージェントのワークフローを構築します。

今回は、以下のCSVに含まれるウェブ広告におけるトラフィックと発注成果のデモデータに対して、SQLを用いたQ&Aを行うエージェントを実装します。

Traffic_Orders_Demo_Data.csv

本チュートリアルの成果物は以下のリポジトリのSQL Agent Appで公開しています。

環境のセットアップ

morph new コマンドで新しいプロジェクトを立ち上げます。この時パッケージマネージャーはpoetryを選択することを推奨しています。

morph new sql-agent-app

pyproject.tomlには、以下のようにdependenciesを設定します。DuckDBの実行はmorph-dataに含まれる execute_sql で行うため、個別にインストールする必要はありません。

[tool.poetry.dependencies]
python = "<3.13,>=3.9"
morph-data = "0.2.0"
langchain = "0.3.16"
langchain-core = "0.3.32"
langchain-openai = "0.3.2"

このチュートリアルではOpenAI APIを使うので、 .env にAPI KEYをセットしてください。

OPENAI_API_KEY=[Put your API KEY]

デモデータとして使用するCSVは、 data/Traffic_Orders_Demo_Data.csv に保存をしてください。

フロントエンドの構築

Morphでは、 src/pages 以下にmdxファイルを配置してフロントエンドを構築ができます。

今回は<Chat />コンポーネントを使用して、Langchainで構築したAIとチャットができる画面を構築します。データを見ながら質問をしたいので、横並びで<DataTable />を用いてデータを表形式で表示しています。

<Grid />やTailwind CSSのユーティリティクラスを使用して、レイアウトも調整しています。

# Q&A Agent over SQL

You can ask questions to the SQL Agent about the data and get the answer in SQL.

Only admin user can ask questions to the SQL Agent.

<Grid cols={2} className="py-4">
    <div>
        <DataTable loadData="example_data" height={400} />
    </div>
    <div className="p-4 bg-gray-50 rounded-lg shadow-sm border border-gray-200 hover:shadow-md transition-shadow">
        <Chat postData="sql_agent" />
    </div>
</Grid>

これで、ターミナルで morph serve を実行して、 http://localhost:8080 にアクセスをすると画面が確認できます。

chat

Langchainを使用して、Text to SQLの実行を伴うQ&Aエージェントを実装する

LangchainのRunnable APIとStructuredOutputParserを用いてユーザーの質問を回答するために必要なSQLを生成します。そのSQLをDuckDBで実行した結果を活用することで、再度Runnable APIを使用してユーザーの質問に対する回答をするAIエージェントを構築します。

MDXの<Chat />からは prompt, thread_id が自動的に送信されてきます。

prompt: ユーザーが入力したメッセージ
thread_id: Chat UIで表示されているスレッドを一意に特定するID (今回は使用しない)

prompt を受け取ってLangchainのinvoke関数に渡すmessgesで使用することでAIエージェントとフロントエンドの連携を実現します。

Langchainのユーザーの質問に合わせたSQLの生成・データ取得

最初に、Morphのフレームワークで扱うことができるように @morph.func を付与した関数を作成して、LangchainのOpenAIのモデルを使用するインスタンスを初期化します。

import morph
from morph import MorphGlobalContext
from morph_lib.types import MorphChatStreamChunk
from langchain_openai import ChatOpenAI

@morph.func
def sql_agent(
    context: MorphGlobalContext,
) -> Generator[MorphChatStreamChunk, None, None]:
    chat = ChatOpenAI(
        model="gpt-4o",
        temperature=1,
        streaming=False
    )

今回使用するCSVに関して、ファイルのパスやデータの説明をシステムプロンプトとして記述します。 format_instructions を与えることによって、StructuredOutputParserからSQL出力のフォーマット定義を渡せるようにします。

SYSTEM_TEMPLATE = """Please execute SQL queries on a table named `./data/Traffic_Orders_Demo_Data.csv` in DuckDB with the following schema:
date: text - date
source: text - traffic source (Coupon, Google Organic など)
traffic: int - traffic count
orders: int - order count

This table contains traffic and order data for the marketing campaigns.

As a source, you have the following data:
- Coupon
- Google Organic
- Google Paid
- TikTok Ads
- Meta Ads
- Referral

Generate a SQL query to answer the user's question.
{format_instructions}
"""

StructuredOutputParserを使用して、生成されたテキストを構造化データ（JSON）としてパース可能にします。

そして、LangChainのRunnable APIを活用し、SQLを生成するためのデータフローを構築します。

# Setup the SQL query output parser
sql_schema = ResponseSchema(name="sql", description="The SQL query to execute")
output_parser = StructuredOutputParser.from_response_schemas([sql_schema])
format_instructions = output_parser.get_format_instructions()

prompt = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_TEMPLATE),
    ("human", "{question}")
])
# Create the chain for SQL generation
chain = (
    {"question": RunnablePassthrough(), "format_instructions": lambda _: format_instructions}
    | prompt
    | chat
    | StrOutputParser()
    | output_parser
)

作成したデータフローを実行してSQLを生成します。 stream_chat 関数を使用して生成されたSQLは、チャットにストリーミングで返却します。

from morph_lib.stream import stream_chat

# Generate SQL query
result = chain.invoke(context.vars["prompt"])
sql = result["sql"]
		# display generated sql
yield stream_chat(f"""
### SQL
{sql}
""")

次に、生成されたSQLを実行します。 execute_sql の第2引数に”DUCKDB”と指定することでDuckDBのエンジンを使用することができます。

実行結果はDataFrameとして返却されるので、LLMに渡すためにMarkdownの文字列に変換します。

from morph_lib.database import execute_sql

# Execute SQL and get results
data = execute_sql(sql, "DUCKDB")
data_md = data.to_markdown(index=False)

SQLの結果を用いた回答生成

取得した結果を用いて再度LangchainのRunnable API を活用し、プロンプトの作成・質問処理・ストリーミング出力をすることにより、ユーザーの質問に回答するエージェントを作成します。

# Create analysis prompt with results
analysis_prompt = ChatPromptTemplate.from_messages([
    ("system", f"""Please answer in markdown format.
You can use the following data:
{data_md}

The data is from the following SQL query:
{sql}
"""),
    ("human", "{question}")
])

# Create analysis chain
analysis_chain = (
    {"question": RunnablePassthrough()}
    | analysis_prompt
    | chat
)

# stream analysis result
for chunk in analysis_chain.stream(context.vars["prompt"]):
    if chunk.content:
        yield stream_chat(chunk.content)

Pythonコード全体

from typing import Generator

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough
from langchain.output_parsers import ResponseSchema, StructuredOutputParser
from morph_lib.database import execute_sql
from morph_lib.stream import stream_chat
from morph_lib.types import MorphChatStreamChunk

import morph
from morph import MorphGlobalContext

SYSTEM_TEMPLATE = """Please execute SQL queries on a table named `./data/Traffic_Orders_Demo_Data.csv` in DuckDB with the following schema:
date: text - date
source: text - traffic source (Coupon, Google Organic など)
traffic: int - traffic count
orders: int - order count

This table contains traffic and order data for the marketing campaigns.

As a source, you have the following data:
- Coupon
- Google Organic
- Google Paid
- TikTok Ads
- Meta Ads
- Referral

Generate a SQL query to answer the user's question.
{format_instructions}
"""

@morph.func
def sql_agent(
    context: MorphGlobalContext,
) -> Generator[MorphChatStreamChunk, None, None]:
    chat = ChatOpenAI(
        model="gpt-4o",
        temperature=0,
        streaming=False,
    )

    # Setup the SQL query output parser
    sql_schema = ResponseSchema(name="sql", description="The SQL query to execute")
    output_parser = StructuredOutputParser.from_response_schemas([sql_schema])
    format_instructions = output_parser.get_format_instructions()

    prompt = ChatPromptTemplate.from_messages([
        ("system", SYSTEM_TEMPLATE),
        ("human", "{question}")
    ])

    # Create the chain for SQL generation
    chain = (
        {"question": RunnablePassthrough(), "format_instructions": lambda _: format_instructions}
        | prompt
        | chat
        | StrOutputParser()
        | output_parser
    )

    # Generate SQL query
    result = chain.invoke(context.vars["prompt"])
    sql = result["sql"]
    # display generated sql
    yield stream_chat(f"""
### SQL
{sql}
""")

    # Execute SQL and get results
    data = execute_sql(sql, "DUCKDB")
    data_md = data.to_markdown(index=False)
    # Create analysis prompt with results
    analysis_prompt = ChatPromptTemplate.from_messages([
        ("system", f"""Please answer in markdown format.
You can use the following data:
{data_md}

The data is from the following SQL query:
{sql}
"""),
        ("human", "{question}")
    ])

    # Create analysis chain
    analysis_chain = (
        {"question": RunnablePassthrough()}
        | analysis_prompt
        | chat
    )

    # stream analysis result
    for chunk in analysis_chain.stream(context.vars["prompt"]):
        if chunk.content:
            yield stream_chat(chunk.content)

デプロイしてアプリを共有する

Morphのダッシュボードでは、Github連携を使用してデプロイを自動化する機能がサポートされています。

Githubにコードをプッシュし、ダッシュボードからリポジトリを選択するだけで自動的に認証がかかった状態のみアクセスが可能な社内アプリとしてデプロイができます。

参考

デプロイが完了したら、ダッシュボードの”Open”からアプリにアクセスしてみましょう。作成したSQL Agentのアプリをチームに簡単に共有することができます！

MorphテックブログPublication

AIアプリ・データアプリのためのPython + Markdownフレームワークの Morphのテックブログです。

Discussion

ログインするとコメントできます