🧑‍💻

第4回に向けて、過去3回の Zenn Hackathon 作品を検索・分析できるシステムを自作した

に公開

はじめに

こんにちは!Alexです。 AIエンジニアをしていて、普段は個人開発や受託開発案件をやっています。

今回は Zenn AI Agent Hackathon(第1〜3回)の約400作品を検索・分析できるシステムを開発した話をシェアしたいと思います。

🤔 なぜ作ったのか?

第4回 Zenn AI Agent Hackathon が開催されることを知り、「よし、参加しよう!」と意気込んだものの...いざ考え始めようとすると、こんな疑問が浮かんできました。

「似たような方向性のプロジェクトはあった?」 「受賞作品ってどんな傾向があるんだろう?」

Zenn AI Agent Hackathonはこれまで3回開催され、累計393作品が投稿されています。各作品はZennの記事として公開されており、技術スタック・実装アプローチ・アイデアなど、AIエージェント開発の宝庫とも言えるデータが眠っています。

しかし、問題がありました:

  • 📚 データが分散している - 各作品は個別のZenn記事として存在し、横断的に検索する手段がない
  • 🔎 探しにくい - 「RAGを使った作品は?」「受賞作品の傾向は?」といった質問に答えられるツールがない
  • 📊 分析できない - 技術トレンドや統計情報を俯瞰的に見る方法がない

...ということで、全393作品のデータを収集・整理し、検索・分析できるシステムを構築してみました。

できたシステム

本システムは、検索を実現する Agentic RAG と、外部ツールとの連携を可能にする MCP サーバーの2つの側面を持っています。

まずはクイックスタートでシステムを動かしてみましょう。その後でシステム構成を解説していきます。

クイックスタート

まずは動かしてみたい方向けに、セットアップ手順を紹介します。

リポジトリのクローン

git clone https://github.com/yh00722/zenn-hackathon-search
cd zenn-hackathon-search

初回セットアップ:データベース構築

初回起動時は、SQLite と ChromaDB のベクトルインデックスを構築する必要があります。

cd backend
uv sync

# 1. CSVデータをSQLiteにインポート
uv run python -m services.importer

# 2. ChromaDBにベクトルインデックスを構築
uv run python -c "
from services.rag import get_rag_service
rag = get_rag_service()
rag.index_projects()   # 本文チャンクのインデックス
rag.index_summaries()  # 要約のインデックス
"

Note: この処理は初回のみ必要です。OpenAI API を使用してベクトル化するため、環境変数に OPENAI_API_KEY(または Azure OpenAI の設定)が必要です。

Web UI(RAG Chat)を使う

# バックエンド起動
cd backend
uv sync
uv run uvicorn main:app --reload

# フロントエンド起動(別ターミナル)
cd frontend
npm install
npm run dev

ブラウザで http://localhost:3000 にアクセスすると、以下の機能が使えます:

  • プロジェクト一覧 - 393作品をブラウズ、回別フィルタ、キーワード検索
  • ランキング - いいね数順でトップ作品を確認
  • RAG Chat - 自然言語で質問すると、関連プロジェクトを検索して回答を生成

MCP サーバーを使う

ここで、 Claude Desktop を例として、MCP サーバーを起動する手順を説明します。 ~/Library/Application Support/Claude/claude_desktop_config.json に以下を追加:

OpenAI を使用する場合:

{
  "mcpServers": {
    "zenn-hackathon": {
      "command": "/opt/homebrew/bin/uv",
      "args": [
        "run",
        "--directory", "/path/to/zenn-hackathon/backend",
        "python", "-m", "mcp_server.server"
      ],
      "env": {
        "OPENAI_PROVIDER": "openai",
        "OPENAI_API_KEY": "sk-your-openai-api-key",
        "OPENAI_CHAT_MODEL": "gpt-5.2",
        "OPENAI_EMBEDDING_MODEL": "text-embedding-3-large"
      }
    }
  }
}

Azure OpenAI を使用する場合:

