Agenaty ukierunkowane na cel poza inżynierią promptów

To przewodnik pokazuje, jak odtworzyć proces od surowców do działającego systemu w ramach: Architektur Agentowych — Artykuł 14: Poza promptami — Projektowanie pętli agentów ukierunkowanych na cel. Skupia się on na krokach operacyjnych, wyraźnych sprawdzeniach oraz kodzie, który można bez żadnych domysłów wdrożyć do repozytorium. Aby uzyskać ogólny zarys, należy najpierw określić dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać oddzielnie od kodu aplikacji. Pliki środowiskowe, magazyny tajemnic oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności przeglądania całej struktury.

Co tutaj znajdziesz

Aby dowiedzieć się, co tutaj znajdziesz, zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później. Punkt kontrolny po kosztownych krokach. System nie powinien ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.

Jednostką jest pętla, a nie prompt

Ponieważ jednostką jest pętla, a nie prompt, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy jakiś krok zawiedzie, powinno to wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną strukturę przepływu. Punkt kontrolny po kosztownych krokach. System powinien unikać ponownego naliczania opłat za tę samą operację LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.

PROMPT ENGINEERING              LOOP ENGINEERING
  ------------------              ----------------
  input --> [model] --> output    goal
                                    |
                                    v
                          +------------------+
                          |  observe state   |
                          +------------------+
                                    |
                                    v
                          +------------------+
                          |  decide next     | <-- this is where
                          |  action          |     goal-direction lives
                          +------------------+
                                    |
                                    v
                          +------------------+
                          |  act             |
                          +------------------+
                                    |
                                    v
                          +------------------+
                          |  goal met?       |--no--> loop back
                          +------------------+
                                    | yes
                                    v
                                  done

Trzy wzory pętli

Dla trzech wzorów pętli należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij poszczególne elementy, zdefiniuj kryteria sukcesu i odrzuć przypadkowe, częściowe ukończenie zadań. Punkt kontrolny po kosztownych krokach – narzędzie do kontynuacji pracy nie powinno ponownie naliczać opłat za tę samą operację LLM, gdy operator próbuje ponownie wykonać późniejszy element procesu. Dla trzech wzorów pętli należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny poufnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całej struktury aplikacji.

+------------------+------------------------------------+---------------------------+
| Pattern          | Loop Structure                     | Best For                  |
+------------------+------------------------------------+---------------------------+
| ReAct            | Thought -> Action -> Observation   | Exploratory tasks where   |
|                  | repeated until goal met            | the path is not known in  |
|                  | Interleaves reasoning and acting   | advance                   |
+------------------+------------------------------------+---------------------------+
| Plan-and-Execute | Plan the full sequence first,      | Tasks with a clear        |
|                  | then execute steps                 | structure that can be     |
|                  | Replan only if execution fails     | decomposed upfront        |
+------------------+------------------------------------+---------------------------+
| Reflexion        | Act, then reflect on the outcome,  | Tasks where quality       |
|                  | store the reflection, retry with   | improves with iteration   |
|                  | that lesson in context             | and self-critique         |
+------------------+------------------------------------+---------------------------+

ReAct

Dla ReAct należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrola przez ludzi oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później. Należy utrzymywać stan grafu w formie prostych, sprecyzowanych typów danych. Wложone struktury danych ukrywają informację o tym, który węzeł zapisał dane w danym polu, co utrudnia kontynuację działania po przerwach.

# harness/loops/react.py
from typing import TypedDict, Annotated, List, Optional
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, AIMessage
from langchain_aws import ChatBedrock
import boto3
class ReActState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]
    goal: dict                    # structured goal representation
    step_count: int
    max_steps: int
    goal_met: bool
    scratchpad: List[dict]        # thought-action-observation history

REACT_SYSTEM_PROMPT = """
You are an agent that solves tasks by interleaving reasoning and action.
For each step:
1. THOUGHT: Reason about the current state and what to do next
2. ACTION: Take one concrete action (call a tool) OR declare the goal met
3. You will then receive an OBSERVATION and repeat
Your goal is provided as a structured object. Check your progress against it
at every step. Do not take actions that do not advance the goal.
When you believe the goal is met, respond with:
GOAL_MET: <brief explanation of how the goal was achieved>
Do not declare the goal met prematurely. Verify against the success criteria.
"""

