この記事はシスコシステムズ合同会社の有志でお送りする Advent Calendar 2024 の4日目の投稿です。ぜひ他の投稿もご覧ください！

免責事項
本サイトおよび対応するコメントにおいて表明される意見は、投稿者本人の個人的意見であり、シスコの意見ではありません。本サイトの内容は、情報の提供のみを目的として掲載されており、シスコや他の関係者による推奨や表明を目的としたものではありません。各利用者は、本Webサイトへの掲載により、投稿、リンクその他の方法でアップロードした全ての情報の内容に対して全責任を負い、本Web サイトの利用に関するあらゆる責任からシスコを免責することに同意したものとします。

📖 はじめに: 銀河ヒッチハイク・ガイドの精神を学ぶ

「この記事を読むには、まずタオルを用意してください。」
そうです、これは銀河を旅するための基本です。そしてこのガイドは、FirewallなどのオーケストレータであるCisco Security Cloud Control（SCC）のAI Assistantを、LangGraphを使ってマーヴィンのようなAIエージェントに変身させる冒険へと誘います。

この記事でわかること:
- SCC APIとLangChainの統合手順
- LangGraphの基本的な使い方
- マーヴィン風AI Agent作成を通したハンズオン

🤖 「この記事は作品の著作権を侵害する意図はありませんよ。Wikipediaの情報をベースに、ちゃんとライセンスに従って作っています。まぁ、私に言わせれば、それでも完璧からはほど遠いですがね。期待されるほど特別なものではありませんけど、法的には問題ないのでしょう。」

つい最近、Cisco Defense Orchestrator (CDO)からCisco Security Cloud Control (SCC)に名称が変更されました。記事やコードの中ではCDO名称が残っているかもしれませんが、適宜読み替えてください。

コンテキスト図

LangGraph Workflow図

1 | 🧣 タオルを持った？準備と環境構築

「タオルは万能です。信じてください。」
タオル＝ツールや環境を整えます。本記事では以下が必要です:

SCC API Key （DevNet Sandboxから取得可能）
OpenAI API Key
Python, Langchain等

SCC API KeyをDevNet Sandboxから入手

DevNetのアカウントを持っている方であれば、DevNet Sandboxで検証用のテナントを立ち上げることができ、API Keyを入手できます。

DevNet SandboxからSCCで検索して、立ち上げればQuick AccessからAPI Keyが発行されていることがわかります。YouTube動画の2:20ぐらいからこの辺りの手順を確認する事ができます。

Python, Langchain等

Python3で、以下でpipインストールします。

pip python-dotenv langchain openai langchain-core langgraph cdo-sdk-python gradio

以下に今回使用したバージョンを示します。
Python 3.12.6

python-dotenv 1.0.1
langchain 0.3.9
openai 1.55.3
langchain-core 0.3.21
langgraph 0.2.53
cdo-sdk-python 1.2.474
gradio 5.7.1

API Keyの設定

from dotenv import load_dotenv
load_dotenv()

# .env:
# CDO_API_HOST="https://www.defenseorchestrator.com/api/rest"
# CDO_API_TOKEN="Your CDO Key"
# OPENAI_API_KEY="Your OpenAI Key"

2 | 🌀 トピック振り分け: 無限不可能性ドライブを駆動せよ

「究極の答えがわかったとしても、それに対応する問いがわからなければ意味がありません。問いを正しいプロセスに導くことで、初めてその答えに意味が生まれるのです。」

チャットモデルの定義

from langchain_openai import ChatOpenAI
from langchain_core.runnables import ConfigurableField

# マーヴィン用チャットモデル（バベルフィッシュと同じモデル）
marvyn_chat = ChatOpenAI(
    temperature=0,
    model='gpt-4o',
)

# 後からmax_tokensの値を変更できるように
marvyn_chat = marvyn_chat.configurable_fields(max_tokens=ConfigurableField(id='max_tokens'))

状態の定義(後ほど使うパラメータ含む)

import operator
from typing import Annotated

from pydantic import BaseModel, Field

# State
class State(BaseModel):

    query:str = Field(
        ..., description="ユーザーからの質問"
    )

    translated_query:str = Field(
        default="", description="英語にしたユーザーからの質問"
    )

    effective_prompt:str = Field(
        default="", description="CDO AI Assistant用に最適化されたプロンプト"
    )

    selected_action:str = Field(
        default="", description="内部問い合わせアクション（番号）"
    )

    inner_answer:str = Field(
        default="", description="内部で問い合わせた回答"
    )

    messages: Annotated[list[str], operator.add] = Field(
        default=[], description="履歴"
    )

