Difyの自己完結型外来ガイダンスワークフロー（チャットフロー

注：この記事はDify v0.7.2を使用しています。

このチャットフローは、Webまたは音声対話で患者データを収集できる外来案内用チャットボットの構築方法を示しています。簡単に理解すると、患者情報（年齢、性別、症状）に基づいて、おすすめの診療科を案内するということです。

I.ワークフロー思考

1.ワークフローのスクリーンショット

外来オリエンテーションの完全なワークフローのスクリーンショットを以下に示す：

2.ワークフロー擬似フローチャート

外来患者ガイダンスワークフローの目的は、患者が入力した年齢、性別、症状に基づいて、患者に診療科を推薦することである。最初の対話では、患者が一度に全部または一部の情報を入力する可能性があるため、"Parameter Extractor "ノードを使用して、患者の最初の入力から性別、症状、年齢を抽出する。対話変数 is_first_message は、「変数割り当て」ノードによって 0 に設定される。

3.問題 1

その後の対話では、一度に認識されるフィールド情報は1種類のみであるが、患者は複数のフィールドを提供することができ、フィールドは年齢、性別、症状の順に認識される。

4.問題 2

患者情報に基づいてすでに診療科が推奨されている場合、再度何らかの情報を入力すると、最後の診療科が推奨されます。診療科フィールドが空でないため、ダイレクトレスポンスプロセスが実行されます。

II.セッション変数

セッション変数は、ユーザー設定や対話履歴など、LLMが必要とするコンテキスト情報を格納するために使用されます。読み書き可能です。セッション変数は複数ラウンドの対話シナリオを指向しているため、セッション変数はチャットフロー型（チャットアシスタント -> ワークフローオーケストレーション）アプリケーションでのみ使用できます。

1.セッション変数データタイプ

(1) ストリング

(2) 数字

(3) 対象物

(4) Array[文字列] 文字列の配列

(5) Array[number] 数値の配列

(6) Array[object] オブジェクトの配列

2.セッション変数性格描写

(1) セッション変数は、ほとんどのノード内でグローバルに参照できます；

(2) セッション変数の書き込みには変数代入ノード

(3) セッション変数は読み書き可能です。

3.このワークフローで定義されたセッション変数

このワークフローのために定義された5つのセッション変数は、初回対話かどうか、年齢、性別、症状、診療科である。

III.ワークフロー関連インターフェース

1.ワークフローへのインターフェイスの取得

(1) ソースコードの場所

• ソースコードの場所： dedify-0.7.2
• ソースコードの場所: dedify-0.7.2

(2) ワークフロー操作の取得

http://localhost:5001/console/api/apps/3c309371-54f6-4bfb-894a-4193b189ffa5/workflows/draft。draftは、draftと同じく、ワークフローページでのデバッグを示す。

(3) ワークフロー実装の取得

ワークフローは、ワークフローデータテーブルにレコードを保存する。現在バージョンはドラフトであり、ワークフローがパブリッシュされるたびに、そのテーブルにレコードが挿入される。下図の通りである：

(4) ワークフローのデータ構造

インターフェースによって返されるワークフローのデータ構造を以下に示す：

class Workflow(db.Model):
__tablename__ = 'workflows'
__table_args__ = (
db.PrimaryKeyConstraint('id', name='workflow_pkey'),
db.Index('workflow_version_idx', 'tenant_id', 'app_id', 'version'),
)
id: Mapped[str] = db.Column(StringUUID, server_default=db.text('uuid_generate_v4()'))
tenant_id: Mapped[str] = db.Column(StringUUID, nullable=False)
app_id: Mapped[str] = db.Column(StringUUID, nullable=False)
type: Mapped[str] = db.Column(db.String(255), nullable=False)
version: Mapped[str] = db.Column(db.String(255), nullable=False)
graph: Mapped[str] = db.Column(db.Text)
features: Mapped[str] = db.Column(db.Text)
created_by: Mapped[str] = db.Column(StringUUID, nullable=False)
created_at: Mapped[datetime] = db.Column(db.DateTime, nullable=False, server_default=db.text('CURRENT_TIMESTAMP(0)'))
updated_by: Mapped[str] = db.Column(StringUUID)
updated_at: Mapped[datetime] = db.Column(db.DateTime)
_environment_variables: Mapped[str] = db.Column('environment_variables', db.Text, nullable=False, server_default='{}')
_conversation_variables: Mapped[str] = db.Column('conversation_variables', db.Text, nullable=False, server_default='{}')