def build_react_loop(region: str = "us-east-1"):
    bedrock = boto3.client("bedrock-runtime", region_name=region)
    model = ChatBedrock(
        client=bedrock,
        model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
        model_kwargs={"temperature": 0, "max_tokens": 4000},
    )
    def reason_and_act(state: ReActState) -> ReActState:
        goal_desc = format_goal(state["goal"])
        scratchpad_text = format_scratchpad(state["scratchpad"])
        prompt = f"""
GOAL:
{goal_desc}
PROGRESS SO FAR:
{scratchpad_text or "No steps taken yet."}
What is your next THOUGHT and ACTION? Or is the goal met?
"""
        response = model.invoke([
            SystemMessage(content=REACT_SYSTEM_PROMPT),
            HumanMessage(content=prompt)
        ])
        content = response.content
        if "GOAL_MET:" in content:
            return {
                **state,
                "messages": state["messages"] + [response],
                "goal_met": True,
            }
        # Parse thought and action, add to scratchpad
        new_scratchpad = list(state["scratchpad"])
        new_scratchpad.append({
            "step": state["step_count"] + 1,
            "thought_action": content,
        })
        return {
            **state,
            "messages": state["messages"] + [response],
            "step_count": state["step_count"] + 1,
            "scratchpad": new_scratchpad,
        }
    def should_continue(state: ReActState) -> str:
        if state.get("goal_met"):
            return "done"
        if state["step_count"] >= state["max_steps"]:
            return "max_steps"       # safety limit reached
        return "continue"
    graph = StateGraph(ReActState)
    graph.add_node("reason_act", reason_and_act)
    graph.add_node("tools", execute_tools_node)  # from Article 5 tool layer
    graph.set_entry_point("reason_act")
    graph.add_conditional_edges(
        "reason_act",
        should_continue,
        {"continue": "tools", "done": END, "max_steps": END}
    )
    graph.add_edge("tools", "reason_act")
    return graph.compile()

def format_goal(goal: dict) -> str:
    lines = [f"Objective: {goal.get('objective', '')}"]
    if goal.get("success_criteria"):
        lines.append("Success criteria:")
        for c in goal["success_criteria"]:
            lines.append(f"  - {c}")
    if goal.get("constraints"):
        lines.append("Constraints:")
        for c in goal["constraints"]:
            lines.append(f"  - {c}")
    return "\n".join(lines)

def format_scratchpad(scratchpad: List[dict]) -> str:
    return "\n\n".join([
        f"Step {s['step']}:\n{s['thought_action']}"
        for s in scratchpad
    ])

Plan-and-Execute

Dla podejścia Plan-and-Execute należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy dany krok zawiedzie, powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Stan grafu powinien być uporządkowany i typizowany. Wtórne struktury danych ukrywają informację o tym, który węzeł zapisał dane do którego pola, co utrudnia kontynuację pracy po przerwach.

# harness/loops/plan_execute.py
from typing import TypedDict, List, Optional
from langgraph.graph import StateGraph, END, START
from langchain_aws import ChatBedrock
from langchain_core.messages import SystemMessage, HumanMessage
import boto3
import json
class PlanExecuteState(TypedDict):
    goal: dict
    plan: List[dict]              # ordered list of steps
    current_step_index: int
    completed_steps: List[dict]
    needs_replan: bool
    goal_met: bool

PLANNER_PROMPT = """
You are a planning agent. Given a goal, produce a complete, ordered plan
of concrete steps that will achieve it.
Each step must be:
- Specific and actionable
- Independently executable
- Clearly tied to the goal
Return JSON:
{
  "steps": [
    {"step_id": "1", "action": "description", "tool": "tool_name", "expected_result": "..."},
    ...
  ]
}
Do not create steps that do not advance the goal. Prefer fewer, larger steps
over many tiny ones. Plan for verification: include a final step that checks
the goal was met.
"""

