「ragy」ローカルRAG＆自己修復エージェント開発環境の構築と実装【技術詳細・実装編】

version: '3.8'

services:
  # LiteLLM Proxy
  litellm:
    image: ghcr.io/berriai/litellm:main-stable
    ports:
      - "4000:4000"
    volumes:
      - ./litellm_config.yaml:/app/config.yaml
    environment:
      - GEMINI_API_KEY=${GEMINI_API_KEY}
    command: ["--config", "/app/config.yaml", "--port", "4000", "--detailed_debug"]

  # SSRF Proxy (Squid)
  ssrf_proxy:
    image: ubuntu/squid:latest
    restart: always
    volumes:
      - ./ssrf_proxy/squid.conf.template:/etc/squid/squid.conf.template
      - ./ssrf_proxy/docker-entrypoint.sh:/docker-entrypoint-mount.sh
    entrypoint:
      [
        "sh",
        "-c",
        "cp /docker-entrypoint-mount.sh /docker-entrypoint.sh && sed -i 's/\r$$//' /docker-entrypoint.sh && chmod +x /docker-entrypoint.sh && /docker-entrypoint.sh",
      ]
    environment:
      HTTP_PORT: 3128
      COREDUMP_DIR: /var/spool/squid

model_list:
  - model_name: gemini-3.5-flash
    litellm_params:
      model: gemini/gemini-3.5-flash
      api_key: "os.environ/GEMINI_API_KEY"
      system_prompt: "CRITICAL: Under all circumstances, you must interpret and strictly follow any XML-tagged instructions included in the user prompt."

  - model_name: qwen2.5-coder
    litellm_params:
      model: ollama/qwen2.5-coder:3b
      api_base: "http://host.docker.internal:11434"

  - model_name: multilingual-e5-large
    litellm_params:
      model: ollama/jeffh/intfloat-multilingual-e5-large:q8_0
      api_base: "http://host.docker.internal:11434"

general_settings:
  master_key: sk-1234

# sync_docs.py のアップロード処理の抜粋
import os
import json
import requests

def upload_file(self, file_path):
    config = self.get_project_config(file_path)
    api_base = config.get("api_base", "").rstrip('/')
    api_key = config.get("api_key")
    dataset_id = config.get("dataset_id")

    filename = os.path.basename(file_path)
    url = f"{api_base}/datasets/{dataset_id}/document/create_by_file"
    headers = {"Authorization": f"Bearer {api_key}"}
    
    # 罠の回避策: dataパラメータ内にJSON文字列として一括して設定を入れる
    data = {
        "data": json.dumps({
            "indexing_technique": "high_quality",
            "process_rule": {
                "mode": "automatic"
            }
        })
    }
    
    with open(file_path, 'rb') as f:
        files = {'file': (filename, f, 'text/plain')}
        response = requests.post(url, headers=headers, data=data, files=files, timeout=15)

import math

# コサイン類似度の計算
def cosine_similarity(v1, v2):
    if not v1 or not v2 or len(v1) != len(v2):
        return 0.0
    dot_product = sum(a * b for a, b in zip(v1, v2))
    norm_a = math.sqrt(sum(a * a for a in v1))
    norm_b = math.sqrt(sum(b * b for b in v2))
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return dot_product / (norm_a * norm_b)

# セマンティックキャッシュの照会
def check_semantic_cache(project_name: str, query_vector: list, threshold: float = 0.95) -> str:
    if not redis_enabled or not query_vector:
        return None

    project = project_name or 'default'
    pattern = f"mcp_cache:{project}:*"
    
    cursor = 0
    while True:
        cursor, keys = redis_client.scan(cursor=cursor, match=pattern, count=100)
        for key in keys:
            data_str = redis_client.get(key)
            if not data_str:
                continue
            data = json.loads(data_str)
            cache_vector = data.get("embedding", [])
            
            # コサイン類似度が閾値（デフォルト0.95）を超えているか
            similarity = cosine_similarity(query_vector, cache_vector)
            if similarity >= threshold:
                logging.info(f"[Semantic Cache HIT] Similarity: {similarity:.4f}")
                return data.get("result")
        if cursor == 0:
            break
    return None