{
  "mcpServers": {
    "zenn-hackathon": {
      "command": "/opt/homebrew/bin/uv",
      "args": [
        "run",
        "--directory", "/path/to/zenn-hackathon/backend",
        "python", "-m", "mcp_server.server"
      ],
      "env": {
        "OPENAI_PROVIDER": "azure",
        "AZURE_OPENAI_API_KEY": "your-api-key",
        "AZURE_OPENAI_ENDPOINT": "https://your-resource.openai.azure.com/",
        "AZURE_OPENAI_API_VERSION": "2024-12-01-preview",
        "AZURE_CHAT_DEPLOYMENT": "gpt-5.2",
        "AZURE_EMBEDDING_DEPLOYMENT": "text-embedding-3-large"
      }
    }
  }
}

Claude Desktop を再起動すると、MCP ツールが使えるようになります。

デモ動画

実際の動作は以下の投稿動画をご覧ください。

https://x.com/Alexand66834533/status/2004795777307349321?s=20

これからシステムの詳細を解説していきます。

データ収集と前処理

検索システムを作る前に、まずデータを集める必要があります。

データソース

過去3回のハッカソンの作品は、以下のページで公開されています:

収集方法

各ハッカソンページから、BeautifulSoup でスクレイピングしました。

# 収集した情報
- プロジェクト名、URL、作者名
- 説明文(description)
- いいね数、ブックマーク数
- 受賞情報(award_name, award_comment)
- 記事本文(content_raw)

前処理パイプライン

収集したデータをそのまま使うのではなく、検索しやすい形に加工しています。

ポイント:

  • 2種類のベクトルストア - 要約ベクトル(プロジェクト探索用)と本文ベクトル(説明文検索用)を分離
  • LLM要約 - 記事本文を300文字程度に要約し、プロジェクトの概要を把握しやすく
  • チャンク分割 - 長い記事は500文字単位で分割し、関連する段落を検索可能に

システムアーキテクチャ

全体構成

まずはシステム全体の構成を見てみましょう。

MCPツールの概要

MCPサーバーでは 15種類のツール を提供しています。以下は代表的なツールです。

Web UI(Next.js 15)

MCP サーバーとは別に、Web UI も用意しています。ブラウザから直接アクセスでき、以下の機能を提供しています。

プロジェクト一覧・検索

  • キーワード検索 - プロジェクト名・説明文から検索
  • 回別フィルタ - 第1回/第2回/第3回で絞り込み
  • ランキング表示 - いいね数順でソート

RAG Chat

ChatGPT のような対話インターフェースで、自然言語で質問できます。

// 例:チャットでよく使われるクエリ
const EXAMPLE_QUERIES = [
    "受賞作品の共通点は何ですか?",
    "いいね数トップ5のプロジェクトは?",
    "Flutterを使った作品を教えて",
    "ヘルスケア分野のプロジェクトは?",
];

特徴:

  • ストリーミングレスポンス - トークン単位でリアルタイム表示
  • ソース表示 - 回答の根拠となったプロジェクトへのリンク
  • 日本語 IME 対応 - Enter で確定しても送信されない

MCPサーバーの実装

FastMCPでの構築

MCPサーバーは FastMCP 2.0 を使って構築しています。デコレータベースでシンプルに書けるのがポイントです。
MCP サーバーで定義できる主な要素:

要素 用途 デコレータ
Tool LLM が呼び出せる関数 @mcp.tool
Resource 静的なデータや説明文 @mcp.resource
Prompt 再利用可能なプロンプトテンプレート @mcp.prompt

今回は主に Tool を使って検索機能を実装しています。

from fastmcp import FastMCP

# サーバー作成
mcp = FastMCP(
    "Zenn Hackathon",
    instructions="""
    このサーバーは Zenn AI Agent Hackathon(第1〜3回)の
    約400作品を検索・分析するためのツールを提供します。
    """
)

基本ツールの例:search_projects

基本ツールはSQLiteに直接クエリを投げるシンプルな実装です。

@mcp.tool
def search_projects(
    query: str = "",
    edition: int | None = None,
    winner_only: bool = False,
    team_only: bool = False,
    limit: int = 20
) -> list[dict]:
    """
    プロジェクトをキーワードで検索します。

    Args:
        query: 検索キーワード(空の場合は全件)
        edition: ハッカソン届次 (1, 2, 3)
        winner_only: 受賞作品のみ
        team_only: チーム作品のみ
        limit: 最大取得件数
    """
    sql = """
        SELECT id, project_name, url, author_name,
               description, likes, is_winner, award_name
        FROM projects
        WHERE 1=1
    """
    params = []

    if query:
        sql += " AND (project_name LIKE ? OR description LIKE ?)"
        params.extend([f"%{query}%", f"%{query}%"])

    if edition is not None:
        sql += " AND hackathon_id = ?"
        params.append(edition)

    if winner_only:
        sql += " AND is_winner = 1"

    sql += " ORDER BY likes DESC LIMIT ?"
    params.append(limit)

    with db.get_connection() as conn:
        rows = conn.execute(sql, tuple(params)).fetchall()
        return [dict(row) for row in rows]

