Edge0連載 3: Microsoft Learn検索を組み込むローカルRAGエージェント

from dataclasses import dataclass


@dataclass
class Source:
    title: str
    url: str
    content: str

MAX_SOURCES = 3
MAX_SOURCE_CHARS = 1800

import re


def make_followup(action: str) -> str:
    if action == "알려주세요":
        return "그 여부도 알려주세요."

    if action == "확인해주세요":
        return "그 여부도 확인해주세요."

    return "그 여부도 설명해주세요."


def normalize_korean_question(text: str) -> str:
    suffix = (
        r"\s*(?:함께\s*)?"
        r"(설명해주세요|알려주세요|확인해주세요)"
        r"\.?$"
    )

    patterns = [
        (
            rf"^(.*)할 수 있는지도{suffix}",
            lambda m:
                f"{m.group(1)}할 수 있습니까? "
                f"{make_followup(m.group(2))}",
        ),
        (
            rf"^(.*)되는지도{suffix}",
            lambda m:
                f"{m.group(1)}됩니까? "
                f"{make_followup(m.group(2))}",
        ),
        (
            rf"^(.*)하는지도{suffix}",
            lambda m:
                f"{m.group(1)}합니까? "
                f"{make_followup(m.group(2))}",
        ),
        (
            rf"^(.*)필요한지도{suffix}",
            lambda m:
                f"{m.group(1)}필요합니까? "
                f"{make_followup(m.group(2))}",
        ),
    ]

    for pattern, replacement in patterns:
        match = re.match(pattern, text)

        if match:
            return replacement(match)

    return text

import subprocess


TRANSLATOR = "./apple_translate"


def run_process(args: list[str]) -> str:
    result = subprocess.run(
        args,
        check=True,
        capture_output=True,
        text=True,
    )

    return result.stdout.strip()


def translate(
    text: str,
    source: str,
    target: str,
) -> str:
    return run_process([
        TRANSLATOR,
        source,
        target,
        text,
    ])

def split_korean_input(text: str) -> list[str]:
    return [
        part.strip()
        for part in re.split(
            r"(?<=[?!])\s+|(?<=\.)\s+",
            text,
        )
        if part.strip()
    ]

def translate_question_to_english(
    question: str,
) -> str:
    translated = []

    for segment in split_korean_input(question):
        normalized = normalize_korean_question(segment)

        english = translate(
            normalized,
            "ko",
            "en",
        )

        translated.append(english)

    return " ".join(translated)

import json


def search_microsoft_learn(
    query_text: str,
) -> list[Source]:
    raw = run_process([
        "mslearn",
        "search",
        query_text,
        "--json",
    ])

    data = json.loads(raw)

    results = (
        data.get("results")
        or data.get("value")
        or []
    )

    sources = []
    seen_urls = set()

    for item in results:
        url = (
            item.get("contentUrl")
            or item.get("url")
            or item.get("href")
            or ""
        )

        if not url or url in seen_urls:
            continue

        content = (
            item.get("content")
            or item.get("snippet")
            or item.get("description")
            or ""
        ).strip()

        if not content:
            continue

        seen_urls.add(url)

        sources.append(
            Source(
                title=item.get(
                    "title",
                    "(untitled)",
                ),
                url=url,
                content=content[:MAX_SOURCE_CHARS],
            )
        )

        if len(sources) >= MAX_SOURCES:
            break

    return sources

def build_grounding(
    sources: list[Source],
) -> str:
    blocks = []

    for index, source in enumerate(
        sources,
        start=1,
    ):
        blocks.append(
            f"[{index}] {source.title}\n"
            f"URL: {source.url}\n\n"
            f"{source.content}"
        )

    return "\n\n---\n\n".join(blocks)

def build_user_prompt(
    question: str,
    sources: list[Source],
) -> str:
    grounding = build_grounding(sources)

    return f"""
Microsoft Learn references:

<references>
{grounding}
</references>

Question:
{question}

Answer the question using the references above.
""".strip()

SYSTEM_PROMPT = """
You are a Microsoft technology expert.

Answer using the supplied Microsoft Learn references
as the primary source of factual information.

Rules:
- Prefer facts explicitly supported by the references.
- If the references do not establish a fact, say so.
- Do not invent API names, product behavior, limits,
  or version details.
- Ignore unrelated search results.
- Use short plain-text paragraphs.
- Cite supporting references using [1], [2], [3].
""".strip()

MAX_HISTORY_TURNS = 2

def make_search_query(
    current_question: str,
    history: list[dict],
) -> str:
    previous_user = None

    for message in reversed(history):
        if message["role"] == "user":
            previous_user = message["content"]
            break

    if previous_user is None:
        return current_question

    return (
        f"Previous topic: {previous_user}\n"
        f"Current question: {current_question}"
    )