v3: gstack-code-gen 生成

agents/agent1_parser.py

@@ -0,0 +1,34 @@
+import json
+from data.field_tree import FieldTree, Field
+from agents.llm import LLMClient
+
+P1 = "You are a COBOL COPYBOOK parser. Output JSON: {\"fields\":[{\"name\":\"...\",\"level\":N,\"pic\":\"...\",\"usage\":\"DISPLAY|COMP-3|COMP\",\"offset\":N,\"length\":N,\"decimal\":N,\"signed\":bool,\"occurs\":N|null,\"redefines\":\"...\"|null,\"conditions\":[{\"name\":\"...\",\"value\":\"...\"}],\"children\":[...]}]} Return JSON only."
+
+
+class Agent1Parser:
+    def __init__(self, llm: LLMClient):
+        self.llm = llm
+
+    def parse(self, text: str) -> FieldTree:
+        r = self.llm.call([{"role": "system", "content": P1}, {"role": "user", "content": text}])
+        try:
+            return self._load(json.loads(r))
+        except:
+            return FieldTree(copybook_name="parse_error")
+
+    def _load(self, d):
+        return FieldTree(fields=self._fields(d.get("fields", []), 0))
+
+    def _fields(self, raw, off):
+        result = []
+        cur = off
+        for rf in raw:
+            f = Field(name=rf.get("name", ""), level=rf.get("level", 0), pic=rf.get("pic", ""),
+                      usage=rf.get("usage", "DISPLAY"), offset=cur, length=rf.get("length", 0),
+                      decimal=rf.get("decimal", 0), signed=rf.get("signed", False),
+                      occurs=rf.get("occurs"), redefines=rf.get("redefines"),
+                      conditions=rf.get("conditions", []))
+            f.children = self._fields(rf.get("children", []), cur)
+            cur += f.length
+            result.append(f)
+        return result

agents/agent2_data.py

@@ -0,0 +1,24 @@
+import json
+from data.field_tree import FieldTree
+from data.test_case import TestCase, TestSuite, SparkConfig
+from agents.llm import LLMClient
+
+P2 = "You are a COBOL test data designer. Given a FieldTree, generate boundary test cases. Output: {\"test_cases\":[{\"id\":\"TC-001\",\"fields\":{\"FIELD\":value},\"coverage_targets\":[\"DP-001\"]}]} JSON only."
+
+
+class Agent2Data:
+    def __init__(self, llm: LLMClient):
+        self.llm = llm
+
+    def design(self, tree: FieldTree, target="boundary", spark_mode=False) -> TestSuite:
+        tree_d = {"fields": [{"name": f.name, "pic": f.pic, "usage": f.usage, "length": f.length,
+                              "decimal": f.decimal, "signed": f.signed} for f in tree.flatten().values()]}
+        r = self.llm.call([{"role": "system", "content": P2}, {"role": "user", "content": json.dumps(tree_d)}])
+        try:
+            tcs = [TestCase(**tc) for tc in json.loads(r).get("test_cases", [])]
+        except:
+            tcs = [TestCase(id="TC-FALLBACK", fields={"BR-AMT": 0})]
+        s = TestSuite(test_cases=tcs)
+        if spark_mode:
+            s.spark_config = SparkConfig(num_records=1000)
+        return s

agents/agent3_diagnostic.py

@@ -0,0 +1,13 @@
+from agents.llm import LLMClient
+from data.diff_result import FieldResult
+
+P3 = "You are a COBOL-Java diff analyzer. Given a field mismatch, explain why. Output: {\"issue_type\":\"...\",\"confidence\":0.5,\"reason\":\"...\",\"suggestion\":\"...\"} You NEVER decide PASS/FAIL. JSON only."
+
+
+class Agent3Diagnostic:
+    def __init__(self, llm: LLMClient):
+        self.llm = llm
+
+    def analyze(self, fr: FieldResult) -> str:
+        p = f"Field: {fr.field_name}\nCOBOL: {fr.cobol_value}\nJava: {fr.java_value}\nStatus: {fr.status}"
+        return self.llm.call([{"role": "system", "content": P3}, {"role": "user", "content": p}])

agents/llm.py

@@ -0,0 +1,41 @@
+import json, hashlib, os
+from pathlib import Path
+import httpx
+
+
+class LLMClient:
+    def __init__(self, model="gpt-4o-mini", timeout=15, cache_dir=".cache/llm"):
+        self.model = model
+        self.timeout = timeout
+        self.dir = Path(cache_dir)
+        self.dir.mkdir(parents=True, exist_ok=True)
+
+    def _key(self, msgs):
+        return hashlib.sha256(json.dumps(msgs, sort_keys=True).encode()).hexdigest()
+
+    def _get(self, k):
+        p = self.dir / f"{k}.json"
+        return json.loads(p.read_text())["response"] if p.exists() else None
+
+    def _set(self, k, v):
+        (self.dir / f"{k}.json").write_text(json.dumps({"response": v}))
+
+    def call(self, messages, retries=1):
+        k = self._key(messages)
+        c = self._get(k)
+        if c:
+            return c
+        key = os.environ.get("LLM_API_KEY", os.environ.get("OPENAI_API_KEY", ""))
+        base = os.environ.get("LLM_API_BASE", "https://api.openai.com/v1")
+        for a in range(retries + 1):
+            try:
+                r = httpx.post(f"{base}/chat/completions", json={"model": self.model, "messages": messages},
+                               headers={"Authorization": f"Bearer {key}"}, timeout=self.timeout)
+                r.raise_for_status()
+                v = r.json()["choices"][0]["message"]["content"]
+                self._set(k, v)
+                return v
+            except Exception:
+                if a == retries:
+                    raise
+        return ""