2.ワークフローインターフェイスの更新

(1) ソースコードの場所

• ソースの場所：dedify-0.7.2/webapp/components/workflowhooks/use-nodes-sync-draft.ts
• ソースコードの場所：dedify-0.7.2

(2) ワークフロー操作の更新

http://localhost:5001/console/api/apps/3c309371-54f6-4bfb-894a-4193b189ffa5/workflows/draft.

(3) ワークフローのフロントエンド・コードを更新する

DifyワークフローのフロントエンドにはReact Flowフレームワークが使用されているため、ワークフローが変更された際に追加、削除、変更などのPOST操作が行われている可能性が推測される。バックエンドのコンソールのログから、ワークフローが変更されたときにworkflows/draft?_token=::

とおすworkflows/draft?_token=フロントエンドのコードを探しても、それを使う場所はひとつしか見つからなかった。

POSTオペレーションが実行されるので、Payloadパラメータがあるはずだ。

依存関係getPostParams変更が発生すると、非同期に実行される。syncWorkflowDraft(postParams).

(因果関係を示す）syncWorkflowDraft(postParams)実際の呼び出しはPOSTである：

(4) ワークフロー機能の概要

右dify-0.7.2\web\service\workflow.tsその機能をまとめると以下のようになる：

3.ワークフローの実装

への言及Chatflowの作成、更新、実行、削除操作のプロセス実装6]の "Chatflow Implementation "の項を参照。

4.ワークフロー・パブリッシング

(1) ソースコードの場所

• ソースコードの場所： dedify-0.7.2
• ソースコードの場所: dedify-0.7.2
• ソースの場所: dedify-0.7.2

(2) ワークフロー公開インターフェース

ワークフローの公開をクリックすると実行されます。http://localhost:5001/console/api/apps/3c309371-54f6-4bfb-894a-4193b189ffa5/workflows/publishインターフェース

主に、新しいワークフローを作成し、アプリのワークフローイベントをトリガーする。以下に示す：

(3) ワークフロー記録の新規作成

これは主に、バージョンフィールドの内容が 2024-09-07 09:11:25.894535である新しいワークフローレコードを作成することである。以下に示すとおりである：

(4) app_published_workflow_was_updatedイベントの実装

@app_published_workflow_was_updated.connect
def handle(sender, **kwargs):
app = sender
published_workflow = kwargs.get("published_workflow")
published_workflow = cast(Workflow, published_workflow)
dataset_ids = get_dataset_ids_from_workflow(published_workflow)  # 从工作流中获取数据集ID
app_dataset_joins = db.session.query(AppDatasetJoin).filter(AppDatasetJoin.app_id == app.id).all()  # 获取应用数据集关联
removed_dataset_ids = []  # 用于存储移除的数据集ID
if not app_dataset_joins:  # 如果没有应用数据集关联
added_dataset_ids = dataset_ids  # 添加数据集ID
else:  # 如果有应用数据集关联
old_dataset_ids = set()  # 用于存储旧的数据集ID
for app_dataset_join in app_dataset_joins:  # 遍历应用数据集关联
old_dataset_ids.add(app_dataset_join.dataset_id)  # 添加数据集ID
added_dataset_ids = dataset_ids - old_dataset_ids  # 添加数据集ID
removed_dataset_ids = old_dataset_ids - dataset_ids  # 移除数据集ID
if removed_dataset_ids:  # 如果有移除的数据集ID
for dataset_id in removed_dataset_ids:  # 遍历移除的数据集ID
db.session.query(AppDatasetJoin).filter(
AppDatasetJoin.app_id == app.id, AppDatasetJoin.dataset_id == dataset_id
).delete()  # 删除应用数据集关联
if added_dataset_ids:  # 如果有添加的数据集ID
for dataset_id in added_dataset_ids:  # 遍历添加的数据集ID
app_dataset_join = AppDatasetJoin(app_id=app.id, dataset_id=dataset_id)  # 创建应用数据集关联
db.session.add(app_dataset_join)  # 添加应用数据集关联
db.session.commit()  # 提交事务