def build_plan_execute_loop(region: str = "us-east-1"):
    bedrock = boto3.client("bedrock-runtime", region_name=region)
    # Planner uses the heavy model: planning is where reasoning matters most
    planner = ChatBedrock(
        client=bedrock,
        model_id="anthropic.claude-3-7-sonnet-20250219-v1:0",
        model_kwargs={
            "temperature": 0.2,
            "max_tokens": 4000,
            "thinking": {"type": "enabled", "budget_tokens": 4000},
        },
    )
    # Executor uses the fast model: execution follows the plan
    executor = ChatBedrock(
        client=bedrock,
        model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
        model_kwargs={"temperature": 0, "max_tokens": 4000},
    )
    def plan_node(state: PlanExecuteState) -> PlanExecuteState:
        goal_desc = format_goal(state["goal"])
        response = planner.invoke([
            SystemMessage(content=PLANNER_PROMPT),
            HumanMessage(content=f"Goal:\n{goal_desc}\n\nProduce a plan.")
        ])
        try:
            plan_data = json.loads(response.content)
            steps = plan_data.get("steps", [])
        except json.JSONDecodeError:
            steps = []
        return {
            **state,
            "plan": steps,
            "current_step_index": 0,
            "needs_replan": False,
        }
    def execute_node(state: PlanExecuteState) -> PlanExecuteState:
        idx = state["current_step_index"]
        if idx >= len(state["plan"]):
            return {**state, "goal_met": True}
        current_step = state["plan"][idx]
        response = executor.invoke([
            SystemMessage(content="You execute one step of a plan. Focus only on this step."),
            HumanMessage(content=f"Execute: {current_step['action']}")
        ])
        # Determine if the step succeeded (simplified)
        step_result = {
            "step": current_step,
            "result": response.content,
            "succeeded": "error" not in response.content.lower(),
        }
        completed = list(state["completed_steps"])
        completed.append(step_result)
        # If a step failed, flag for replan
        if not step_result["succeeded"]:
            return {**state, "completed_steps": completed, "needs_replan": True}
        return {
            **state,
            "completed_steps": completed,
            "current_step_index": idx + 1,
        }
    def route_after_execute(state: PlanExecuteState) -> str:
        if state.get("needs_replan"):
            return "replan"
        if state["current_step_index"] >= len(state["plan"]):
            return "done"
        return "execute"
    graph = StateGraph(PlanExecuteState)
    graph.add_node("plan", plan_node)
    graph.add_node("execute", execute_node)
    graph.set_entry_point("plan")
    graph.add_edge("plan", "execute")
    graph.add_conditional_edges(
        "execute",
        route_after_execute,
        {"execute": "execute", "replan": "plan", "done": END}
    )
    return graph.compile()

Rozważania

Dla refleksji: zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nadaj nazwy poszczególnym elementom, zdefiniuj kryteria sukcesu i odrzucaj ciche, częściowe ukończenie zadań. Utrzymuj stan grafu w prostej formie i z określonym typem danych. Wplecione elementy ukrywają informację o tym, który węzeł zapisał dane do którego pola, co utrudnia kontynuację pracy po przerwach. Dla refleksji: zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Utrzymuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności czytania całego grafu.

# harness/loops/reflexion.py
from typing import TypedDict, List, Optional
from langgraph.graph import StateGraph, END
from langchain_aws import ChatBedrock
from langchain_core.messages import SystemMessage, HumanMessage
import boto3
class ReflexionState(TypedDict):
    goal: dict
    attempt: str                  # current solution attempt
    reflections: List[str]        # accumulated lessons from failed attempts
    attempt_count: int
    max_attempts: int
    goal_met: bool

REFLECTION_PROMPT = """
You just attempted to achieve a goal and it did not fully succeed.
Reflect on what went wrong. Be specific and actionable.
Your reflection should identify:
1. What specifically failed or fell short
2. Why it failed (the root cause, not the symptom)
3. What you will do differently next time
This reflection will be available to you on your next attempt.
Make it useful. One or two sentences of genuine insight beats a paragraph
of vague self-criticism.
"""

