File size: 6,602 Bytes
7b5783b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | """
AI Summarization Engine for Code Storage Service (Phase 4.6.12 Hardening)
Handles LLM-powered summary generation and action-oriented naming
for extracted code blocks.
"""
import asyncio
import json
import os
from typing import Any, cast
from src.server.config.logfire_config import search_logger
def _get_model_choice_logic() -> str:
"""Get MODEL_CHOICE with direct fallback from credentials or environment."""
try:
from src.server.services.credential_service import credential_service
if credential_service._cache_initialized and "MODEL_CHOICE" in credential_service._cache:
model = credential_service._cache["MODEL_CHOICE"]
else:
model = os.getenv("MODEL_CHOICE")
if not model:
raise ValueError("MODEL_CHOICE is not configured in environment or credentials")
return cast(str, model)
except Exception as e:
search_logger.error(f"Error getting model choice logic: {e}")
raise
def generate_code_example_summary_logic(
code: str,
context_before: str,
context_after: str,
language: str = "",
provider: str | None = None,
) -> dict[str, str]:
"""
Generate a summary and name for a code example using its surrounding context.
"""
model_choice = _get_model_choice_logic()
prompt = f"""<context_before>
{context_before[-500:] if len(context_before) > 500 else context_before}
</context_before>
<code_example language="{language}">
{code[:1500] if len(code) > 1500 else code}
</code_example>
<context_after>
{context_after[:500] if len(context_after) > 500 else context_after}
</context_after>
Based on the code example and its surrounding context, provide:
1. A concise, action-oriented name (1-4 words) that describes what this code DOES.
2. A summary (2-3 sentences) that describes what this code example demonstrates.
Format your response as JSON:
{{
"example_name": "Action-oriented name",
"summary": "Description"
}}
"""
try:
import openai
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
from src.server.services.credential_service import credential_service
if credential_service._cache_initialized and "OPENAI_API_KEY" in credential_service._cache:
cached_key = credential_service._cache["OPENAI_API_KEY"]
if isinstance(cached_key, dict) and cached_key.get("is_encrypted"):
from src.server.services.credentials.crypto_utils import CryptoUtils
api_key = CryptoUtils.decrypt_value(cached_key["encrypted_value"])
else:
api_key = cached_key
else:
api_key = os.getenv("OPENAI_API_KEY", "")
if not api_key:
raise ValueError("No OpenAI API key available for code summarization")
client = openai.OpenAI(api_key=api_key)
from src.server.services.prompt_service import prompt_service
default_instruction = "You are a helpful assistant that analyzes code examples."
system_prompt = prompt_service.get_prompt("CODE_EXAMPES_AUDITOR", default=default_instruction)
response = client.chat.completions.create(
model=model_choice,
messages=[
{
"role": "system",
"content": system_prompt,
},
{"role": "user", "content": prompt},
],
response_format={"type": "json_object"},
)
raw_content = response.choices[0].message.content
response_content = raw_content.strip() if raw_content is not None else ""
if not response_content:
raise ValueError("LLM returned empty summary")
result = json.loads(response_content)
return {
"example_name": result.get(
"example_name",
f"Code Example ({language})" if language else "Code Example",
),
"summary": result.get("summary", "Code example for demonstration purposes."),
}
except Exception as e:
search_logger.error(f"Summarization AI Error: {e}")
return {
"example_name": (f"Code Example ({language})" if language else "Code Example"),
"summary": "Code example for demonstration purposes.",
}
async def generate_code_summaries_batch_logic(
service_instance,
code_blocks: list[dict[str, Any]],
max_workers: int | None = None,
progress_callback: Any = None,
provider: str | None = None,
) -> list[dict[str, str]]:
"""
Generate summaries for a batch of code blocks concurrently.
"""
if not code_blocks:
return []
# Get max_workers from settings if not provided
if max_workers is None:
try:
from src.server.services.credential_service import credential_service
if credential_service._cache_initialized and "CODE_SUMMARY_MAX_WORKERS" in credential_service._cache:
max_workers = int(credential_service._cache["CODE_SUMMARY_MAX_WORKERS"])
else:
max_workers = int(os.getenv("CODE_SUMMARY_MAX_WORKERS", "3"))
except Exception:
max_workers = 3
search_logger.info(f"Generating summaries for {len(code_blocks)} code blocks with max_workers={max_workers}")
# Semaphore to limit concurrent requests
semaphore = asyncio.Semaphore(max_workers)
completed_count = 0
lock = asyncio.Lock()
async def _sum_single(block: dict[str, Any]) -> dict[str, str]:
nonlocal completed_count
async with semaphore:
# CPU/IO intensive LLM call to thread
result = await asyncio.to_thread(
generate_code_example_summary_logic,
code=block["code"],
context_before=block.get("context_before", ""),
context_after=block.get("context_after", ""),
language=block.get("language", ""),
provider=provider,
)
async with lock:
completed_count += 1
if progress_callback:
await progress_callback(
{
"status": "code_summarization",
"log": f"Generated {completed_count}/{len(code_blocks)} code summaries",
}
)
return result
tasks = [_sum_single(block) for block in code_blocks]
results = await asyncio.gather(*tasks)
return list(results)
|