vector-search

建設分野のベクトル検索

概要

DDC の方法論（第 4.4 章）に基づき、このスキルでは建設データのセマンティックベクトル検索を実装します。キーワードの一致にとどまらず、単語だけでなく意味に基づいてドキュメントやデータを検索できます。

書籍の参考文献：「Современные технологии работы с данными」／「Modern Data Technologies」

「Векторные базы данных позволяют находить семантически похожие документы, даже если они используют разную терминологию.」 — DDC Book、第 4.4 章

クイックスタート

from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance, PointStruct

# Initialize embedding model
model = SentenceTransformer('all-MiniLM-L6-v2')

# Create Qdrant client (in-memory for demo)
client = QdrantClient(":memory:")

# Create collection
client.create_collection(
    collection_name="construction_docs",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE)
)

# Sample construction documents
documents = [
    "Concrete mix design for C30 grade with water-cement ratio 0.45",
    "Steel reinforcement specifications for structural columns",
    "Waterproofing membrane installation for basement walls",
    "Fire-rated door specifications for escape routes"
]

# Index documents
for idx, doc in enumerate(documents):
    embedding = model.encode(doc).tolist()
    client.upsert(
        collection_name="construction_docs",
        points=[PointStruct(id=idx, vector=embedding, payload={"text": doc})]
    )

# Search
query = "basement moisture protection"
query_vector = model.encode(query).tolist()
results = client.search(
    collection_name="construction_docs",
    query_vector=query_vector,
    limit=3
)

for result in results:
    print(f"Score: {result.score:.3f} - {result.payload['text']}")

ベクトルデータベースのセットアップ

Qdrant のセットアップ

from qdrant_client import QdrantClient
from qdrant_client.models import (
    VectorParams, Distance, PointStruct,
    Filter, FieldCondition, MatchValue
)
import uuid

class ConstructionVectorDB:
    """Vector database for construction documents and data"""

    def __init__(self, host="localhost", port=6333, in_memory=False):
        if in_memory:
            self.client = QdrantClient(":memory:")
        else:
            self.client = QdrantClient(host=host, port=port)

        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        self.collections = {}

    def create_collection(self, name, description=None):
        """Create a new collection"""
        self.client.create_collection(
            collection_name=name,
            vectors_config=VectorParams(
                size=384,  # Dimension for all-MiniLM-L6-v2
                distance=Distance.COSINE
            )
        )
        self.collections[name] = description

    def index_documents(self, collection_name, documents, metadata=None):
        """Index documents with embeddings"""
        points = []

        for idx, doc in enumerate(documents):
            embedding = self.model.encode(doc).tolist()

            payload = {"text": doc}
            if metadata and idx < len(metadata):
                payload.update(metadata[idx])

            points.append(PointStruct(
                id=str(uuid.uuid4()),
                vector=embedding,
                payload=payload
            ))

        self.client.upsert(
            collection_name=collection_name,
            points=points
        )

        return len(points)

    def search(self, collection_name, query, limit=5, filters=None):
        """Semantic search"""
        query_vector = self.model.encode(query).tolist()

        search_filter = None
        if filters:
            conditions = [
                FieldCondition(key=k, match=MatchValue(value=v))
                for k, v in filters.items()
            ]
            search_filter = Filter(must=conditions)

        results = self.client.search(
            collection_name=collection_name,
            query_vector=query_vector,
            limit=limit,
            query_filter=search_filter
        )

        return [
            {
                'score': r.score,
                'text': r.payload.get('text'),
                'metadata': {k: v for k, v in r.payload.items() if k != 'text'}
            }
            for r in results
        ]

    def hybrid_search(self, collection_name, query, keyword_filter=None, limit=5):
        """Combine semantic search with keyword filtering"""
        # First semantic search
        semantic_results = self.search(collection_name, query, limit=limit*2)

        # Then keyword filter if provided
        if keyword_filter:
            filtered = [
                r for r in semantic_results
                if keyword_filter.lower() in r['text'].lower()
            ]
            return filtered[:limit]

        return semantic_results[:limit]

ChromaDB の代替案

import chromadb
from chromadb.utils import embedding_functions

class ChromaConstructionDB:
    """ChromaDB-based vector search for construction"""

    def __init__(self, persist_directory=None):
        if persist_directory:
            self.client = chromadb.PersistentClient(path=persist_directory)
        else:
            self.client = chromadb.Client()

        self.embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction(
            model_name="all-MiniLM-L6-v2"
        )

    def create_collection(self, name):
        """Create or get collection"""
        return self.client.get_or_create_collection(
            name=name,
            embedding_function=self.embedding_fn
        )

    def index_specifications(self, collection_name, specs):
        """Index construction specifications"""
        collection = self.create_collection(collection_name)

        ids = [f"spec_{i}" for i in range(len(specs))]
        documents = [s['text'] for s in specs]
        metadatas = [{k: v for k, v in s.items() if k != 'text'} for s in specs]

        collection.add(
            ids=ids,
            documents=documents,
            metadatas=metadatas
        )

        return len(specs)

    def search(self, collection_name, query, n_results=5, where=None):
        """Search specifications"""
        collection = self.create_collection(collection_name)

        results = collection.query(
            query_texts=[query],
            n_results=n_results,
            where=where
        )

        return [
            {
                'id': results['ids'][0][i],
                'text': results['documents'][0][i],
                'metadata': results['metadatas'][0][i] if results['metadatas'] else {},
                'distance': results['distances'][0][i] if results['distances'] else None
            }
            for i in range(len(results['ids'][0]))
        ]