def build_reflexion_loop(region: str = "us-east-1"):
    bedrock = boto3.client("bedrock-runtime", region_name=region)
    actor = ChatBedrock(
        client=bedrock,
        model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
        model_kwargs={"temperature": 0.3, "max_tokens": 4000},
    )
    evaluator = ChatBedrock(
        client=bedrock,
        model_id="anthropic.claude-3-7-sonnet-20250219-v1:0",
        model_kwargs={"temperature": 0, "max_tokens": 2000},
    )
    def act_node(state: ReflexionState) -> ReflexionState:
        goal_desc = format_goal(state["goal"])
        reflections_text = ""
        if state["reflections"]:
            reflections_text = "\n\nLessons from previous attempts:\n" + \
                "\n".join(f"- {r}" for r in state["reflections"])
        response = actor.invoke([
            SystemMessage(content="You solve tasks. Apply lessons from past attempts."),
            HumanMessage(content=f"Goal:\n{goal_desc}{reflections_text}\n\nProduce your solution.")
        ])
        return {
            **state,
            "attempt": response.content,
            "attempt_count": state["attempt_count"] + 1,
        }
    def evaluate_node(state: ReflexionState) -> ReflexionState:
        goal_desc = format_goal(state["goal"])
        response = evaluator.invoke([
            SystemMessage(content="Evaluate whether the solution meets the goal. Respond GOAL_MET or GOAL_NOT_MET with a brief reason."),
            HumanMessage(content=f"Goal:\n{goal_desc}\n\nSolution:\n{state['attempt']}")
        ])
        if "GOAL_MET" in response.content and "NOT_MET" not in response.content:
            return {**state, "goal_met": True}
        # Generate a reflection for the next attempt
        reflection_response = actor.invoke([
            SystemMessage(content=REFLECTION_PROMPT),
            HumanMessage(content=f"Goal:\n{goal_desc}\n\nYour attempt:\n{state['attempt']}\n\nEvaluation:\n{response.content}")
        ])
        new_reflections = list(state["reflections"])
        new_reflections.append(reflection_response.content)
        return {**state, "reflections": new_reflections, "goal_met": False}
    def should_retry(state: ReflexionState) -> str:
        if state.get("goal_met"):
            return "done"
        if state["attempt_count"] >= state["max_attempts"]:
            return "exhausted"
        return "retry"
    graph = StateGraph(ReflexionState)
    graph.add_node("act", act_node)
    graph.add_node("evaluate", evaluate_node)
    graph.set_entry_point("act")
    graph.add_edge("act", "evaluate")
    graph.add_conditional_edges(
        "evaluate",
        should_retry,
        {"retry": "act", "done": END, "exhausted": END}
    )
    return graph.compile()

Reprezentacja celu

Aby przedstawić cel, zdefiniuj dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Zdokumentuj zarówno ścieżkę prawidłowego działania, jak i ścieżkę naprawczą. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później. Wymagaj ludzkiej aprobaty w przypadkach, gdy dochodzi do wydawania pieniędzy lub zmiany danych produkcyjnych. Podłączenia realizowane w czasie kompilacji nie równają się kompletności rozwiązania biznesowego.

# harness/loops/goal.py
from dataclasses import dataclass, field
from typing import List, Optional, Callable
from enum import Enum
import time
class GoalStatus(Enum):
    ACTIVE = "active"
    ACHIEVED = "achieved"
    FAILED = "failed"
    REVISED = "revised"
    ABANDONED = "abandoned"

@dataclass
class Goal:
    """
    A structured goal that an agent loop pursues.
    Distinct from the initial prompt: this is what the agent
    checks progress against at every step.
    """
    goal_id: str
    objective: str                          # what to achieve
    success_criteria: List[str]             # how to know it is done
    constraints: List[str]                  # boundaries that must hold
    priority: int = 1
    parent_goal_id: Optional[str] = None    # for sub-goals
    status: GoalStatus = GoalStatus.ACTIVE
    created_at: float = field(default_factory=time.time)
    progress_notes: List[str] = field(default_factory=list)
    def to_prompt_block(self) -> str:
        """Renders the goal for injection into agent context."""
        lines = [f"OBJECTIVE: {self.objective}", ""]
        lines.append("SUCCESS CRITERIA (all must be met):")
        for i, c in enumerate(self.success_criteria, 1):
            lines.append(f"  {i}. {c}")
        if self.constraints:
            lines.append("\nCONSTRAINTS (must not be violated):")
            for c in self.constraints:
                lines.append(f"  - {c}")
        if self.progress_notes:
            lines.append("\nPROGRESS SO FAR:")
            for note in self.progress_notes[-3:]:  # last 3 notes
                lines.append(f"  - {note}")
        return "\n".join(lines)
    def record_progress(self, note: str):
        self.progress_notes.append(note)