高度なツール:セマンティック検索

ベクトル検索は ChromaDB と OpenAI Embedding を組み合わせて実現しています。

@mcp.tool
def semantic_search_summary(
    query: str,
    k: int = 6,
    edition: int | None = None,
    winner_only: bool = False
) -> list[dict]:
    """
    【推奨】プロジェクト単位の主題検索

    「〇〇関連のプロジェクト」「△△を使った作品」など
    主題・テーマで類似プロジェクトを探したい場合に使用。
    """
    from services.rag import get_rag_service
    rag = get_rag_service()

    # フィルター構築
    chroma_filter = {}
    if edition is not None:
        chroma_filter["edition"] = edition
    if winner_only:
        chroma_filter["is_winner"] = True

    # 要約ベクトルストアで検索
    docs = rag.summary_vectorstore.similarity_search(
        query=query,
        k=k,
        filter=chroma_filter if chroma_filter else None
    )

    return [{
        "project_name": doc.metadata.get("project_name"),
        "url": doc.metadata.get("url"),
        "summary": doc.page_content,
        "is_winner": doc.metadata.get("is_winner", False),
    } for doc in docs]

ポイント:

  • ChromaDB のメタデータフィルタ - 回別や受賞作品での絞り込みが可能
  • Embedding は事前計算済み - 検索時は類似度計算のみで高速
  • 返り値は必要最小限 - URL とサマリーのみ返し、コンテキストを節約

Agentic RAGの実装

LangGraphによるReActパターン

従来の RAG は「質問 → 検索 → 回答」の1回きりの流れですが、Agentic RAG はエージェントが自律的に検索戦略を決定し、必要に応じて複数回検索を繰り返します。

これを実現するために、ReActパターンを採用しています:

  1. Reasoning - LLM が「どのツールを使うべきか」を判断
  2. Acting - 選んだツールを実行
  3. Observation - 結果を確認し、十分な情報が得られたか判断
  4. 繰り返し - 不十分なら別のツールで再検索

LangGraph を使うと、このループを宣言的に構築できます。

状態の定義

LangGraphの状態は TypedDict で型安全に定義します。

from typing import Annotated, TypedDict, Literal
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    """Agent の状態(LangGraph 推奨パターン)"""
    question: str                                 # 元の質問
    messages: Annotated[list, add_messages]       # 会話履歴
    iteration: int                                # 現在のイテレーション
    final_answer: str | None                      # 最終回答

グラフの構築

ReActパターンのグラフを構築するコードです。

class AgenticRAG:
    """LangGraph ベースの Agentic RAG"""

    def __init__(self, max_iterations: int = 3):
        self.max_iterations = max_iterations

        self.llm = AzureChatOpenAI(
            azure_deployment=settings.AZURE_CHAT_DEPLOYMENT,
            openai_api_key=settings.AZURE_OPENAI_API_KEY,
            azure_endpoint=settings.AZURE_OPENAI_ENDPOINT,
            api_version=settings.AZURE_OPENAI_API_VERSION
        )

        # ツールをバインド
        self.llm_with_tools = self.llm.bind_tools(TOOLS)
        self.graph = self._build_graph()

    def _build_graph(self) -> StateGraph:
        """ReActパターンのグラフを構築"""
        workflow = StateGraph(AgentState)

        # ノードを追加
        workflow.add_node("agent", self._agent_node)
        workflow.add_node("tools", ToolNode(TOOLS))
        workflow.add_node("increment_iteration", self._increment_iteration)
        workflow.add_node("generate", self._generate_node)

        # エントリーポイント
        workflow.set_entry_point("agent")

        # 条件分岐
        workflow.add_conditional_edges(
            "agent",
            self._should_continue,
            {
                "continue": "tools",
                "generate": "generate",
            }
        )

        # ツール実行後 → イテレーション増加 → Agent に戻る
        workflow.add_edge("tools", "increment_iteration")
        workflow.add_edge("increment_iteration", "agent")

        # 生成後 → 終了
        workflow.add_edge("generate", END)

        return workflow.compile()