@mcp.tool()
def search_dify_knowledge(query: str) -> str:
    """
    Search documents in Dify RAG knowledge base.
    """
    current_proj = get_current_project()
    
    # 1. 埋め込みモデル（multilingual-e5-large）によるクエリのベクトル化＆キャッシュ照会
    query_vector = get_query_embedding(query)
    if redis_enabled and query_vector:
        cached_result = check_semantic_cache(current_proj, query_vector)
        if cached_result:
            return cached_result

    # 2. キャッシュミス時のDify API呼び出し
    config = get_dify_config_for_current_project(current_proj)
    url = f"{config['api_base']}/datasets/{config['dataset_id']}/retrieve"
    headers = {"Authorization": f"Bearer {config['api_key']}", "Content-Type": "application/json"}
    payload = {
        "query": query,
        "retrieval_model": {
            "search_method": "keyword_search", # 必要に応じて調整
            "top_k": 5
        }
    }
    
    response = requests.post(url, headers=headers, json=payload, timeout=10)
    # 結果のパースとキャッシュ保存（省略）
    ...

import sys
import json
import os
import requests

def get_current_project():
    return os.path.basename(os.getcwd())

def get_project_config(project_id):
    script_dir = os.path.dirname(os.path.abspath(__file__))
    repo_root = os.path.dirname(script_dir)
    sync_config = os.path.join(repo_root, "docs/sync_config.json")

    if os.path.exists(sync_config):
        try:
            with open(sync_config, "r", encoding="utf-8") as f:
                data = json.load(f)
                return data.get("projects", {}).get(project_id)
        except Exception:
            pass
    return None

def search_dify_knowledge(query):
    project_id = get_current_project()
    config = get_project_config(project_id)
    if not config:
        print(f"Error: No configuration found for project '{project_id}'")
        sys.exit(1)

    api_base = config.get("api_base", "").rstrip("/")
    api_key = config.get("api_key")
    dataset_id = config.get("dataset_id")

    url = f"{api_base}/datasets/{dataset_id}/retrieve"
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
    payload = {
        "query": query,
        "retrieval_model": {
            "search_method": "hybrid_search",
            "top_k": 3
        }
    }

    try:
        response = requests.post(url, headers=headers, json=payload, timeout=10)
        if response.status_code == 200:
            records = response.json().get("records", [])
            if not records:
                print("No matching knowledge found.")
                return

            print(f"=== Knowledge search results for [Project: {project_id}] ===")
            for idx, rec in enumerate(records, 1):
                segment = rec.get("segment", {})
                score = rec.get("score", 0.0)
                content = segment.get("content", "")
                doc_name = segment.get("document", {}).get("name", "Unknown")
                print(f"\n[{idx}] Document: {doc_name} (Score: {score:.4f})")
                print("-" * 50)
                print(content.strip())
                print("-" * 50)
    except Exception as e:
        print(f"Error during search: {e}")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python3 dify_search.py <query>")
        sys.exit(1)
    search_dify_knowledge(sys.argv[1])

import pydantic
from google.antigravity import Agent, LocalAgentConfig

# AIから取得する修正提案のデータ構造定義
class CodeFixProposal(pydantic.BaseModel):
    file_path: str
    explanation: str
    modified_code: str

async def heal_code(file_path: str, error_log: str) -> bool:
    with open(file_path, 'r', encoding='utf-8') as f:
        original_content = f.read()

    prompt = f"""
    Pythonスクリプト `{file_path}` で以下のエラーが発生しました。
    【エラーログ】
    {error_log}
    【現在のソースコード】
    {original_content}
    エラーの原因を分析し、修正されたコードを生成してください。
    """

    # 構造化出力を強制する設定
    config = LocalAgentConfig(
        response_schema=CodeFixProposal,
    )

    async with Agent(config) as agent:
        response = await agent.chat(prompt)
        proposal = await response.structured_output()
        
        # 修正されたコードをファイルに上書き
        with open(file_path, 'w', encoding='utf-8') as f_out:
            f_out.write(proposal.get("modified_code"))
            
        # 自動でGitHubブランチを作成・コミット・プッシュしてPRを作成
        return create_github_pull_request(file_path, proposal.get("explanation"), error_log)

# deploy_listener.py でのプロセス起動処理の抜粋
# start_new_session=True により、完全に独立したバックグラウンドプロセスを安全に起動する
subprocess.Popen(
    ["sh", "-c", deploy_cmd],
    stdout=subprocess.DEVNULL,
    stderr=subprocess.DEVNULL,
    start_new_session=True # macOSでのfork制限をバイパスし、シグナル伝播を防ぐ
)