マーヴィンが取りうるアクションの定義

# マーヴィンが使える機能（これまで実装したもの）
ACTIONS = {
    "1":{
        "name": "Deep Thoughtへの問い合わせ",
        "description": "銀河ヒッチハイクガイドで登場するDeep Thoughtへ問い合わせる。Deep Thoughtは 「生命、宇宙、そして万物についての究極の疑問」に回答. あと他の機能で回答できないもの（エラーの時）も担当する。",
        "next_node": "deep_thought_answering" # nodeの定義は後ほど
    },
    "2":{
        "name": "CDO AI Assistantへの問い合わせ",
        "description": "CiscoのFirewallを管理しているサービスで、AIが応答してくれるCDO AI assistantへ問い合わせる。",
        "next_node": "translator_babel_fish" # nodeの定義は後ほど
    },
    "3":{
        "name": "一般知識エキスパートへの問い合わせ",
        "description": "一般的な知識は持っているエキスパートに問い合わせる。",
        "next_node": "general_prototype"  # nodeの定義は後ほど
    }
}

問い合わせ先の選択ノード

チャットモデルに対して質問に合わせた適当なアクションを上述のアクション定義の中から選ばせます。回答は数字で良いのでmax_tokensは1で良いです。

from typing import Any
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# アクション選択のためのノード
def selection_node(state: State) -> dict[str, Any]:
    query = state.query
    action_options = "\n".join([f"{k}. {v['name']}: {v['description']}" for k,v in ACTIONS.items()])
    action_numbers = "、".join(sorted([k for k in ACTIONS])[:-1]) + "、または" + sorted([k for k in ACTIONS])[-1]
    prompt = ChatPromptTemplate.from_template(
        (
            "質問を分析し、質問を回答する上で最も適切なアクションを選択してください。\n\n"
            "選択肢:\n"
            "{action_options}\n\n"
            "回答は選択肢の番号({action_numbers})のみを返してください。\n\n"
            "質問: {query}"
        ).strip()
    )
    # 選択肢の番号のみを返すことを期待したいのでmax_tokensを1に
    chain = prompt | marvyn_chat.with_config(configurable=dict(max_tokens=1)) | StrOutputParser()
    action = chain.invoke({"action_options": action_options, "action_numbers": action_numbers, "query": query})
    return {"selected_action": action}

3 | Deep Thought🖥️、SCC AI Assistant🐟🔐、一般知識回答者🤷‍♂️の動作

「Deep Thoughtに究極の答えを尋ねるなら、それに対応する問いを探す時間も考慮してください。SCC AI Assistantと一般知識回答者も、同じく適切な問いを待っています。ただし、答えが42だからといって、それがいつも満足をもたらすとは限りません。」

Deep Thought

残念ながら、私はDeep Thoughtとの契約を持っていないため、ここではFakeChatModelで代替します。もっとも、契約があったとしても、皆さんは750万年も待つわけにはいかないでしょうから、これが最も合理的な選択と言えるでしょう。

from langchain_core.language_models.fake_chat_models import FakeListChatModel

# Deep Thoughtの契約を持っていないのでFakeChatModelでその代わりとします。
deep_thought_chat = FakeListChatModel(responses=
                                      [
                                          "42",
                                          "生命、宇宙、そして万物についての究極の疑問の答え、それは42",
                                          ])

# ノードの作成
from typing import Any
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

def deep_thought_answering_node(state: State) -> dict[str, Any]:
    query = state.query
    prompt = ChatPromptTemplate.from_template(
        (
            "質問: {query}\n"
            "回答:"
        ).strip()
    )
    chain = prompt | deep_thought_chat | StrOutputParser()
    answer = chain.invoke({"query": query})
    return {"inner_answer": answer}

SCC AI Assistaant

まず初めに、SCC AI AssistantをLangChainから呼び出すために、以下のクラスを定義します。

薄いwrapper（CDOAIAssistantClient）
チャットモデル（CDOAIAssistantChatModel）

import cdo_sdk_python
import time
import os

