CoT 유출 없이 추론 강화 - ReAct·SC 실전

// react-agent.ts
import OpenAI from "openai";

type ToolResult = {
  tool: string;
  input: unknown;
  output: unknown;
};

const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

async function tool_search(query: string) {
  // 실제로는 사내 검색/벡터DB/웹 검색 등을 연결
  return {
    topHits: [
      { title: "Runbook: EKS 503 triage", url: "https://internal/runbook/eks-503", snippet: "Check ingress, endpoints, pod readiness..." }
    ]
  };
}

async function tool_kubectl(args: { namespace: string; resource: string }) {
  // 실제로는 RBAC 걸린 실행기(또는 관측 API)로 대체
  return {
    namespace: args.namespace,
    resource: args.resource,
    summary: "3 pods Running, 1 pod CrashLoopBackOff, endpoints empty"
  };
}

const tools = {
  tool_search,
  tool_kubectl
};

function safeUserFacingAnswer(answer: string, evidence: ToolResult[]) {
  // CoT 대신 증거 요약만 노출
  return {
    answer,
    evidence: evidence.map(e => ({ tool: e.tool, output: e.output }))
  };
}

export async function reactSolve(userQuestion: string) {
  const evidence: ToolResult[] = [];

  const system = [
    "You are an operations assistant.",
    "Do not reveal chain-of-thought.",
    "Use tools when needed.",
    "When you respond to the user, output only JSON with keys: answer, evidence.",
    "Evidence must be brief and based on tool outputs.",
    "If unsure, ask a clarifying question in answer."
  ].join("\n");

  let messages: any[] = [
    { role: "system", content: system },
    { role: "user", content: userQuestion }
  ];

  for (let step = 0; step < 6; step++) {
    const resp = await client.chat.completions.create({
      model: "gpt-4.1-mini",
      messages,
      temperature: 0.2,
      // function calling 또는 tool calling을 쓰는 방식이 가장 안전
      tools: [
        {
          type: "function",
          function: {
            name: "tool_search",
            description: "Search runbooks/docs.",
            parameters: {
              type: "object",
              properties: { query: { type: "string" } },
              required: ["query"]
            }
          }
        },
        {
          type: "function",
          function: {
            name: "tool_kubectl",
            description: "Get cluster resource summary.",
            parameters: {
              type: "object",
              properties: {
                namespace: { type: "string" },
                resource: { type: "string" }
              },
              required: ["namespace", "resource"]
            }
          }
        }
      ]
    });

    const msg = resp.choices[0].message;

    // 툴 호출이 있으면 실행
    if (msg.tool_calls && msg.tool_calls.length > 0) {
      messages.push(msg);

      for (const call of msg.tool_calls) {
        const name = call.function.name as keyof typeof tools;
        const input = JSON.parse(call.function.arguments || "{}");
        const output = await tools[name](input as any);

        evidence.push({ tool: name, input, output });

        messages.push({
          role: "tool",
          tool_call_id: call.id,
          content: JSON.stringify(output)
        });
      }

      continue;
    }

    // 최종 응답은 JSON으로만 받는다
    const parsed = JSON.parse(msg.content || "{}");
    return safeUserFacingAnswer(parsed.answer, evidence);
  }

  return safeUserFacingAnswer(
    "추가 정보가 필요합니다. 대상 네임스페이스와 서비스/인그레스 이름을 알려주세요.",
    evidence
  );
}

# self_consistency.py
from collections import Counter
import json
from openai import OpenAI

client = OpenAI()

def sample_answer(question: str, seed: int):
  system = "\n".join([
    "You are a helpful assistant.",
    "Do not reveal chain-of-thought.",
    "Return only JSON: {\"final\": string}."
  ])

  resp = client.chat.completions.create(
    model="gpt-4.1-mini",
    temperature=0.8,
    messages=[
      {"role": "system", "content": system},
      {"role": "user", "content": question + "\nUse concise final answer."}
    ],
  )

  content = resp.choices[0].message.content or "{}"
  return json.loads(content).get("final", "")

def self_consistent(question: str, k: int = 7):
  answers = [sample_answer(question, i) for i in range(k)]
  counts = Counter(answers)
  best, freq = counts.most_common(1)[0]
  return {
    "final": best,
    "agreement": freq / k,
    "candidates": counts.most_common(5)
  }

if __name__ == "__main__":
  q = "EKS에서 503이 발생할 때 가장 먼저 확인할 3가지는?"
  print(self_consistent(q, k=9))

if risk == high:
  use ReAct (tools required)
else:
  run SC
  if agreement < threshold:
    use ReAct
    run SC again with evidence
return final