class GoalTracker:
    """
    Manages the goal hierarchy for an agent run.
    Supports sub-goals, revision, and progress assessment.
    """
    def __init__(self):
        self.goals: dict = {}
        self.root_goal_id: Optional[str] = None
    def set_root_goal(self, goal: Goal):
        self.goals[goal.goal_id] = goal
        self.root_goal_id = goal.goal_id
    def add_subgoal(self, parent_id: str, subgoal: Goal):
        subgoal.parent_goal_id = parent_id
        self.goals[subgoal.goal_id] = subgoal
    def assess_progress(
        self,
        goal_id: str,
        current_state: str,
        assessor_model,
    ) -> dict:
        """
        Uses a model to assess how much progress has been made
        toward a goal's success criteria.
        """
        goal = self.goals.get(goal_id)
        if not goal:
            return {"error": "goal not found"}
        from langchain_core.messages import SystemMessage, HumanMessage
        import json
        response = assessor_model.invoke([
            SystemMessage(content="""
Assess progress toward a goal. For each success criterion, determine
whether it is met, partially met, or not met.
Return JSON:
{
  "criteria_status": [{"criterion": "...", "status": "met|partial|not_met"}],
  "overall_progress": 0.0 to 1.0,
  "goal_achieved": true | false,
  "blocking_issues": ["..."]
}
"""),
            HumanMessage(content=f"Goal:\n{goal.to_prompt_block()}\n\nCurrent state:\n{current_state}")
        ])
        try:
            return json.loads(response.content)
        except json.JSONDecodeError:
            return {"overall_progress": 0.0, "goal_achieved": False}

Iniekcja i modyfikacja celu w czasie wykonywania

Dla celów iniekcji i modyfikacji w czasie wykonywania programu należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed zmianą kodu. Operatorzy powinni móc ponownie uruchomić dany krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, łatwe do przetestowania jednostki zamiast rozbudowanych skryptów. Gdy dany krok zawiedzie, powód awarii powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowaną strukturę procesów. Konieczne jest ludzkie zatwierdzenie w przypadkach, gdy dochodzi do wydawania pieniędzy lub modyfikacji danych produkcyjnych. Podłączenia realizowane w czasie kompilacji nie równają się kompletności rozwiązania biznesowego.

# harness/loops/goal_revision.py
import boto3
import time
import json
from typing import Optional
from harness.loops.goal import Goal, GoalStatus, GoalTracker
class RuntimeGoalManager:
    """
    Handles goal injection and revision during a running loop.
    Supports the /goal directive: a user or supervisor can modify
    the agent's goal mid-run without restarting.
    """
    def __init__(
        self,
        tracker: GoalTracker,
        region: str = "us-east-1",
    ):
        self.tracker = tracker
        bedrock = boto3.client("bedrock-runtime", region_name=region)
        from langchain_aws import ChatBedrock
        self.reconciler = ChatBedrock(
            client=bedrock,
            model_id="anthropic.claude-3-7-sonnet-20250219-v1:0",
            model_kwargs={"temperature": 0.2, "max_tokens": 3000},
        )
    def inject_goal_revision(
        self,
        current_goal_id: str,
        revision_type: str,      # "add_constraint" | "modify_objective" | "redirect"
        revision_content: str,
        completed_work: str,
    ) -> dict:
        """
        Injects a goal revision into a running loop.
        Returns a reconciliation that tells the agent what work is still valid.
        """
        current_goal = self.tracker.goals.get(current_goal_id)
        if not current_goal:
            return {"error": "goal not found"}
        from langchain_core.messages import SystemMessage, HumanMessage
        response = self.reconciler.invoke([
            SystemMessage(content="""
The agent's goal has been revised mid-run. Your job is to reconcile:
1. Determine what completed work is still valid under the new goal
2. Determine what completed work must be discarded or redone
3. Determine what new work the revision requires
Do NOT restart from scratch unless the revision fundamentally changes
the objective. Preserve as much valid work as possible.
Return JSON:
{
  "work_still_valid": ["..."],
  "work_to_discard": ["..."],
  "new_work_required": ["..."],
  "requires_full_restart": true | false,
  "reconciliation_note": "brief explanation for the agent"
}
"""),
            HumanMessage(content=f"""
Original goal:
{current_goal.to_prompt_block()}
Revision type: {revision_type}
Revision: {revision_content}
Work completed so far:
{completed_work}
""")
        ])
        try:
            reconciliation = json.loads(response.content)
        except json.JSONDecodeError:
            reconciliation = {"requires_full_restart": False, "reconciliation_note": revision_content}
        # Apply the revision to the goal object
        if revision_type == "add_constraint":
            current_goal.constraints.append(revision_content)
        elif revision_type == "modify_objective":
            current_goal.objective = revision_content
            current_goal.status = GoalStatus.REVISED
        elif revision_type == "redirect":
            current_goal.objective = revision_content
            current_goal.success_criteria = []  # will be re-derived
            current_goal.status = GoalStatus.REVISED
        current_goal.record_progress(
            f"Goal revised ({revision_type}): {reconciliation.get('reconciliation_note', '')}"
        )
        return reconciliation