建設分野に特化した用途

仕様書の検索

class SpecificationSearchEngine:
    """Search engine for construction specifications"""

    def __init__(self, db: ConstructionVectorDB):
        self.db = db
        self.collection = "specifications"

    def index_specifications(self, specs_df):
        """Index specifications from DataFrame"""
        self.db.create_collection(self.collection, "Construction specifications")

        documents = specs_df['description'].tolist()
        metadata = specs_df.drop('description', axis=1).to_dict('records')

        return self.db.index_documents(self.collection, documents, metadata)

    def find_similar_specs(self, query, category=None, limit=5):
        """Find similar specifications"""
        filters = {'category': category} if category else None
        return self.db.search(self.collection, query, limit=limit, filters=filters)

    def find_related_materials(self, material_name, limit=10):
        """Find specifications related to a material"""
        query = f"specifications for {material_name} materials"
        return self.db.search(self.collection, query, limit=limit)

    def search_by_requirement(self, requirement, limit=5):
        """Search by performance requirement"""
        query = f"specification meeting requirement: {requirement}"
        return self.db.search(self.collection, query, limit=limit)

規格・基準の検索

class StandardsSearchEngine:
    """Search engine for building standards and codes"""

    def __init__(self, db: ConstructionVectorDB):
        self.db = db
        self.collection = "standards"

    def index_standards(self, standards):
        """Index building standards

        Args:
            standards: List of dicts with 'code', 'title', 'section', 'text'
        """
        self.db.create_collection(self.collection, "Building standards and codes")

        documents = [s['text'] for s in standards]
        metadata = [{k: v for k, v in s.items() if k != 'text'} for s in standards]

        return self.db.index_documents(self.collection, documents, metadata)

    def find_applicable_standards(self, context, limit=5):
        """Find standards applicable to a given context"""
        return self.db.search(self.collection, context, limit=limit)

    def search_fire_codes(self, query):
        """Search fire safety codes"""
        full_query = f"fire safety code requirement: {query}"
        return self.db.search(
            self.collection,
            full_query,
            limit=10,
            filters={'category': 'fire_safety'}
        )

    def search_accessibility(self, query):
        """Search accessibility standards (ADA, etc.)"""
        full_query = f"accessibility requirement: {query}"
        return self.db.search(
            self.collection,
            full_query,
            limit=10,
            filters={'category': 'accessibility'}
        )

工事項目の検索（OpenConstructionEstimate）

class WorkItemSearchEngine:
    """Search engine for construction work items and unit prices"""

    def __init__(self, db: ConstructionVectorDB):
        self.db = db
        self.collection = "work_items"

    def index_work_items(self, items_df):
        """Index work items database

        Args:
            items_df: DataFrame with columns:
                - code: Work item code
                - description: Work description
                - unit: Unit of measure
                - unit_price: Price per unit
                - category: Work category
        """
        self.db.create_collection(self.collection, "Construction work items")

        documents = items_df['description'].tolist()
        metadata = items_df.drop('description', axis=1).to_dict('records')

        return self.db.index_documents(self.collection, documents, metadata)

    def find_similar_work(self, description, limit=10):
        """Find similar work items by description"""
        results = self.db.search(self.collection, description, limit=limit)

        return [
            {
                'description': r['text'],
                'code': r['metadata'].get('code'),
                'unit': r['metadata'].get('unit'),
                'unit_price': r['metadata'].get('unit_price'),
                'similarity': r['score']
            }
            for r in results
        ]

    def estimate_from_description(self, work_description, quantity):
        """Get cost estimate from work description"""
        matches = self.find_similar_work(work_description, limit=3)

        if not matches:
            return None

        best_match = matches[0]
        unit_price = best_match.get('unit_price', 0)

        return {
            'matched_item': best_match['description'],
            'code': best_match['code'],
            'unit': best_match['unit'],
            'unit_price': unit_price,
            'quantity': quantity,
            'total_cost': unit_price * quantity,
            'similarity_score': best_match['similarity']
        }

建設分野向け RAG

検索拡張生成

from openai import OpenAI

