72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
"""LLM 服务兼容层。
|
|
|
|
在完整 AI 能力未接入时提供最小实现，保证 API 端点可运行。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
|
|
class LLMService:
|
|
async def generate(self, prompt: str, **_: Any) -> str:
|
|
return f"[stub] {prompt[:200]}"
|
|
|
|
async def generate_task_breakdown_async(self, description: str, **_: Any) -> list[dict]:
|
|
if not description.strip():
|
|
return []
|
|
return [
|
|
{
|
|
"task_name": "需求分析",
|
|
"type": "AI",
|
|
"priority": "medium",
|
|
"estimated_hours": 4,
|
|
"description": description[:200],
|
|
}
|
|
]
|
|
|
|
async def get_embedding_async(self, text: str, **_: Any) -> list[float]:
|
|
seed = float(len(text) % 10) / 10.0
|
|
return [seed for _ in range(16)]
|
|
|
|
async def extract_context_insights_async(self, text: str, **_: Any) -> dict:
|
|
return {"summary": text[:120], "keywords": []}
|
|
|
|
async def find_relevant_context(self, **_: Any) -> list[dict]:
|
|
return []
|
|
|
|
async def evaluate_with_voting_async(self, signal_content: str, impact_score: float, **_: Any) -> dict:
|
|
is_strategic = impact_score >= 0.8
|
|
return {
|
|
"is_strategic": is_strategic,
|
|
"confidence": min(1.0, max(0.2, impact_score)),
|
|
"decision_proposal": {
|
|
"summary": signal_content[:120],
|
|
"action": "follow_up" if is_strategic else "observe",
|
|
},
|
|
"tasks": [],
|
|
"_voting": {
|
|
"enabled": False,
|
|
"models_used": [],
|
|
"strategic_votes": 0,
|
|
"total_votes": 0,
|
|
"unanimous": True,
|
|
},
|
|
}
|
|
|
|
async def audit_code_change_async(
|
|
self,
|
|
task_name: str,
|
|
commit_message: str,
|
|
code_diff: str,
|
|
**_: Any,
|
|
) -> dict:
|
|
passed = bool(task_name and commit_message and code_diff is not None)
|
|
return {
|
|
"passed": passed,
|
|
"reason": "basic_stub_check_passed" if passed else "invalid_input",
|
|
}
|
|
|
|
|
|
llm_service = LLMService()
|