利用可能なツール

Agentic RAGには5つのツールを持たせています。

@tool
def semantic_search_summary(query: str, k: int = 4) -> list[dict]:
    """要約からプロジェクト検索(探索向け)"""
    ...

@tool
def semantic_search_content(query: str, k: int = 5) -> list[dict]:
    """本文から詳細検索(実装向け)"""
    ...

@tool
def text2sql_query(query: str) -> dict:
    """SQLで統計・ランキング取得"""
    ...

@tool
def get_project_detail(project_name: str) -> dict:
    """特定プロジェクトの詳細"""
    ...

@tool
def keyword_search(keyword: str) -> list[dict]:
    """キーワードでLIKE検索"""
    ...

TOOLS = [
    semantic_search_summary,
    semantic_search_content,
    text2sql_query,
    get_project_detail,
    keyword_search
]

💡 開発で学んだこと

MCPサーバーとAgentic RAGを開発する中で、いくつかの学びがありました。

ツールの説明文が超重要

MCPサーバーでツールを定義する際、docstring(説明文)の書き方でLLMの挙動が大きく変わります。

# ❌ 悪い例:説明が曖昧
@mcp.tool
def search(q: str) -> list:
    """検索します"""
    ...

# ✅ 良い例:具体的な使用場面を記述
@mcp.tool
def semantic_search_summary(query: str, k: int = 6) -> list[dict]:
    """
    【推奨】プロジェクト単位の主題検索

    ユーザーが「〇〇関連のプロジェクト」「△△を使った作品」など
    主題・テーマで類似プロジェクトを探したい場合に使用。

    各プロジェクトの要約(ContentSummary)から検索するため、
    プロジェクト単位で1件ずつ返し、重複なく関連作品を見つけやすい。
    """
    ...

ポイント:

  • 【推奨】 のようなマーカーでツールの優先度を示す
  • 「どんな時に使うか」を具体的に書く
  • 他のツールとの違いを明確にする

返り値のサイズに注意

Agentic RAGでツールを連続呼び出しすると、返り値がどんどんメッセージ履歴に蓄積されます。記事本文などを返すと、すぐにコンテキストウィンドウが埋まってしまいます。

# ❌ 問題:全文を返している
@tool
def get_project_content(name: str) -> str:
    """プロジェクトの記事本文を取得"""
    project = db.get_project(name)
    return project["content_raw"]  # 5000文字以上あることも...

# ✅ 改善:要約や抜粋のみ返す
@tool
def get_project_summary(name: str) -> dict:
    """プロジェクトの概要を取得"""
    project = db.get_project(name)
    return {
        "name": project["project_name"],
        "summary": project["content_summary"],  # 300文字程度
        "url": project["url"]
    }

対策:

  • 要約やメタデータのみを返す
  • 必要に応じて get_project や semantic_search_content などのツールで詳細を検索
  • 返り値を構造化し、必要最小限の情報に絞る

💡 今後の改善案として、Anthropic の Agent Skills も検討できます。
Agent Skills は指示をファイルベースで管理し、必要な時だけ読み込む仕組みです。
これにより、コンテキストウィンドウを節約しつつ、ツールの使い方を正確に伝えられます。

2段階ベクトル検索の効果

最初は content_raw のチャンクだけでベクトル検索していましたが、同じプロジェクトが複数回ヒットしてしまう問題がありました。

解決策として、2種類のベクトルストアを用意:

ストア 用途 粒度
summary_store プロジェクト探索 1プロジェクト = 1ドキュメント
content_store 実装詳細検索 1プロジェクト = 複数チャンク

これにより、「どんなプロジェクトがあるか」と「具体的な実装方法」を使い分けて検索できるようになりました。

まとめ

今回のシステムで実現したこと:

  1. FastMCP 2.0 でシンプルにMCPサーバーを構築
  2. LangGraph でReActパターンのAgentic RAGを実装
  3. ChromaDB でベクトル検索を実現
  4. Text2SQL で自然言語からSQLを生成

MCPサーバーを作ると、Claude Desktopなど複数のAIツールから同じデータにアクセスできるようになるので、便利。

参考リンク

Discussion