このコードの機能は app_published_workflow_was_updated アプリケーションの公開ワークフローが更新されたときに、アプリケーションとデータセットの関連関係を更新するためのシグナル。具体的な手順は以下の通り：

• シグナルを送信するアプリケーション・オブジェクトを取得する app ワークフローオブジェクトの更新 published_workflow.
• 各論 get_dataset_ids_from_workflow 更新されたワークフローからデータセットIDコレクションを抽出する関数 dataset_ids.
• データベースに問い合わせて、現在のアプリケーションの AppDatasetJoin 記録
• 追加・削除するデータセットIDを計算する：
  • もし現在 AppDatasetJoin レコードがある場合は、抽出されたすべてのデータセットIDを追加する必要がある。
  • そうでなければ、追加・削除するデータセットIDを計算する。
• 削除の必要性 AppDatasetJoin 記録
• 新規追加 AppDatasetJoin 記録
• データベーストランザクションを送信します。

(5) AppDatasetJoinテーブルの役割

AppDatasetJoinテーブルは、アプリケーション（App）とデータセット（Dataset）の多対多の関係を維持する役割を果たす。各レコードは、アプリケーションとデータセットの関連を表す。具体的なフィールドは以下のとおり：

• id: レコードを一意に識別する主キー。
• app_id: アプリケーションの一意な識別子。
• dataset_id: データセットの一意識別子．
• created_at: レコード作成のタイムスタンプ。
• このテーブルを使えば、どのデータセットが特定のアプリケーションに関連付けられているか、あるいは特定のデータセットがどのアプリケーションに関連付けられているかを照会することができる。

(6) get_dataset_ids_from_workflow()の実装

def get_dataset_ids_from_workflow(published_workflow: Workflow) -> set:  # 从工作流中获取数据集ID
dataset_ids = set()  # 用于存储数据集ID
graph = published_workflow.graph_dict  # 获取工作流图
if not graph:  # 如果没有图
return dataset_ids  # 返回空集合
nodes = graph.get("nodes", [])  # 获取图中的节点
# fetch all knowledge retrieval nodes  # 获取所有知识检索节点
knowledge_retrieval_nodes = [
node for node in nodes if node.get("data", {}).get("type") == NodeType.KNOWLEDGE_RETRIEVAL.value
]  # 获取所有知识检索节点
if not knowledge_retrieval_nodes:  # 如果没有知识检索节点
return dataset_ids  # 返回空集合
for node in knowledge_retrieval_nodes:  # 遍历知识检索节点
try:
node_data = KnowledgeRetrievalNodeData(**node.get("data", {}))  # 获取节点数据
dataset_ids.update(node_data.dataset_ids)  # 更新数据集ID
except Exception as e:  # 如果出现异常
continue
return dataset_ids

get_dataset_ids_from_workflow関数の主な機能は、与えられたワークフローオブジェクトからすべての知識検索ノードの関連データセットIDを抽出することである：

• データセットIDを格納するために、空の dataset_ids コレクションを初期化する。
• ワークフローグラフのグラフ構造を取得する。
• グラフ構造が空の場合、空の dataset_ids コレクションを返す。
• グラフ内のすべてのノードを取得する。
• KNOWLEDGE_RETRIEVAL knowledge_retrieval_nodes 型のすべてのノードをフィルタリングします。
• これらの知識検索ノードを反復処理し、データセットIDを抽出し、dataset_idsコレクションで更新する。
• 抽出されたすべてのデータセットIDのコレクションを返す。

5.ワークフローその他のインターフェース

ワークフロー・オーケストレーションの会話型アプリケーションAPIは非常に明確に説明されている。

書誌