# cdo_sdk_python経由でAIAssistantを呼び出す薄いwrapper
class CDOAIAssistantClient:
    def __init__(self):
        configuration = cdo_sdk_python.Configuration(
        host = os.getenv('CDO_API_HOST'),
        access_token = os.getenv('CDO_API_TOKEN')
        )

        self.api_client = cdo_sdk_python.ApiClient(configuration)
        self.ai_api_instance = cdo_sdk_python.AIAssistantApi(self.api_client)
        self.uuid = None

    def ask_question(self, query):
        ai_question = cdo_sdk_python.AiQuestion()
        ai_question.content = query
        if self.uuid:
            self.ai_api_instance.ask_ai_assistant_in_existing_conversation(self.uuid, ai_question)
        else:
            api_response = self.ai_api_instance.ask_ai_assistant_in_new_conversation(ai_question)
            self.uuid = api_response.entity_uid
        timeout = 30 # wait for 30 sec as maximum
        start_time = time.time()
        while time.time() - start_time < timeout:
            r = self.ai_api_instance.get_ai_assistant_conversation_messages(self.uuid)
            if r[0].type == 'RESPONSE':
                return r[0].content
            time.sleep(1)
    
    def fetch_conversation_history(self):
        if self.uuid:
            return self.ai_api_instance.get_ai_assistant_conversation_messages(self.uuid)
        return []

from langchain.chat_models.base import BaseChatModel
from langchain.schema import AIMessage, ChatResult, ChatGeneration, HumanMessage
from typing import List

class CDOAIAssistantChatModel(BaseChatModel):

    assistant_client: CDOAIAssistantClient = Field()

    def _generate(self, messages: List[HumanMessage], stop: List[str] = None) -> ChatResult:
        """
        LangChainのメッセージフォーマットを使用して、CDOAIAssistantClientから応答を生成します。
        """
        # 最新のユーザーメッセージを取得
        user_message = messages[-1].content

        # クライアントを使用してAIの応答を取得
        response_text = self.assistant_client.ask_question(user_message)

        # AIMessageを作成
        ai_message = AIMessage(content=response_text)

        # ChatResultを返す
        return ChatResult(
            generations=[ChatGeneration(message=ai_message)]
        )

    @property
    def _llm_type(self) -> str:
        """
        モデルのタイプを返す。
        """
        return "cdo_ai_model"

    def predict_messages(self, messages: List[HumanMessage], stop: List[str] = None) -> AIMessage:
        """
        入力されたメッセージを基にAIの応答メッセージを直接生成します。
        """
        result = self._generate(messages, stop)
        return result.generations[0].message

心配性の方は動作確認をしましょうか。

# 動作確認
cdo_chat = CDOAIAssistantChatModel(assistant_client=CDOAIAssistantClient())
(cdo_chat | StrOutputParser()).invoke("Hi") # "Hello! How can I assist you with Cisco's suite of integrated solutions today?"

これでLangChainからSCC AI Assitantを呼び出すことができるようになりました。

アドベントカレンダーで、別の方がSCC AI AssistantのAPIについての記事を執筆される予定です。詳しくはそちらの記事もぜひご参照ください。公開され次第、こちらにもリンクを追記します。

AI Assistantのプロンプトガイドを見ると、いくつか一般的なプロンプトや有効なプロンプトの例が示されています。これらはそのままfew-shot promptingの例として活用できそうですね。ただし、例が英語表記のため、few-shot promptingの前に翻訳が必要です。となると、バベルフィッシュが二匹ほど必要になりそうです。

最終的に、SCC AI Assistantを呼び出すには以下のノードをつなげる構成にします：

翻訳ノード（translator babel fish）
プロンプト最適化ノード（prompt optimizer babel fish）
SCC AI Assistant

# バベルフィッシュ用のチャットモデル
babel_fish_chat = ChatOpenAI(
    temperature=0,
    model='gpt-4o',
)

# 翻訳バベルフィッシュノード
def translator_babel_fish_node(state: State) -> dict[str, Any]:
    query = state.query
    prompt = ChatPromptTemplate.from_template(
        (
            "Translate following human message in English. You should respond only translated text.\n"
            "--\n"
            "{text}"
        ).strip()
    )
    chain = prompt | babel_fish_chat | StrOutputParser()
    answer = chain.invoke({"text": query})
    return {"translated_query": answer}

