name: 🤖 AI Issue Smart Reply
on:
issues:
types: [opened]
issue_comment:
types: [created]
workflow_dispatch:
inputs:
target_type:
description: '目标类型'
required: true
type: choice
options:
- issue
- pr
issue_number:
description: 'Issue 或 PR 编号（如 42）'
required: true
type: string
permissions:
contents: write
issues: write
pull-requests: read
concurrency:
group: ai-issue-${{ github.event.issue.number || github.event.inputs.issue_number }}
cancel-in-progress: false
jobs:
ai-smart-reply:
if: |
vars.AI_ISSUE_REPLY_ENABLED == 'true' && (
github.event_name == 'issues' ||
github.event_name == 'workflow_dispatch' ||
(github.event_name == 'issue_comment' && startsWith(github.event.comment.body, '/ai-analyze'))
)
runs-on: ubuntu-latest
timeout-minutes: 20
env:
AI_AUTO_LABEL: ${{ vars.AI_AUTO_LABEL || 'true' }}
AI_ENABLE_DUPLICATE_CHECK: ${{ vars.AI_ENABLE_DUPLICATE_CHECK || 'true' }}
AI_ENABLE_PR_SEARCH: ${{ vars.AI_ENABLE_PR_SEARCH || 'true' }}
AI_ENABLE_COMMIT_SEARCH: ${{ vars.AI_ENABLE_COMMIT_SEARCH || 'true' }}
AI_ONLY_TEMPLATE_TYPES: ${{ vars.AI_ONLY_TEMPLATE_TYPES || 'false' }}
AI_MAX_CONTEXT_CHARS: ${{ vars.AI_MAX_CONTEXT_CHARS || '50000' }}
AI_MAX_DUP_CANDIDATES: ${{ vars.AI_MAX_DUP_CANDIDATES || '80' }}
AI_MAX_PR_CANDIDATES: ${{ vars.AI_MAX_PR_CANDIDATES || '20' }}
AI_MAX_COMMIT_CANDIDATES: ${{ vars.AI_MAX_COMMIT_CANDIDATES || '20' }}
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: true
steps:
- name: Checkout repository
uses: actions/checkout@v4
with:
fetch-depth: 100
- name: Install dependencies
shell: bash
run: |
set -euo pipefail
sudo apt-get update
sudo apt-get install -y jq python3 ripgrep
- name: Prepare issue payload
id: issue
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
INPUT_NUM: ${{ github.event.issue.number || github.event.inputs.issue_number }}
TARGET_TYPE: ${{ github.event.inputs.target_type || 'issue' }}
AUTO_TITLE: ${{ github.event.issue.title }}
AUTO_BODY: ${{ github.event.issue.body || '' }}
AUTO_AUTHOR: ${{ github.event.issue.user.login }}
AUTO_URL: ${{ github.event.issue.html_url }}
EVENT_NAME: ${{ github.event_name }}
COMMENT_BODY: ${{ github.event.comment.body || '' }}
shell: bash
run: |
set -euo pipefail
mkdir -p .ai_runtime
python3 - <<'PY'
import json, os, pathlib, subprocess
input_num = os.environ['INPUT_NUM']
target_type = os.environ['TARGET_TYPE']
auto_title = os.environ.get('AUTO_TITLE', '')
event_name = os.environ.get('EVENT_NAME', '')
comment_body = os.environ.get('COMMENT_BODY', '')
re_analyze = False
if event_name == 'issue_comment':
# 评论触发：通过 gh CLI 获取 issue 数据
cmd = ['gh', 'issue', 'view', input_num,
'--json', 'number,title,body,author,url']
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
d = json.loads(result.stdout)
author = d.get('author', {})
payload = {
"number": d['number'],
"title": d.get('title', ''),
"body": d.get('body', '') or '',
"author": author.get('login', '') if isinstance(author, dict) else str(author),
"url": d.get('url', ''),
}
re_analyze = True
elif auto_title:
# 自动触发：使用事件环境变量
payload = {
"number": int(input_num),
"title": auto_title,
"body": os.environ.get('AUTO_BODY', ''),
"author": os.environ.get('AUTO_AUTHOR', ''),
"url": os.environ.get('AUTO_URL', ''),
}
else:
# 手动触发：通过 gh CLI 获取数据
cmd = ['gh', 'pr' if target_type == 'pr' else 'issue', 'view', input_num,
'--json', 'number,title,body,author,url']
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
d = json.loads(result.stdout)
author = d.get('author', {})
payload = {
"number": d['number'],
"title": d.get('title', ''),
"body": d.get('body', '') or '',
"author": author.get('login', '') if isinstance(author, dict) else str(author),
"url": d.get('url', ''),
}
payload["re_analyze"] = re_analyze
pathlib.Path('.ai_runtime/issue.json').write_text(
json.dumps(payload, ensure_ascii=False, indent=2), encoding='utf-8'
)
print(f"成功获取 {target_type} #{input_num}: {payload['title']}")
PY
echo "payload=$(cat .ai_runtime/issue.json | jq -c .)" >> "$GITHUB_OUTPUT"
echo "issue_number=$(jq -r '.number' .ai_runtime/issue.json)" >> "$GITHUB_OUTPUT"
echo "re_analyze=$(jq -r '.re_analyze' .ai_runtime/issue.json)" >> "$GITHUB_OUTPUT"
- name: React to /ai-analyze comment
if: github.event_name == 'issue_comment' && steps.issue.outputs.re_analyze == 'true'
uses: actions/github-script@v7
env:
ISSUE_NUMBER: ${{ steps.issue.outputs.issue_number }}
COMMENT_ID: ${{ github.event.comment.id }}
with:
script: |
const issue_number = parseInt(process.env.ISSUE_NUMBER, 10) || context.issue.number;
const comment_id = parseInt(process.env.COMMENT_ID, 10);
if (comment_id) {
await github.rest.reactions.createForIssueComment({
owner: context.repo.owner,
repo: context.repo.repo,
comment_id,
content: 'rocket'
});
}
- name: Stage 1 - classify and rewrite query
id: rewrite
shell: bash
env:
LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
LLM_BASE_URL: ${{ secrets.LLM_BASE_URL }}
LLM_MODEL: ${{ secrets.LLM_MODEL }}
run: |
set -euo pipefail
cat > .ai_runtime/rewrite_schema.json <<'JSON'
{
"name": "issue_rewrite",
"parameters": {
"type": "object",
"properties": {
"summary": { "type": "string" },
"issue_type": {
"type": "string",
"enum": ["bug", "enhancement", "question", "documentation", "needs_more_info"]
},
"search_queries": {
"type": "array",
"items": { "type": "string" }
},
"keywords": {
"type": "array",
"items": { "type": "string" }
},
"components": {
"type": "array",
"items": { "type": "string" }
},
"likely_paths": {
"type": "array",
"items": { "type": "string" }
}
},
"required": ["summary", "issue_type", "search_queries", "keywords", "components", "likely_paths"]
}
}
JSON
cat > .ai_runtime/rewrite_system_prompt.txt <<'EOF'
你负责把 Issue 重写成适合仓库检索的查询。
输出必须是严格 JSON。
search_queries 控制在 3-6 条，keywords 控制在 5-12 条。
EOF
python3 -c "
import json, pathlib
issue = json.loads(pathlib.Path('.ai_runtime/issue.json').read_text(encoding='utf-8'))
pathlib.Path('.ai_runtime/rewrite_input_prompt.txt').write_text(
'请分析这个 Issue，并输出适合仓库知识库/代码库检索的查询。\n\n标题：\n' + issue.get('title','') + '\n\n正文：\n' + issue.get('body',''),
encoding='utf-8'
)
"
python3 - <<'PY'
import json, os, pathlib, urllib.request, urllib.error
base_url = os.environ['LLM_BASE_URL'].rstrip('/')
api_key = os.environ['LLM_API_KEY']
model = os.environ['LLM_MODEL']
system_prompt = pathlib.Path('.ai_runtime/rewrite_system_prompt.txt').read_text(encoding='utf-8')
input_prompt = pathlib.Path('.ai_runtime/rewrite_input_prompt.txt').read_text(encoding='utf-8')
schema = json.loads(pathlib.Path('.ai_runtime/rewrite_schema.json').read_text(encoding='utf-8'))
candidates = [
(f"{base_url}/chat/completions", {
"model": model,
"temperature": 0.2,
"response_format": {"type": "json_object"},
"messages": [
{"role": "system", "content": system_prompt + "\n你必须直接输出 JSON 对象，不要输出 markdown。"},
{"role": "user", "content": input_prompt},
],
}, 'chat_json'),
(f"{base_url}/chat/completions", {
"model": model,
"temperature": 0.2,
"messages": [
{"role": "system", "content": system_prompt + "\n你必须直接输出 JSON 对象，不要输出 markdown 或代码块。"},
{"role": "user", "content": input_prompt + "\n\n请直接输出 JSON，不要包含 ```json 代码块标记。"},
],
}, 'chat_plain'),
]
headers = {
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'github-actions-ai-triage/1.0',
}
last_err = None
content = ''
mode_used = ''
required_fields = {"summary", "issue_type", "search_queries", "keywords", "components", "likely_paths"}
def validate_rewrite(s):
try:
obj = json.loads(s) if isinstance(s, str) else s
return required_fields.issubset(set(obj.keys()))
except Exception:
return False
for attempt in range(2):
for url, payload, mode in candidates:
if attempt > 0:
extra = "\n\n重要：必须输出包含 summary, issue_type, search_queries, keywords, components, likely_paths 全部字段的 JSON。"
payload = dict(payload)
msgs = [dict(m) for m in payload['messages']]
msgs[-1] = dict(msgs[-1])
msgs[-1]['content'] = msgs[-1].get('content', '') + extra
payload['messages'] = msgs
data = json.dumps(payload, ensure_ascii=False).encode('utf-8')
req = urllib.request.Request(url, data=data, headers=headers, method='POST')
try:
with urllib.request.urlopen(req, timeout=180) as resp:
text = resp.read().decode('utf-8', errors='replace')
obj = json.loads(text)
raw = ((obj.get('choices') or [{}])[0].get('message') or {}).get('content') or ''
# 尝试从 markdown 代码块中提取 JSON
if raw and not raw.strip().startswith('{'):
import re
m = re.search(r'```(?:json)?\s*\n?(.*?)\n?```', raw, re.S)
if m:
raw = m.group(1).strip()
if raw and validate_rewrite(raw):
content = raw
mode_used = mode
pathlib.Path('.ai_runtime/rewrite_raw.txt').write_text(content, encoding='utf-8')
break
except urllib.error.HTTPError as e:
err_body = e.read().decode('utf-8', errors='replace')[:500]
last_err = f'HTTP {e.code}: {err_body}'
print(f'[attempt {attempt+1}] {mode} 失败: {last_err}')
import time; time.sleep(2)
except Exception as e:
last_err = e
print(f'[attempt {attempt+1}] {mode} 异常: {e}')
if content:
break
if not content:
raise RuntimeError(f'LLM rewrite call failed: {last_err}')
pathlib.Path('.ai_runtime/issue_rewrite.txt').write_text(content, encoding='utf-8')
with open(os.environ['GITHUB_OUTPUT'], 'a', encoding='utf-8') as f:
f.write(f'mode_used={mode_used}\n')
PY
python3 .github/scripts/write_output.py issue_rewrite .ai_runtime/issue_rewrite.txt
- name: Normalize rewrite result
id: rewrite_norm
shell: bash
run: |
set -euo pipefail
cat > .ai_runtime/rewrite_raw.txt <<'EOF'
${{ steps.rewrite.outputs.issue_rewrite }}
EOF
python3 - <<'PY'
import json, re, pathlib
raw = pathlib.Path(".ai_runtime/rewrite_raw.txt").read_text(encoding="utf-8", errors="ignore").strip()
def parse_json(s):
try:
return json.loads(s)
except Exception:
m = re.search(r'\{.*\}', s, flags=re.S)
return json.loads(m.group(0)) if m else {}
d = parse_json(raw) if raw else {}
issue_type = d.get("issue_type")
if issue_type not in {"bug","enhancement","question","documentation","needs_more_info"}:
issue_type = "needs_more_info"
def clean_list(v, n):
if not isinstance(v, list):
return []
out = []
seen = set()
for x in v:
if isinstance(x, str):
s = x.strip()
if s and s.lower() not in seen:
seen.add(s.lower())
out.append(s)
if len(out) >= n:
break
return out
result = {
"summary": d.get("summary") if isinstance(d.get("summary"), str) else "",
"issue_type": issue_type,
"search_queries": clean_list(d.get("search_queries"), 6),
"keywords": clean_list(d.get("keywords"), 12),
"components": clean_list(d.get("components"), 8),
"likely_paths": clean_list(d.get("likely_paths"), 8)
}
pathlib.Path(".ai_runtime/rewrite.json").write_text(
json.dumps(result, ensure_ascii=False, indent=2),
encoding="utf-8"
)
print(json.dumps(result, ensure_ascii=False))
PY
echo "json=$(cat .ai_runtime/rewrite.json | jq -c .)" >> "$GITHUB_OUTPUT"
- name: Optional filter by issue type
if: env.AI_ONLY_TEMPLATE_TYPES == 'true'
shell: bash
run: |
set -euo pipefail
TYPE="$(jq -r '.issue_type' .ai_runtime/rewrite.json)"
case "$TYPE" in
bug|enhancement|documentation)
echo "continue"
;;
*)
echo "Skipping non-target issue type: $TYPE"
exit 78
;;
esac
- name: Retrieve context from repository scan
id: context_scan
shell: bash
run: |
set -euo pipefail
python3 - <<'PY'
import json, os, pathlib, re
TEXT_EXTS = {
".md", ".mdx", ".rst", ".txt", ".js", ".jsx", ".ts", ".tsx", ".py", ".go",
".java", ".rs", ".php", ".rb", ".yml", ".yaml", ".json", ".toml", ".sh", ".vue"
}
IGNORE_DIRS = {
".git", "node_modules", "dist", "build", "coverage", ".next", ".nuxt",
"__pycache__", ".venv", "venv", "vendor", "target", "out", ".ai_runtime"
}
rewrite = json.loads(pathlib.Path(".ai_runtime/rewrite.json").read_text(encoding="utf-8"))
issue = json.loads(pathlib.Path(".ai_runtime/issue.json").read_text(encoding="utf-8"))
# 构建查询词
query_terms = []
for key in ("search_queries", "keywords", "components", "likely_paths"):
query_terms.extend(rewrite.get(key, []))
query_terms.append(issue.get("title", ""))
query_terms = [t for t in query_terms if isinstance(t, str) and t.strip()]
def tokenize(text):
return set(re.findall(r"[A-Za-z0-9_./:#-]{3,}", text.lower()))
q_tokens = tokenize(" ".join(query_terms))
max_chars = int(os.environ.get("AI_MAX_CONTEXT_CHARS", "50000"))
root = pathlib.Path(".")
rows = []
for path in root.rglob("*"):
if not path.is_file():
continue
rel = path.relative_to(root).as_posix()
if any(part in rel for part in IGNORE_DIRS):
continue
if path.suffix.lower() not in TEXT_EXTS and not path.name.startswith("README"):
continue
try:
text = path.read_text(encoding="utf-8", errors="ignore")
except Exception:
continue
lower = text.lower()
score = 0
for t in query_terms:
tl = t.lower()
if tl in lower:
score += 2
if tl in rel.lower():
score += 4
if score <= 0:
continue
lines = text.splitlines()
for i, line in enumerate(lines):
ls = line.lower()
hit = sum(1 for t in query_terms if t.lower() in ls)
if hit <= 0:
continue
start = max(0, i - 18)
end = min(len(lines), i + 19)
block = "\n".join(f"{j+1:>5}: {lines[j]}" for j in range(start, end))
rows.append({
"path": rel,
"start_line": start + 1,
"end_line": end,
"text": block,
"score": score + hit * 5
})
rows.sort(key=lambda x: (-x["score"], x["path"], x["start_line"]))
# 输出去重后的结果
parts = []
cur = 0
seen = set()
for row in rows:
key = (row["path"], row["start_line"], row["end_line"])
if key in seen:
continue
seen.add(key)
block = f"--- FILE: {row['path']} (lines {row['start_line']}-{row['end_line']}) ---\n{row['text']}\n"
if cur + len(block) > max_chars:
break
parts.append(block)
cur += len(block)
pathlib.Path(".ai_runtime/final_context.txt").write_text("\n".join(parts), encoding="utf-8")
print(f"扫描完成: {len(rows)} 个匹配块, {len(parts)} 个输出, {cur} 字符")
PY
python3 .github/scripts/write_output.py context .ai_runtime/final_context.txt
- name: Fetch duplicate candidates
if: env.AI_ENABLE_DUPLICATE_CHECK == 'true'
id: dupes
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
shell: bash
run: |
set -euo pipefail
gh issue list \
--state all \
--limit "${AI_MAX_DUP_CANDIDATES}" \
--json number,title,body,state,updatedAt,url \
> .ai_runtime/issues_raw.json
python3 - <<'PY'
import json, pathlib, re
rewrite = json.loads(pathlib.Path(".ai_runtime/rewrite.json").read_text(encoding="utf-8"))
queries = " ".join(rewrite.get("search_queries", [])) + " " + " ".join(rewrite.get("keywords", []))
q = set(re.findall(r"[a-z0-9_./:-]{3,}", queries.lower()))
items = json.loads(pathlib.Path(".ai_runtime/issues_raw.json").read_text(encoding="utf-8"))
scored = []
for it in items:
text = ((it.get("title") or "") + "\n" + (it.get("body") or "")).lower()
toks = set(re.findall(r"[a-z0-9_./:-]{3,}", text))
inter = len(q & toks)
if inter == 0:
continue
union = max(len(q | toks), 1)
score = inter * 2 + (inter / union) * 100
scored.append({
"number": it["number"],
"title": it.get("title", ""),
"state": it.get("state", ""),
"url": it.get("url", ""),
"score": round(score, 2),
})
scored.sort(key=lambda x: (-x["score"], x["number"]))
top = scored[:12]
pathlib.Path(".ai_runtime/dupes.txt").write_text(
"\n".join([f"#{x['number']} [{x['state']}] score={x['score']}: {x['title']}" for x in top]),
encoding="utf-8"
)
PY
python3 .github/scripts/write_output.py candidates .ai_runtime/dupes.txt
- name: Prepare duplicate candidates fallback
if: env.AI_ENABLE_DUPLICATE_CHECK != 'true'
id: dupes_disabled
shell: bash
run: |
set -euo pipefail
echo "No duplicate candidates collected." > .ai_runtime/dupes.txt
python3 .github/scripts/write_output.py candidates .ai_runtime/dupes.txt
- name: Fetch recent PR candidates
if: env.AI_ENABLE_PR_SEARCH == 'true'
id: prs
env:
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
shell: bash
run: |
set -euo pipefail
gh pr list \
--state all \
--limit "${AI_MAX_PR_CANDIDATES}" \
--json number,title,body,mergedAt,state,url \
> .ai_runtime/prs_raw.json
python3 - <<'PY'
import json, pathlib, re
rewrite = json.loads(pathlib.Path(".ai_runtime/rewrite.json").read_text(encoding="utf-8"))
q = set(re.findall(r"[a-z0-9_./:-]{3,}", " ".join(rewrite.get("search_queries", []) + rewrite.get("keywords", [])).lower()))
items = json.loads(pathlib.Path(".ai_runtime/prs_raw.json").read_text(encoding="utf-8"))
out = []
for it in items:
text = ((it.get("title") or "") + "\n" + (it.get("body") or "")).lower()
toks = set(re.findall(r"[a-z0-9_./:-]{3,}", text))
inter = len(q & toks)
if inter <= 0:
continue
out.append((inter, it))
out.sort(key=lambda x: -x[0])
lines = []
for _, it in out[:8]:
lines.append(f"PR #{it['number']} [{it.get('state','')}]: {it.get('title','')} {it.get('url','')}")
pathlib.Path(".ai_runtime/prs.txt").write_text("\n".join(lines), encoding="utf-8")
PY
python3 .github/scripts/write_output.py candidates .ai_runtime/prs.txt
- name: Prepare PR candidates fallback
if: env.AI_ENABLE_PR_SEARCH != 'true'
id: prs_disabled
shell: bash
run: |
set -euo pipefail
echo "No PR candidates collected." > .ai_runtime/prs.txt
python3 .github/scripts/write_output.py candidates .ai_runtime/prs.txt
- name: Fetch recent commit candidates
if: env.AI_ENABLE_COMMIT_SEARCH == 'true'
id: commits
shell: bash
run: |
set -euo pipefail
python3 - <<'PY'
import json, pathlib, subprocess
rewrite = json.loads(pathlib.Path(".ai_runtime/rewrite.json").read_text(encoding="utf-8"))
keywords = rewrite.get("keywords", [])[:8]
try:
log = subprocess.check_output(
["git", "log", "--pretty=format:%H%x09%s", "-n", "200"],
text=True
)
except Exception:
log = ""
lines = []
for row in log.splitlines():
if "\t" not in row:
continue
sha, subject = row.split("\t", 1)
score = 0
lower = subject.lower()
for k in keywords:
if k.lower() in lower:
score += 1
if score > 0:
lines.append((score, f"{sha[:12]} {subject}"))
lines.sort(key=lambda x: -x[0])
pathlib.Path(".ai_runtime/commits.txt").write_text(
"\n".join([x[1] for x in lines[:10]]),
encoding="utf-8"
)
PY
python3 .github/scripts/write_output.py candidates .ai_runtime/commits.txt
- name: Prepare commit candidates fallback
if: env.AI_ENABLE_COMMIT_SEARCH != 'true'
id: commits_disabled
shell: bash
run: |
set -euo pipefail
echo "No commit candidates collected." > .ai_runtime/commits.txt
python3 .github/scripts/write_output.py candidates .ai_runtime/commits.txt
- name: Prepare final duplicate candidates
id: final_dupes
shell: bash
run: |
set -euo pipefail
if [ -s .ai_runtime/dupes.txt ]; then
cp .ai_runtime/dupes.txt .ai_runtime/final_dupes.txt
else
echo "No duplicate candidates collected." > .ai_runtime/final_dupes.txt
fi
python3 .github/scripts/write_output.py candidates .ai_runtime/final_dupes.txt
- name: Prepare final PR candidates
id: final_prs
shell: bash
run: |
set -euo pipefail
if [ -s .ai_runtime/prs.txt ]; then
cp .ai_runtime/prs.txt .ai_runtime/final_prs.txt
else
echo "No PR candidates collected." > .ai_runtime/final_prs.txt
fi
python3 .github/scripts/write_output.py candidates .ai_runtime/final_prs.txt
- name: Prepare final commit candidates
id: final_commits
shell: bash
run: |
set -euo pipefail
if [ -s .ai_runtime/commits.txt ]; then
cp .ai_runtime/commits.txt .ai_runtime/final_commits.txt
else
echo "No commit candidates collected." > .ai_runtime/final_commits.txt
fi
python3 .github/scripts/write_output.py candidates .ai_runtime/final_commits.txt
- name: Collect enriched context for deeper analysis
id: enriched
shell: bash
run: |
set -euo pipefail
python3 - <<'PY'
import json, os, pathlib, re, subprocess
issue = json.loads(pathlib.Path('.ai_runtime/issue.json').read_text(encoding='utf-8'))
rewrite = json.loads(pathlib.Path('.ai_runtime/rewrite.json').read_text(encoding='utf-8'))
# --- 1. 从 issue body 提取错误模式 ---
body = issue.get('body', '') or ''
error_patterns = []
# 提取堆栈跟踪（支持多种语言格式）
stacktrace_re = re.compile(
r'(?:(?:Traceback|Exception|Error|at\s+\S+\s*\(|File\s+".*?"|.*?Error:.*|.*?Exception:.*|'
r'(?:Caused by|Caused):.*|.*?\.java:\d+|.*?\.ts:\d+|.*?\.js:\d+|.*?\.py:\d+|'
r'(?:errno|EACCES|ENOENT|ECONNREFUSED|ETIMEDOUT)\s.*|'
r'(?:SIGTERM|SIGKILL|SIGSEGV).*|'
r'(?:fatal|panic|crash|FATAL|PANIC|CRASH)\s.*))',
re.MULTILINE | re.IGNORECASE
)
for m in stacktrace_re.finditer(body):
error_patterns.append(m.group(0).strip())
# 提取代码块内容
code_blocks = re.findall(r'```[\w]*\n(.*?)```', body, re.DOTALL)
for block in code_blocks[:5]:
stripped = block.strip()
if stripped and len(stripped) < 2000:
error_patterns.append(f"[code block] {stripped}")
# 提取 URL（可能指向日志、截图等）
urls = re.findall(r'https?://[^\s\)\]>]+', body)
# 去重并限制数量
seen = set()
unique_errors = []
for ep in error_patterns:
key = ep[:100].lower()
if key not in seen:
seen.add(key)
unique_errors.append(ep)
error_patterns = unique_errors[:15]
# --- 2. 读取 likely_paths 指向的文件内容（关键词定位 + 上下文窗口） ---
likely_paths = rewrite.get('likely_paths', [])
search_terms = []
for key in ("search_queries", "keywords", "components"):
search_terms.extend(rewrite.get(key, []))
search_terms = [t for t in search_terms if isinstance(t, str) and len(t) >= 3]
file_contents = {}
max_file_chars = 8000 # 每个文件最多 8000 字符
context_lines = 40 # 关键词命中行前后各取 40 行
for fp in likely_paths[:6]:
p = pathlib.Path(fp)
if not p.is_file():
p = pathlib.Path('.') / fp
if not (p.is_file() and p.stat().st_size < 500_000):
continue
try:
content = p.read_text(encoding='utf-8', errors='replace')
except Exception:
continue
if len(content) <= max_file_chars:
file_contents[fp] = content
continue
# 关键词定位：找到匹配行，提取上下文窗口
lines = content.splitlines()
matched_ranges = []
for i, line in enumerate(lines):
ll = line.lower()
if any(t.lower() in ll for t in search_terms):
start = max(0, i - context_lines)
end = min(len(lines), i + context_lines + 1)
matched_ranges.append((start, end))
if matched_ranges:
# 合并重叠区间
matched_ranges.sort()
merged = [matched_ranges[0]]
for s, e in matched_ranges[1:]:
if s <= merged[-1][1]:
merged[-1] = (merged[-1][0], max(merged[-1][1], e))
else:
merged.append((s, e))
# 提取合并后的区间内容
parts = []
cur = 0
for s, e in merged:
if s > cur:
parts.append(f"... [跳过 {s - cur} 行] ...")
parts.append("\n".join(lines[s:e]))
cur = e
if cur < len(lines):
parts.append(f"... [剩余 {len(lines) - cur} 行] ...")
result = "\n".join(parts)
if len(result) > max_file_chars:
result = result[:max_file_chars] + f'\n... [截断，共 {len(content)} 字符]'
file_contents[fp] = result
else:
# 无关键词命中，回退到读取前半部分
file_contents[fp] = content[:max_file_chars] + f'\n... [截断，共 {len(content)} 字符]'
# --- 3. 获取相关文件的 git 历史 ---
git_history = {}
for fp in list(file_contents.keys())[:4]:
try:
log = subprocess.check_output(
['git', 'log', '--pretty=format:%h %ad %s', '--date=short', '-n', '5', '--', fp],
text=True, stderr=subprocess.DEVNULL
)
if log.strip():
git_history[fp] = log.strip().splitlines()
except Exception:
pass
# --- 4. 获取仓库结构概览（顶层目录） ---
repo_structure = []
try:
for item in sorted(pathlib.Path('.').iterdir()):
if item.name.startswith('.'):
continue
if item.is_dir():
repo_structure.append(f"{item.name}/")
else:
repo_structure.append(item.name)
except Exception:
pass
enriched = {
"error_patterns": error_patterns,
"code_blocks_from_issue": [b.strip() for b in code_blocks[:5] if b.strip()],
"urls_from_issue": urls[:10],
"file_contents": file_contents,
"git_history": git_history,
"repo_structure": repo_structure[:30],
}
pathlib.Path('.ai_runtime/enriched_context.json').write_text(
json.dumps(enriched, ensure_ascii=False, indent=2),
encoding='utf-8'
)
# 摘要统计
print(f"错误模式: {len(error_patterns)} 条")
print(f"读取文件: {len(file_contents)} 个")
print(f"Git 历史: {len(git_history)} 个文件")
print(f"Issue URL: {len(urls)} 个")
PY
- name: Stage 2 - maintainer-grade triage
id: triage
shell: bash
env:
LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
LLM_BASE_URL: ${{ secrets.LLM_BASE_URL }}
LLM_MODEL: ${{ secrets.LLM_MODEL }}
run: |
set -euo pipefail
cat > .ai_runtime/triage_schema.json <<'JSON'
{
"name": "issue_response",
"parameters": {
"type": "object",
"properties": {
"summary": { "type": "string" },
"classification": {
"type": "string",
"enum": ["bug", "enhancement", "question", "documentation", "needs_more_info"]
},
"support_status": {
"type": "string",
"enum": ["supported", "partially_supported", "not_supported", "already_fixed_unreleased", "needs_more_info"]
},
"analysis": { "type": "string" },
"solution": { "type": "string" },
"workaround": { "type": "string" },
"error_pattern_summary": { "type": "string" },
"related_files": {
"type": "array",
"items": { "type": "string" }
},
"actionable_steps": {
"type": "array",
"items": { "type": "string" }
},
"roadmap": { "type": "string" },
"needs_human_followup": { "type": "boolean" },
"duplicate_confidence": {
"type": "string",
"enum": ["low", "medium", "high"]
},
"duplicate_issues": {
"type": "array",
"items": { "type": "string" }
},
"labels": {
"type": "array",
"items": {
"type": "string",
"enum": ["bug", "enhancement", "question", "documentation", "duplicate", "help wanted", "needs more info"]
}
}
},
"required": ["summary", "classification", "support_status", "analysis", "solution", "workaround", "error_pattern_summary", "related_files", "actionable_steps", "needs_human_followup", "duplicate_confidence", "duplicate_issues", "labels"]
}
}
JSON
python3 - <<'PY'
import pathlib
prompt = (
"你是一位资深仓库维护者。根据提供的 Issue 信息、代码上下文和历史数据，做出专业判断。\n\n"
"## 输出要求\n\n"
"你必须输出严格的 JSON 对象，包含以下字段（缺一不可）：\n\n"
"- summary: 一句话总结 Issue 核心问题（必填）\n"
"- classification: 分类，必须是 bug/enhancement/question/documentation/needs_more_info 之一（必填）\n"
"- support_status: 支持状态，必须是 supported/partially_supported/not_supported/already_fixed_unreleased/needs_more_info 之一（必填）\n"
"- analysis: 详细分析问题根因，引用具体的代码路径和逻辑（必填，至少 2 句话）\n"
"- solution: 详细的修复方案，必须包含：受影响的文件路径、具体的函数/方法名、需要修改的代码逻辑描述、修改后的预期行为。格式示例：\"在 `packages/backend/src/auth/service.ts` 的 `validateToken()` 方法中，第 42 行的过期检查逻辑需要改为...\"（必填，至少 2 句话）\n"
"- workaround: 用户可立即执行的临时解决方案，不依赖代码修改或重新部署。例如：修改配置文件参数、设置环境变量、重启服务、手动执行命令等。如果确实没有临时方案则为空字符串（必填）\n"
"- error_pattern_summary: 错误模式总结，描述从 Issue 中提取的错误类型、发生位置和调用链路（如果存在错误信息则必填，否则为空字符串）\n"
"- related_files: 相关文件列表，格式如 [\"packages/backend/src/auth/service.ts\"]，列出与问题直接相关的源代码文件（必填，至少 1 个，最多 8 个）\n"
"- actionable_steps: 可执行步骤列表，每步必须足够具体以供 AI 编程助手直接执行。格式要求：引用文件路径（如 `src/foo.ts`）、函数名（如 `handleLogin()`）、行号范围、以及需要执行的具体操作（如「将第 15 行的 `==` 改为 `===`」「在 `X` 函数末尾添加空值检查」）。禁止模糊描述如「检查相关代码」「修复问题」。（必填，至少 1 步，最多 6 步）\n"
"- roadmap: 如果 not_supported，描述实现路线；否则为空字符串\n"
"- needs_human_followup: 仅当证据明显不足或问题涉及业务决策时设为 true（必填）\n"
"- duplicate_confidence: 与已有 Issue 的重复程度 low/medium/high（必填）\n"
'- duplicate_issues: 可能重复的 Issue 编号列表，格式如 ["#123"]（必填）\n'
"- labels: 1-3 个标签，只能从 bug/enhancement/question/documentation/duplicate/help wanted/needs more info 中选择（必填）\n\n"
"## 分析增强指引\n\n"
"当「Enriched Context」部分提供以下信息时，你必须充分利用：\n\n"
"1. **错误模式 (error_patterns)**：从 Issue 正文中提取的堆栈跟踪、错误消息、异常信息。"
" 分析错误类型、发生位置、调用链路，定位根因。\n"
"2. **代码块 (code_blocks_from_issue)**：Issue 中附带的代码片段。"
" 检查代码逻辑问题、类型错误、API 误用等。\n"
"3. **相关文件内容 (file_contents)**：仓库中与 Issue 相关的源代码文件。"
" 逐文件分析代码逻辑，引用具体的函数名、行号和代码路径。\n"
"4. **Git 历史 (git_history)**：相关文件的最近提交记录。"
" 检查是否有最近的变更可能导致了问题，或已有修复但未发布。\n"
"5. **仓库结构 (repo_structure)**：仓库顶层目录结构。"
" 帮助判断 Issue 涉及的模块和组件。\n\n"
"## 判断原则\n\n"
"- 如果代码上下文中有明确的相关文件和逻辑，大胆给出分析和方案\n"
"- analysis 中必须引用具体的文件路径（如 `packages/backend/src/auth/service.ts`）和函数名\n"
"- solution 必须写成「AI 编程助手可直接执行」的粒度：文件路径 + 函数名 + 行号 + 具体修改内容 + 预期结果\n"
"- actionable_steps 必须写成可直接执行的指令，每步包含：目标文件、定位方式（行号/函数名/代码模式）、具体操作、验证方式\n"
"- workaround 必须提供用户可立即执行的临时规避方法，不依赖代码修改（如配置变更、环境变量、手动操作步骤）\n"
"- 只有在完全无法判断时才设 needs_human_followup=true\n"
"- labels 必须从允许的列表中选择，不要自创标签\n"
"- 输出必须是严格 JSON，不要输出 markdown 或其他文本"
)
pathlib.Path('.ai_runtime/triage_system_prompt.txt').write_text(prompt, encoding='utf-8')
PY
python3 - <<'PY'
import json, pathlib
issue = json.loads(pathlib.Path('.ai_runtime/issue.json').read_text(encoding='utf-8'))
rewrite_json = pathlib.Path('.ai_runtime/rewrite.json').read_text(encoding='utf-8')
context = pathlib.Path('.ai_runtime/final_context.txt').read_text(encoding='utf-8') if pathlib.Path('.ai_runtime/final_context.txt').exists() else ''
dupes = pathlib.Path('.ai_runtime/final_dupes.txt').read_text(encoding='utf-8') if pathlib.Path('.ai_runtime/final_dupes.txt').exists() else ''
prs = pathlib.Path('.ai_runtime/final_prs.txt').read_text(encoding='utf-8') if pathlib.Path('.ai_runtime/final_prs.txt').exists() else ''
commits = pathlib.Path('.ai_runtime/final_commits.txt').read_text(encoding='utf-8') if pathlib.Path('.ai_runtime/final_commits.txt').exists() else ''
# 读取富化上下文
enriched_str = ''
enriched_path = pathlib.Path('.ai_runtime/enriched_context.json')
if enriched_path.exists():
enriched = json.loads(enriched_path.read_text(encoding='utf-8'))
parts = []
if enriched.get('error_patterns'):
parts.append("[错误模式 - 从 Issue 提取]\n" + "\n---\n".join(enriched['error_patterns']))
if enriched.get('code_blocks_from_issue'):
parts.append("[Issue 中的代码块]\n" + "\n---\n".join(enriched['code_blocks_from_issue']))
if enriched.get('file_contents'):
fc_lines = []
for fpath, content in enriched['file_contents'].items():
fc_lines.append(f"### {fpath}\n```\n{content}\n```")
parts.append("[相关文件内容]\n" + "\n\n".join(fc_lines))
if enriched.get('git_history'):
gh_lines = []
for fpath, history in enriched['git_history'].items():
gh_lines.append(f"### {fpath}\n" + "\n".join(history))
parts.append("[文件 Git 历史]\n" + "\n\n".join(gh_lines))
if enriched.get('repo_structure'):
parts.append("[仓库结构]\n" + "\n".join(enriched['repo_structure']))
enriched_str = "\n\n".join(parts)
prompt = f"""[Issue]
标题：
{issue.get('title','')}
正文：
{issue.get('body','')}
[Stage 1 Rewrite]
{rewrite_json}
[Retrieved Context]
{context}
[Enriched Context]
{enriched_str}
[Duplicate Candidates]
{dupes}
[PR Candidates]
{prs}
[Commit Candidates]
{commits}"""
pathlib.Path('.ai_runtime/triage_input_prompt.txt').write_text(prompt, encoding='utf-8')
PY
python3 - <<'PY'
import json, os, pathlib, urllib.request, urllib.error
base_url = os.environ['LLM_BASE_URL'].rstrip('/')
api_key = os.environ['LLM_API_KEY']
model = os.environ['LLM_MODEL']
system_prompt = pathlib.Path('.ai_runtime/triage_system_prompt.txt').read_text(encoding='utf-8')
input_prompt = pathlib.Path('.ai_runtime/triage_input_prompt.txt').read_text(encoding='utf-8')
schema = json.loads(pathlib.Path('.ai_runtime/triage_schema.json').read_text(encoding='utf-8'))
candidates = [
(f"{base_url}/chat/completions", {
"model": model,
"temperature": 0.2,
"response_format": {"type": "json_object"},
"messages": [
{"role": "system", "content": system_prompt + "\n你必须直接输出 JSON 对象，不要输出 markdown。"},
{"role": "user", "content": input_prompt},
],
}, 'chat_json'),
(f"{base_url}/chat/completions", {
"model": model,
"temperature": 0.2,
"messages": [
{"role": "system", "content": system_prompt + "\n你必须直接输出 JSON 对象，不要输出 markdown 或代码块。"},
{"role": "user", "content": input_prompt + "\n\n请直接输出 JSON，不要包含 ```json 代码块标记。"},
],
}, 'chat_plain'),
]
headers = {
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'github-actions-ai-triage/1.0',
}
last_err = None
content = ''
mode_used = ''
required_fields = {"summary", "classification", "support_status", "analysis", "solution",
"error_pattern_summary", "related_files", "actionable_steps",
"needs_human_followup", "duplicate_confidence", "duplicate_issues", "labels"}
def validate_triage(s):
"""检查内容是否包含 triage 必需字段"""
try:
obj = json.loads(s) if isinstance(s, str) else s
return required_fields.issubset(set(obj.keys()))
except Exception:
return False
# 最多重试 2 次（首次 + 1 次重试）
for attempt in range(2):
for url, payload, mode in candidates:
# 重试时加入更明确的格式要求
if attempt > 0:
extra = "\n\n重要：你必须输出包含以下全部字段的 JSON：summary, classification, support_status, analysis, solution, error_pattern_summary, related_files, actionable_steps, needs_human_followup, duplicate_confidence, duplicate_issues, labels。缺少任何字段都会导致解析失败。"
payload = dict(payload)
msgs = [dict(m) for m in payload['messages']]
msgs[-1] = dict(msgs[-1])
msgs[-1]['content'] = msgs[-1].get('content', '') + extra
payload['messages'] = msgs
data = json.dumps(payload, ensure_ascii=False).encode('utf-8')
req = urllib.request.Request(url, data=data, headers=headers, method='POST')
try:
with urllib.request.urlopen(req, timeout=240) as resp:
text = resp.read().decode('utf-8', errors='replace')
obj = json.loads(text)
raw = ((obj.get('choices') or [{}])[0].get('message') or {}).get('content') or ''
# 尝试从 markdown 代码块中提取 JSON
if raw and not raw.strip().startswith('{'):
import re
m = re.search(r'```(?:json)?\s*\n?(.*?)\n?```', raw, re.S)
if m:
raw = m.group(1).strip()
if raw and validate_triage(raw):
content = raw
mode_used = mode
pathlib.Path('.ai_runtime/triage_raw.txt').write_text(content, encoding='utf-8')
break
elif raw:
last_err = f'LLM 返回内容缺少必需字段: {required_fields - set(json.loads(raw).keys()) if isinstance(raw, str) else "parse error"}'
except urllib.error.HTTPError as e:
err_body = e.read().decode('utf-8', errors='replace')[:500]
last_err = f'HTTP {e.code}: {err_body}'
print(f'[attempt {attempt+1}] {mode} 失败: {last_err}')
import time; time.sleep(2)
except Exception as e:
last_err = e
print(f'[attempt {attempt+1}] {mode} 异常: {e}')
if content:
break
print(f'第 {attempt+1} 次尝试未通过验证，准备重试...')
if not content:
raise RuntimeError(f'LLM triage call failed after retries: {last_err}')
pathlib.Path('.ai_runtime/issue_response.txt').write_text(content, encoding='utf-8')
with open(os.environ['GITHUB_OUTPUT'], 'a', encoding='utf-8') as f:
f.write(f'mode_used={mode_used}\n')
PY
python3 .github/scripts/write_output.py issue_response .ai_runtime/issue_response.txt
- name: Normalize triage result
id: triage_norm
shell: bash
run: |
set -euo pipefail
cat > .ai_runtime/triage_raw.txt <<'EOF'
${{ steps.triage.outputs.issue_response }}
EOF
python3 - <<'PY'
import json, pathlib, re
raw = pathlib.Path(".ai_runtime/triage_raw.txt").read_text(encoding="utf-8", errors="ignore").strip()
def parse_json(s):
try:
return json.loads(s)
except Exception:
m = re.search(r'\{.*\}', s, flags=re.S)
return json.loads(m.group(0)) if m else {}
d = parse_json(raw) if raw else {}
allowed_labels = {"bug","enhancement","question","documentation","duplicate","help wanted","needs more info"}
allowed_class = {"bug","enhancement","question","documentation","needs_more_info"}
allowed_support = {"supported","partially_supported","not_supported","already_fixed_unreleased","needs_more_info"}
allowed_dup = {"low","medium","high"}
def s(v, default=""):
return v if isinstance(v, str) else default
def b(v, default=False):
return v if isinstance(v, bool) else default
def arr(v):
return v if isinstance(v, list) else []
labels = []
seen = set()
for x in arr(d.get("labels")):
if isinstance(x, str):
t = x.strip().lower()
if t in allowed_labels and t not in seen:
labels.append(t)
seen.add(t)
classification = s(d.get("classification"), "needs_more_info")
if classification not in allowed_class:
classification = "needs_more_info"
if classification != "needs_more_info" and classification not in labels and len(labels) < 3:
labels.insert(0, classification)
result = {
"summary": s(d.get("summary"), "AI 未能稳定生成摘要。"),
"classification": classification,
"support_status": s(d.get("support_status"), "needs_more_info"),
"analysis": s(d.get("analysis"), "AI 未能稳定生成分析，请维护者人工复核。"),
"solution": s(d.get("solution"), "暂无稳定自动建议。"),
"workaround": s(d.get("workaround"), ""),
"error_pattern_summary": s(d.get("error_pattern_summary"), ""),
"related_files": [x.strip() for x in arr(d.get("related_files")) if isinstance(x, str) and x.strip()][:8],
"actionable_steps": [x.strip() for x in arr(d.get("actionable_steps")) if isinstance(x, str) and x.strip()][:6],
"roadmap": s(d.get("roadmap"), ""),
"needs_human_followup": b(d.get("needs_human_followup"), False),
"duplicate_confidence": s(d.get("duplicate_confidence"), "low"),
"duplicate_issues": [x.strip() for x in arr(d.get("duplicate_issues")) if isinstance(x, str) and re.fullmatch(r"#\d+", x.strip())],
"labels": labels[:3]
}
# 分析或方案为 fallback 默认值时，自动标记需要人工复核
if result["analysis"] == "AI 未能稳定生成分析，请维护者人工复核。" or \
result["solution"] == "暂无稳定自动建议。":
result["needs_human_followup"] = True
if result["support_status"] not in allowed_support:
result["support_status"] = "needs_more_info"
if result["duplicate_confidence"] not in allowed_dup:
result["duplicate_confidence"] = "low"
if result["duplicate_confidence"] == "high" and result["duplicate_issues"] and "duplicate" not in result["labels"] and len(result["labels"]) < 3:
result["labels"].append("duplicate")
pathlib.Path(".ai_runtime/triage.json").write_text(
json.dumps(result, ensure_ascii=False, indent=2),
encoding="utf-8"
)
print(json.dumps(result, ensure_ascii=False))
PY
echo "json=$(cat .ai_runtime/triage.json | jq -c .)" >> "$GITHUB_OUTPUT"
- name: Create or update AI comment and labels
uses: actions/github-script@v7
env:
AI_AUTO_LABEL: ${{ env.AI_AUTO_LABEL }}
RESULT_JSON: ${{ steps.triage_norm.outputs.json }}
ISSUE_NUMBER: ${{ steps.issue.outputs.issue_number }}
with:
script: |
const marker = "";
const result = JSON.parse(process.env.RESULT_JSON || "{}");
const owner = context.repo.owner;
const repo = context.repo.repo;
const issue_number = parseInt(process.env.ISSUE_NUMBER, 10) || context.issue.number;
function supportText(s) {
switch (s) {
case "supported":
return "✅ 当前仓库上下文显示：该行为已支持。";
case "partially_supported":
return "🟡 当前仓库上下文显示：该行为仅部分支持。";
case "not_supported":
return "⚠️ 当前仓库上下文显示：当前暂不支持。";
case "already_fixed_unreleased":
return "🛠️ 当前仓库上下文显示：可能已修复，但未正式发布。";
default:
return "❓ 当前证据不足，需要更多信息。";
}
}
function classificationEmoji(c) {
switch (c) {
case "bug": return "🐛";
case "enhancement": return "✨";
case "question": return "❓";
case "documentation": return "📝";
default: return "🔍";
}
}
const lines = [
marker,
"### 🤖 AI Issue 智能分析",
"",
`> **分类**: ${classificationEmoji(result.classification)} ${result.classification || "unknown"} | **状态**: ${supportText(result.support_status)}`,
"",
"---",
"",
"#### 📋 摘要",
result.summary || "暂无",
""
];
// 错误模式分析（仅 bug 类型显示）
if (result.error_pattern_summary && result.classification === "bug") {
lines.push("#### 🔍 错误模式");
lines.push(result.error_pattern_summary);
lines.push("");
}
// 相关文件
if (Array.isArray(result.related_files) && result.related_files.length > 0) {
lines.push("#### 📁 相关文件");
for (const f of result.related_files) {
lines.push(`- \`${f}\``);
}
lines.push("");
}
lines.push("#### 🔬 分析");
lines.push(result.analysis || "暂无");
lines.push("");
lines.push("#### 💡 建议方案");
lines.push(result.solution || "暂无");
lines.push("");
// 临时解决方案（workaround）
if (result.workaround) {
lines.push("#### 🩹 临时解决方案");
lines.push(result.workaround);
lines.push("");
}
// 可执行步骤
if (Array.isArray(result.actionable_steps) && result.actionable_steps.length > 0) {
lines.push("#### 🎯 可执行步骤");
for (let i = 0; i < result.actionable_steps.length; i++) {
lines.push(`${i + 1}. ${result.actionable_steps[i]}`);
}
lines.push("");
}
if (result.roadmap && result.support_status === "not_supported") {
lines.push("#### 🗺️ 实现路线");
lines.push(result.roadmap);
lines.push("");
}
if (Array.isArray(result.duplicate_issues) && result.duplicate_issues.length > 0) {
lines.push("#### 🔗 可能重复的 Issue");
lines.push(`- **置信度**: ${result.duplicate_confidence || "low"}`);
lines.push(`- **候选**: ${result.duplicate_issues.join("、")}`);
lines.push("");
}
if (result.needs_human_followup) {
lines.push("#### ⚠️ 维护建议");
lines.push("建议维护者人工复核后再做最终结论。");
lines.push("");
}
lines.push("---");
lines.push("");
lines.push("📌 使用说明
");
lines.push("");
lines.push("| 操作 | 方法 |");
lines.push("|------|------|");
lines.push("| 🔄 重新分析 | 在评论区输入 `/ai-analyze` |");
lines.push("| 🏷️ 自动标签 | 分析结果会自动添加分类标签 |");
lines.push("| ⚠️ 需人工复核 | 低置信度分析会自动标记 `needs-review` |");
lines.push("| 🔗 重复检测 | 自动搜索相似 Issue 并关联 |");
lines.push("");
lines.push(" ");
lines.push("");
lines.push("_🤖 此评论由 AI 自动化工作流生成 | 结构化输出 + 新评论模式 + 标签白名单 + 代码上下文检索 + 错误模式分析 + AI Agent 可消费方案_");
const body = lines.join("\n");
// 始终创建新评论（不再编辑旧评论，保留历史分析记录）
const comments = await github.paginate(
github.rest.issues.listComments,
{ owner, repo, issue_number, per_page: 100 }
);
const existing = comments.find(
c => typeof c.body === "string" && c.body.includes(marker)
);
await github.rest.issues.createComment({
owner,
repo,
issue_number,
body
});
if (process.env.AI_AUTO_LABEL === "true") {
const issue = await github.rest.issues.get({
owner,
repo,
issue_number
});
const existingLabels = new Set(
(issue.data.labels || []).map(l => typeof l === "string" ? l : l.name)
);
const allowed = new Set([
"bug",
"enhancement",
"question",
"documentation",
"duplicate",
"help wanted",
"needs more info",
"needs-review"
]);
// AI 管理的标签：重新分析时先移除这些旧标签
const aiManagedLabels = [
"bug", "enhancement", "question", "documentation",
"duplicate", "needs more info", "needs-review"
];
// 重新分析时移除旧的 AI 管理标签（无论是否有旧评论）
const toRemove = aiManagedLabels.filter(x => existingLabels.has(x));
if (toRemove.length > 0) {
try {
await github.rest.issues.removeLabel({
owner,
repo,
issue_number,
name: toRemove
});
} catch (e) {
// 部分标签可能不存在，忽略错误
}
}
// 计算新标签
const desired = Array.isArray(result.labels) ? result.labels : [];
let toAdd = desired
.map(x => String(x).trim().toLowerCase())
.filter(x => allowed.has(x))
.filter((x, i, arr) => arr.indexOf(x) === i);
// 置信度阈值：低置信度时添加 needs-review
if (result.needs_human_followup) {
if (!toAdd.includes("needs more info")) {
toAdd.push("needs-review");
}
}
// 过滤掉已存在的
toAdd = toAdd.filter(x => !existingLabels.has(x));
if (toAdd.length > 0) {
await github.rest.issues.addLabels({
owner,
repo,
issue_number,
labels: toAdd
});
}
}
- name: Step summary
if: always()
shell: bash
run: |
{
echo "## AI Issue Smart Reply"
echo ""
echo "- Issue: #$(jq -r '.number' .ai_runtime/issue.json 2>/dev/null || echo 'unknown')"
echo "- Auto label: ${AI_AUTO_LABEL}"
echo "- Duplicate check: ${AI_ENABLE_DUPLICATE_CHECK}"
echo "- PR search: ${AI_ENABLE_PR_SEARCH}"
echo "- Commit search: ${AI_ENABLE_COMMIT_SEARCH}"
echo ""
echo "### Rewrite"
echo '```json'
cat .ai_runtime/rewrite.json 2>/dev/null || echo '{}'
echo '```'
echo ""
echo "### Enriched Context"
echo '```json'
python3 -c "
import json, pathlib
p = pathlib.Path('.ai_runtime/enriched_context.json')
if p.exists():
d = json.loads(p.read_text())
print(json.dumps({
'error_patterns': len(d.get('error_patterns', [])),
'file_contents': list(d.get('file_contents', {}).keys()),
'git_history': list(d.get('git_history', {}).keys()),
}, ensure_ascii=False, indent=2))
else:
print('{}')
" 2>/dev/null || echo '{}'
echo '```'
echo ""
echo "### Triage"
echo '```json'
cat .ai_runtime/triage.json 2>/dev/null || echo '{}'
echo '```'
} >> "$GITHUB_STEP_SUMMARY"