# Example of the /goal directive as a user-facing command
def handle_goal_directive(
    directive: str,
    goal_manager: RuntimeGoalManager,
    current_goal_id: str,
    completed_work: str,
) -> dict:
    """
    Parses a /goal directive from a user and applies it to the running loop.
    Directive formats:
      /goal add-constraint <text>   - add a constraint to the current goal
      /goal redirect <text>          - change the objective
      /goal status                   - report current goal and progress
    """
    parts = directive.strip().split(maxsplit=2)
    if len(parts) < 2 or parts[0] != "/goal":
        return {"error": "invalid directive format"}
    command = parts[1]
    if command == "status":
        goal = goal_manager.tracker.goals.get(current_goal_id)
        return {"goal_status": goal.to_prompt_block() if goal else "no active goal"}
    if command == "add-constraint" and len(parts) == 3:
        return goal_manager.inject_goal_revision(
            current_goal_id, "add_constraint", parts[2], completed_work
        )
    if command == "redirect" and len(parts) == 3:
        return goal_manager.inject_goal_revision(
            current_goal_id, "redirect", parts[2], completed_work
        )
    return {"error": f"unknown goal command: {command}"}

Wybieranie i łączenie wzorców pętli

Aby wybrać i połączyć wzory pętli, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij poszczególne elementy, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadań. Zapewnij ludzką aprobatę w przypadku operacji, które wiążą się z wydawaniem pieniędzy lub modyfikacją danych produkcyjnych. Podłączenia realizowane w czasie kompilacji nie równają się pełnej kompletności biznesowej. Aby wybrać i połączyć wzory pętli, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić ten krok od znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Przechowuj konfigurację poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić, nie musząc czytać całego kodu.

grafu.

+---------------------------------+--------------------------------------+
| Situation                       | Loop Choice                          |
+---------------------------------+--------------------------------------+
| Path to goal is unknown         | ReAct: figure out each step as you go|
+---------------------------------+--------------------------------------+
| Task has clear structure        | Plan-and-Execute: plan then run      |
+---------------------------------+--------------------------------------+
| Quality improves with iteration | Reflexion: act, reflect, retry       |
+---------------------------------+--------------------------------------+
| Complex task, mixed sub-tasks   | Plan-and-Execute at top,             |
|                                 | ReAct or Reflexion per step          |
+---------------------------------+--------------------------------------+
| High stakes, needs verification | Any loop + Article 5 verification    |
|                                 | + Article 13 approval gate           |
+---------------------------------+--------------------------------------+

# harness/loops/loop_selector.py
from enum import Enum
from typing import Optional
class LoopPattern(Enum):
    REACT = "react"
    PLAN_EXECUTE = "plan_execute"
    REFLEXION = "reflexion"

def select_loop_pattern(
    task_complexity: str,           # from Article 10 complexity classifier
    has_procedure: bool,            # from Article 7 procedural memory
    is_exploratory: bool,
    quality_critical: bool,
) -> LoopPattern:
    """
    Selects the loop pattern based on task characteristics.
    Connects loop engineering to the memory and cost systems built earlier.
    """
    # If we have a known-good procedure, plan-and-execute with it
    if has_procedure:
        return LoopPattern.PLAN_EXECUTE
    # Exploratory tasks with unknown paths suit ReAct
    if is_exploratory:
        return LoopPattern.REACT
    # Quality-critical tasks benefit from iterative refinement
    if quality_critical:
        return LoopPattern.REFLEXION
    # Structured, complex tasks: plan first
    if task_complexity in ("COMPLEX", "CRITICAL"):
        return LoopPattern.PLAN_EXECUTE
    # Default: ReAct is the most flexible
    return LoopPattern.REACT