class ConstructionRAG:
    """RAG system for construction queries"""

    def __init__(self, vector_db: ConstructionVectorDB, openai_client=None):
        self.db = vector_db
        self.llm = openai_client or OpenAI()

    def answer_query(self, query, collection, n_context=5):
        """Answer query using RAG"""
        # Retrieve relevant context
        context_docs = self.db.search(collection, query, limit=n_context)

        # Build context string
        context = "\n\n".join([
            f"Document {i+1}:\n{doc['text']}"
            for i, doc in enumerate(context_docs)
        ])

        # Generate answer
        prompt = f"""Based on the following construction documents, answer the query.

Context:
{context}

Query: {query}

Provide a detailed, accurate answer based only on the provided context.
If the context doesn't contain enough information, say so."""

        response = self.llm.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "You are a construction industry expert."},
                {"role": "user", "content": prompt}
            ]
        )

        return {
            'answer': response.choices[0].message.content,
            'sources': context_docs,
            'query': query
        }

    def summarize_specifications(self, topic, collection="specifications"):
        """Summarize specifications on a topic"""
        docs = self.db.search(collection, topic, limit=10)

        context = "\n".join([doc['text'] for doc in docs])

        prompt = f"""Summarize the following construction specifications related to: {topic}

Specifications:
{context}

Provide a structured summary with key requirements and recommendations."""

        response = self.llm.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )

        return {
            'summary': response.choices[0].message.content,
            'source_count': len(docs)
        }

ドキュメントのインデックス作成パイプライン

import os
import pdfplumber
from typing import List, Dict

class DocumentIndexingPipeline:
    """Pipeline for indexing construction documents"""

    def __init__(self, vector_db: ConstructionVectorDB):
        self.db = vector_db
        self.chunk_size = 500
        self.chunk_overlap = 50

    def chunk_text(self, text: str) -> List[str]:
        """Split text into chunks"""
        words = text.split()
        chunks = []

        for i in range(0, len(words), self.chunk_size - self.chunk_overlap):
            chunk = ' '.join(words[i:i + self.chunk_size])
            if len(chunk) > 50:  # Skip very small chunks
                chunks.append(chunk)

        return chunks

    def extract_pdf_text(self, pdf_path: str) -> str:
        """Extract text from PDF"""
        text = ""
        with pdfplumber.open(pdf_path) as pdf:
            for page in pdf.pages:
                page_text = page.extract_text()
                if page_text:
                    text += page_text + "\n"
        return text

    def index_document(self, file_path: str, collection: str, metadata: Dict = None):
        """Index a single document"""
        # Extract text
        if file_path.endswith('.pdf'):
            text = self.extract_pdf_text(file_path)
        else:
            with open(file_path, 'r', encoding='utf-8') as f:
                text = f.read()

        # Chunk text
        chunks = self.chunk_text(text)

        # Build metadata for each chunk
        base_metadata = metadata or {}
        base_metadata['source_file'] = os.path.basename(file_path)

        chunk_metadata = [
            {**base_metadata, 'chunk_index': i}
            for i in range(len(chunks))
        ]

        # Index
        return self.db.index_documents(collection, chunks, chunk_metadata)

    def index_directory(self, directory: str, collection: str, extensions=None):
        """Index all documents in a directory"""
        if extensions is None:
            extensions = ['.pdf', '.txt', '.md']

        total_indexed = 0

        for root, _, files in os.walk(directory):
            for file in files:
                if any(file.endswith(ext) for ext in extensions):
                    file_path = os.path.join(root, file)
                    try:
                        count = self.index_document(file_path, collection)
                        total_indexed += count
                        print(f"Indexed: {file} ({count} chunks)")
                    except Exception as e:
                        print(f"Error indexing {file}: {e}")

        return total_indexed

クイックリファレンス

コンポーネント	説明	用途
Qdrant	高性能なベクトル DB	本番環境へのデプロイ
ChromaDB	シンプルな組み込み型ベクトル DB	開発／テスト
SentenceTransformers	埋め込みモデル	テキストからベクトルへの変換
RAG	検索＋生成	ドキュメントに関する質疑応答

建設分野向け埋め込みモデル

# Recommended models by use case
EMBEDDING_MODELS = {
    'general': 'all-MiniLM-L6-v2',           # Fast, 384 dim
    'multilingual': 'paraphrase-multilingual-MiniLM-L12-v2',  # Multi-language
    'quality': 'all-mpnet-base-v2',          # Better quality, 768 dim
    'construction': 'allenai/scibert_scivocab_uncased'  # Technical texts
}

リソース

• 書籍：Artem Boiko 著「Data-Driven Construction」、第 4.4 章
• ウェブサイト：https://datadrivenconstruction.io
• Qdrant：https://qdrant.tech
• ChromaDB：https://www.trychroma.com
• SentenceTransformers：https://www.sbert.net

次のステップ

• LLM の統合については llm-data-automation を参照してください
• ドキュメントの分類については document-classification-nlp を参照してください
• RAG の用途については rag-construction を参照してください