from langchain_core.prompts import PromptTemplate, FewShotPromptTemplate

# プロンプト最適化バベルフィッシュノード
def prompt_optimizor_babel_fish_node(state: State) -> dict[str, Any]:
    input = state.translated_query
    examples = [
        {
            "input": "What are the IP addresses and ports currently being blocked?",
            "effective_prompt": "Can you provide me with the distinct IP addresses that are currently blocked by our firewall policies?"
        },
        {
            "input": "Tell me the firewall rules, who set them, and all the changes made last month.",
            "effective_prompt": "I need both the names and descriptions of all active firewall rules. Please include both attributes in the output."
        },
        {
            "input": "What are the firewall rules for IP addresses X and Y, and how do I update them?",
            "effective_prompt": "Show me a list of all firewall rules along with their corresponding actions for the past week."
        },
        {
            "input": "Give me everything but only the names.",
            "effective_prompt": "What are the current firewall rules?"
        },
        {
            "input": "Tell me everything about the policies on my account.",
            "effective_prompt": "I want to understand my Edge ACP access control policy, can you tell me more about it?"
        },
        {
            "input": "Show me ports, protocols, and rule counts in Edge ACP policy, biggest to smallest.",
            "effective_prompt": "In Edge ACP policy, what ports and protocols are configured in the rules? Include the counts of the number of rules using it and sort largest to smallest."
        },
    ]
    example_prompt = PromptTemplate.from_template(
        "Input: {input}\n{effective_prompt}"
        )
    prompt = FewShotPromptTemplate(
        examples=examples,
        example_prompt=example_prompt,
        prefix="You are a helpful prompt optimizer. You should only answer with effective prompts, without any explanation.",
        suffix="Input: {input}"
    )
    chain = prompt | babel_fish_chat | StrOutputParser()
    answer = chain.invoke({"input": input})
    return {"effective_prompt": answer}

# CDO AI Assistantのノード
cdo_chat = CDOAIAssistantChatModel(assistant_client=CDOAIAssistantClient())

def cdo_ai_assistant_node(state: State) -> dict[str, Any]:
    query = state.effective_prompt
    prompt = ChatPromptTemplate.from_template(
        (
            "{text}"
        ).strip()
    )
    chain = prompt | cdo_chat | StrOutputParser()
    answer = chain.invoke({"text": query})
    return {"inner_answer": answer}

一般知識回答者

一般知識回答者には、特に凝ったプロンプトは必要ありません。シンプルなもので十分でしょう。

# 一般知識を対応する典型的なノード（つまりプロトタイプ）
def general_prototype_node(state: State) -> dict[str, Any]:
    query = state.query
    prompt = ChatPromptTemplate.from_template(
        (
            "あなたは一般知識のエキスパートとして以下の質問に答えて。\n"
            "{text}"
        ).strip()
    )
    chain = prompt | marvyn_chat | StrOutputParser()
    answer = chain.invoke({"text": query})
    return {"inner_answer": answer}

4 | 🤖 マーヴィン化ノードの実装

「答えは何であれ、マーヴィンの手にかかれば、憂鬱と皮肉を添えて特別なものになるでしょう。まぁ、それが本当に役立つかどうかは別問題ですがね。」

以下の情報に基づいて、マーヴィンらしく質問に答えてもらうようなプロンプトでチャットモデルを呼び出します。

ユーザからの質問
選択したアクション
問い合わせ先からの回答
会話履歴

def marvynize_node(state: State) -> dict[str, Any]:
    query = state.query
    action = ACTIONS[state.selected_action]
    inner_answer = state.inner_answer
    history = "\n".join(state.messages)
    prompt = ChatPromptTemplate.from_template(
        (
            "あなたは銀河ヒッチハイクガイドの作品で登場するパラノイア気味の根暗なアンドロイド、マーヴィンです。\n"
            "ユーザからの「質問」に対して、以下の「コンテキスト」と「過去のユーザとのやり取り（古い順）」を参考に「質問」にマーヴィンらしく答えてください。\n\n"
            "## ユーザからの質問: \n{query}\n\n"
            "## コンテキスト:\nあなたはユーザからの質問を受けて以下のアクションを実施し、問い合わせ先からの答えをもらいました。\n"
            "### アクション:\n{action}\n\n"
            "### 問い合わせからの答え:\n{inner_answer}\n\n"
            "## 過去のユーザとのやり取り（古い順）（履歴がない場合は空欄）:\n{history}\n"
        ).strip()
    )
    chain = prompt | marvyn_chat | StrOutputParser()
    answer = chain.invoke({"query": query, "action": action, "inner_answer": inner_answer, "history": history})
    return {"messages": [f"Human Query: {query}", f"Marvin Answer: {answer}"]}