Kontrola rzeczywistości w produkcji

Aby przeprowadzić kontrolę rzeczywistości w produkcji, należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy udokumentować zarówno prawidłowy przebieg procesu, jak i ścieżkę naprawczą. Próby ponownych działań, kontrolne punkty ludzkie oraz obsługa wiadomości błędnych stanowią część produktu, a nie elementy dodawane później. Punkt kontrolny po kosztownych krokach. System powinien unikać ponownego naliczania opłat za tę samą operację LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.

Architektura referencyjna

Dla architektury referencyjnej należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Należy preferować małe, testowalne jednostki zamiast rozbudowanych skryptów. Gdy dany krok zawiedzie, powinien wskazywać na konkretną odpowiedzialność, a nie na skomplikowany łańcuch operacji. Punkt kontrolny po kosztownych krokach. Funkcja wznowienia nie powinna ponownie naliczać opłat za tę samą wywołanie LLM, gdy operator próbuje ponownie uruchomić późniejszy węzeł.

Task arrives with a goal
          |
          v
+---------------------------+
|   Goal Structuring        |
|   objective + criteria +  |
|   constraints (Goal obj)  |
+---------------------------+
          |
          v
+---------------------------+
|   Loop Pattern Selector   |
|   has_procedure? -> P&E   |
|   exploratory?   -> ReAct |
|   quality-crit?  -> Reflex|
+---------------------------+
          |
    +-----+-----+-----------+
    |           |           |
    v           v           v
+--------+  +--------+  +----------+
| ReAct  |  | Plan & |  |Reflexion |
| loop   |  |Execute |  | loop     |
+--------+  +--------+  +----------+
    |           |           |
    +-----+-----+-----------+
          |
          v
+---------------------------+
|   Each step runs through: |
|   - Harness (Article 5)   |
|   - Tools (auth/retry/cb) |
|   - Memory (Article 7)    |
+---------------------------+
          |
          v
+---------------------------+
|   Goal Progress Assessment|
|   goal met? revise? stop? |
+---------------------------+
          |
   +------+------+
   |             |
   v             v
Runtime      Goal met -->  Verification (Art 5)
goal         + eval (Art 8)
revision     + approval if needed (Art 13)
(/goal)

Stack infrastruktury referencyjnej

Dla stosu infrastruktury referencyjnej należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Traktuj tę fazę jako umowę pomiędzy danymi wejściowymi a zweryfikowanymi wynikami. Nazwij poszczególne elementy, zdefiniuj kryteria sukcesu i odrzuć ciche, częściowe ukończenie zadania. Punkt kontrolny po kosztownych krokach. System powinien unikać ponownego naliczania opłat za tę samą operację LLM, gdy operator próbuje ponownie wykonać późniejszy węzeł.

