nopcoder/cognitive_stack.py at main · dotpipe/nopcoder · GitHub

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
class CognitiveStack:
    """
    Turns agents + LLM society into structured intelligence.
    """

    def __init__(self, llm_society, memory):

        self.llm_society = llm_society
        self.memory = memory

    # =====================================================
    # PLAN
    # =====================================================
    def plan(self, context):

        team = self.llm_society.form_team("planning")

        prompt = f"""
You are a PLANNER.

Goal: {context["goal"]}
State: {context["state"]}

Break this into steps.
Return structured plan.
"""

        outputs = [m.generate(prompt) for m in team]

        return self._merge(outputs)

    # =====================================================
    # CRITIQUE
    # =====================================================
    def critique(self, plan):

        team = self.llm_society.form_team("critique")

        prompt = f"""
You are a CRITIC.

Find flaws in this plan:
{plan}

Return only risks + failures.
"""

        outputs = [m.generate(prompt) for m in team]

        return self._merge(outputs)

    # =====================================================
    # EXECUTE
    # =====================================================
    def execute(self, plan, executor):

        return executor.run(plan)

    # =====================================================
    # REFLECT
    # =====================================================
    def reflect(self, experience):

        team = self.llm_society.form_team("reflection")

        prompt = f"""
You are a REFLECTOR.

Extract reusable rules from:

{experience}
"""

        outputs = [m.generate(prompt) for m in team]

        summary = self._merge(outputs)

        self.memory.store(summary)

        return summary

    # =====================================================
    # SIMPLE MERGE
    # =====================================================
    def _merge(self, outputs):

        return max(outputs, key=len)