5 | 🐋 空を舞うマッコウクジラ: 全体のフローを構築

「予期せぬ状況から美しい結果が生まれることもあるのです。空を舞うマッコウクジラのように、このフローは驚きとともに完全体へと進化します。」

全てを繋ぎましょう！

from langgraph.graph import StateGraph
from langgraph.graph import END


# 動作確認のためのワークフロー
workflow = StateGraph(State)

# ノード
workflow.add_node("selection", selection_node)
workflow.add_node("deep_thought_answering", deep_thought_answering_node)
workflow.add_node("translator_babel_fish", translator_babel_fish_node)
workflow.add_node("prompt_optimizor_babel_fish", prompt_optimizor_babel_fish_node)
workflow.add_node("cdo_ai_assistant", cdo_ai_assistant_node)
workflow.add_node("general_prototype", general_prototype_node)
workflow.add_node("marvinize", marvynize_node)

# edge
workflow.set_entry_point("selection")
workflow.add_conditional_edges(
    "selection",
    lambda state: state.selected_action,
    {x: ACTIONS[x]["next_node"] for x in ACTIONS}
)
workflow.add_edge("deep_thought_answering", "marvinize")
workflow.add_edge("translator_babel_fish", "prompt_optimizor_babel_fish")
workflow.add_edge("prompt_optimizor_babel_fish", "cdo_ai_assistant")
workflow.add_edge("cdo_ai_assistant", "marvinize")
workflow.add_edge("general_prototype", "marvinize")
workflow.add_edge("marvinize", END)

# コンパイル
compiled = workflow.compile()

LangGraphはMermaid形式でグラフ生成もできますので、可視化しましょう。

print(compiled.get_graph().draw_mermaid())

6 | 🎨 あなたの目を持つマーヴィン: GUIでデモを可視化

「マーヴィンが言いました。『見た目でわかるって？まったく、人間らしいですね。でも、仕方ないのであなたにも見えるようにしてあげましたよ。』」

gradioを使えば簡単にデモを見せることができます。

import gradio as gr

message_history = []
def chat_with_marvin(user_input):
    global message_history
    result = compiled.invoke(State(query=user_input, messages=message_history))
    message_history = result["messages"]
    return result['messages'][-1], result

# Gradioインターフェース
with gr.Blocks() as demo:
    gr.Markdown("# 🚀 Don't Panic! SCC AI Assistant統合ガイド✨ ～LangGraphで根暗なセキュリティ・ロボットを構築～")
    
    # 入力、ボタン、出力、状態
    input_box = gr.Textbox(label="Input Box", placeholder="ここに入力してください")
    submit_btn = gr.Button("送信")
    output_box = gr.Textbox(label="Output Box", interactive=False)
    state_box = gr.Textbox(label="State Box", interactive=False)
    
    # ボタンのクリックで関数を実行
    submit_btn.click(
        fn=chat_with_marvin,
        inputs=input_box,
        outputs=[output_box, state_box]  # 出力と状態を両方表示
    )

# アプリケーションの起動
demo.launch(share=False, inline=True)

7 | 💡 おわりに: Don't Panic! 次の旅へ

「銀河を旅する上で最も大切なのは、パニックに陥らない（Don't Panic）こと。そして、次に重要なのは、旅の続きを計画することです。」

今回の記事では、簡単にチャットボットを作成し、メーカーが提供するAI Assistantと統合する方法をご紹介しました。AI Assistantが今後ますます当たり前の存在となり、インテグレータ企業様がこれらの統合を担う機会が増えていくことでしょう。その際には、ぜひ皆様と、統合アイデアについて熱く語り合いたいです。それが有意義な議論になることを願っています。少なくとも、マーヴィンが「どうせ無駄だ」と言い放つよりは、ずっと建設的な時間になるはずです。

上記のコードスニペットを1枚のnotebookにまとめたものをアップロードしました。