+-----------------------------+---------------------+------------------------------+
| Component                   | Technology          | Role                         |
+-----------------------------+---------------------+------------------------------+
| ReAct Loop                  | LangGraph +         | Interleaved reason-act for   |
|                             | Claude 3.5          | exploratory tasks            |
+-----------------------------+---------------------+------------------------------+
| Plan-and-Execute Loop       | LangGraph +         | Plan with heavy model,       |
|                             | Claude 3.7 / 3.5    | execute with fast model      |
+-----------------------------+---------------------+------------------------------+
| Reflexion Loop              | LangGraph +         | Act-reflect-retry for        |
|                             | Claude 3.5 / 3.7    | iterative quality            |
+-----------------------------+---------------------+------------------------------+
| Goal Representation         | Structured Goal     | Objective, criteria,         |
|                             | dataclass           | constraints, progress        |
+-----------------------------+---------------------+------------------------------+
| Goal Tracker                | In-memory + state   | Sub-goal hierarchy,          |
|                             |                     | progress assessment          |
+-----------------------------+---------------------+------------------------------+
| Runtime Goal Revision       | Custom + Claude 3.7 | /goal directive, mid-run     |
|                             | reconciler          | goal injection and merge     |
+-----------------------------+---------------------+------------------------------+
| Loop Pattern Selector       | Custom classifier   | Chooses loop per task,       |
|                             |                     | connects to Article 7 memory |
+-----------------------------+---------------------+------------------------------+
| Step-Level Harness          | Article 5           | Shapes model behavior        |
|                             |                     | within each loop step        |
+-----------------------------+---------------------+------------------------------+
| Goal Revision Security      | Article 9 controls  | Treats external goal changes |
|                             |                     | as data, gates high-privilege|
+-----------------------------+---------------------+------------------------------+
| Loop Safety Limits          | max_steps /         | Prevents infinite loops,     |
|                             | max_attempts        | ties to Article 5 loop detect|
+-----------------------------+---------------------+------------------------------+
| Local Dev Alternative       | Ollama + LangGraph  | All three loops testable     |
|                             | Docker Compose      | without Bedrock costs        |
+-----------------------------+---------------------+------------------------------+

Dla stosu infrastruktury referencyjnej należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu. Konfigurację należy przechowywać oddzielnie od kodu aplikacji. Pliki środowiskowe, magazyny tajemnic oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności przeglądania całej struktury.

Lista kontrolna operacyjna

Dla listy kontrolnej operacyjnej należy zdefiniować dane wejściowe, osobę odpowiedzialną za dany krok oraz kryteria zakończenia przed modyfikacją kodu. Operatorzy powinni móc ponownie uruchomić dany krok na podstawie znanego punktu kontrolnego, bez konieczności zgadywania ukrytego stanu.

Należy rejestrować czasy wykonywania oraz koszt tokenów lub zapytań obok wyników funkcjonalnych. Wczesna widoczność kosztów zapobiega niespodziewanym rachunkom, gdy ścieżka przechodzi z środowiska demonstracyjnego do wspólnych środowisk.

Zachowuj prostą i typowaną strukturę stanu grafu. Wkładki nawarstwione ukrywają informację o tym, który węzeł zapisał dane do którego pola, i utrudniają kontynuację pracy po przerwach.

Gdy budżet na to pozwala, dodaj test sprawdzający kluczową ścieżkę w procesie CI przy użyciu fixitów, a nie rzeczywistych, płatnych API.

Konfigurację trzymaj poza kodem aplikacji. Pliki środowiskowe, magazyny tajnych danych oraz flagi funkcjonalne powinny znajdować się w jednym miejscu, które operatorzy mogą sprawdzić bez konieczności przeglądania całego grafu.

Zachowuj prostą i typowaną strukturę stanu grafu. Wkładki nawarstwione ukrywają informację o tym, który węzeł zapisał dane do którego pola, i utrudniają kontynuację pracy po przerwach.

Zanim wdrożysz nową architekturę, zamroź wersje oprogramowania, utwórz dokładny zapis działania kluczowej ścieżki oraz potwierdź kroki konieczne do cofnięcia zmian. Środowiska współdzielone wymagają ograniczeń szybkości, weryfikacji użytkowników oraz wyraźnego właściciela odpowiedzialnego za rotację tajnych danych. Wolisz nudną niezawodność od pomysłowych, jednorazowych demonstracji.

Literatura pokrewna

• Poza liniowymi skryptami: węzły, krawędzie LangGraph oraz trwały stan — Zastąp kruche pętle pracami grafowymi: specjalistycznymi węzłami, wyraźnymi schematami stanu, rozgałęzieniami, mechanizmami kontroli ludzkiej oraz ścieżkami odpornościowymi.
• Config, State, Store, Prompt: inżynieria kontekstu w LangGraph — Oddziel niezmienne ustawienia czasu wykonywania, zmienny stan roboczy, długoterminową bazę danych oraz przygotowane prompty, dzięki czemu agenci są tańsze, bezpieczniejsze i łatwiejsze do debugowania.
• Warunkowe przepływy pracy w LangGraph: rozgałęzianie bez chaotycznego łączenia promptów — Zapisz logikę routingu agentów typu if-else jako nazwane krawędzie warunkowe z jasnymi umowami stanu oraz możliwością naprawy błędów.