Агенты, які дзейнуюць за метай, за межамі інжынерыі запросаў
Агенты, які дзейнуюць за певнымі цілям, патрабуюць наявнасці ціляў, планавальнікаў і ціклів зворачнай свярзкі, а не дужа дзьялегахлівых інструкцый. Тры мадэлі ціклів, якія дазволяюць абыцца ад тэхнікі стварэння інструкцый, з LangGraph.
Крыніцы: Agentic Architectures — Article 14: Beyond Prompts — Engineering Goal-Directed Agent Loops
У гэтым керавану практычна адбудова маршруту ад сыр'ёў да рабочай сістэмы для: Агентных архітэктураў — Артыкул 14: За межамі запросаў — Інжынерыя цэлевых агентных цыклаў. Акцэнт ставіцца на практычныя крокі, чыстае перакананне і код, які можна проста дадаць у рэпазітарый без неабязковасці здогадвацца пра намеры. Для агульнага відображэння паказваюцца вхідныя даны, адпаведальная особа за крок і крытэрыя завершэння пры зміне коду. Аперацыйныя працавнікі должны магчымае перадзваніць крок з вядомай точкі контролю без неабязковасці здагадвацца пра схованы стан. Конфігурацыю трэба залічыць параду ад коду прыкладнага програма. Файлы сераўіса, хранільнікі секрэтных дадзеных і флагі функцый должны знаходзіцца ў аднам месцы, якое працавнікі можу аўдытаваць, не чытаючы весь граф.
Што вы знайдзеце тут
Ёнколі вы хочаце знайсці тут, перад тым як зменіць код, неабяжна адзначыць вхідныя даны, власніка крока і крытэрыі завершэння. Аператары должны магчымаць перзапуск крока з вядомай точкі контролю, не падозрываючы прыхованы стан. Неабяжна задокументаваць як шлях успеху, так і шлях вярнення. Перапрыбуткі, людзкіе перакрыцця і апрацоўка некорэктных паведамленняў ёсць часткай продукту, а не яго пазнейшай дапрацоўкі. Точка контролю пасля дорогіх крокаў. Сістэма вярнення не должна занова ставіць плату за той самы вызыв LLM, калі аператар перапрыбуе пазнейшы вузел.
Елементам выканання ёсць цікл, а не запрос
Калі «Цыкл» — гэта адзінак, а не запрос, перад змянайом код неабходна ўзначыць вхідныя даны, адпаведальнага за крок і крытэрыя завершэння. Аператары должны магчымае перадзеісцавляць крок з вядомай точкі контролю, не спрабоўваючы здагадвацца пра схованы стан. Валіць крашчэ маленькія, тэставаныя адзінакі, чым велікія скрыпты. Калі крок не выйшаў, прычына неудачы павінна вказываць на адзіную адпаведальнасць, а не на заплутаны ланцужок задач. Ствараць точку контролю пасля дорогіх крокаў. Функцыя продакцыі не павінна занова ставіць плата за той самы вызыв LLM, калі аператар перадзеісцавляе пазнейшы вузол.
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
Тры шаблоны цыклаў
Для трохіцых шаблонав цыклаў неабходна прадзефініраваць вхідныя даны, адпаведальнага за крок і крэтыяры выходу пры зміне коду. Аператары должны магчымаць перзапуск кроку з вядомай точкі контролю без неабясненняя схованага стану. Спрыятлівайце гэты ўрадок як кантракт межа вхіднымі данымі і перакананымі выходамі. Даўце назвы артыфактам, прадзефініравце перакананняя успеху і адмовіцеся ад бяспрыводнага частковага завершэння. Точка контролю пасля дорогіх крокаў. Сістэма вярнення не должна зноў выклікаць той самы кал дзеяння LLM, калі аператар перапрыначвае пазнейшы вузел. Для трохіцых шаблонав цыклаў неабходна прадзефініраваць вхідныя даны, адпаведальнага за крок і крэтыяры выходу пры зміне коду. Аператары должны магчымаць перзапуск кроку з вядомай точкі контролю без неабясненняя схованага стану. Зберагачыце настройкі за межамі коду прыемленае. Файлы сераў, хранільнікі секрэтных дадзеных і флагі функцый должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць без чытання цэлага графу.
+------------------+------------------------------------+---------------------------+
| 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
Для ReAct неабяжна прадзефінавацыя вхідных дадзеных, адпраўніка крока і крэтарыяў завершэння працы перад зменым коду. Аперацыйныя працавнікі должны магчымае перапрацаваць крок пачынаючы з вядомага пункта контролю, не прымушаныя здогадвацца пра схованы стан. Неабяжна задокументаваць як шлях успеху, так і шлях вярнення да нормальнага стану. Перапрыбуткі, людзкія контрольныя пункты і обработка некорэктных паведамленняў ёсць часткай продукту, а не элементамі, якія дадаюцца пазней. Неабяжна трываліць стан графа ў простам і тыпаваным фармаце. Вярнутыя структуры дадзеных маскуюць інфармацыю пра тое, які вузел запісаў кожнае поле, і спакошваюць возз'еднанне пасля перарываў.
# 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
Для падчынкі Plan-and-Execute неабяжна ўзначыць вхідныя даны, адпаведальнага за крок і крытэрыя завершэння пры зміне коду. Аперацыйныя працавнікі должны магчыма ўвайсці крок з вядомай точкі контролю, не прабуючы здогадвацца пра схованы стан. Лепш выбіраць маленькія, тэставаныя елементы замест большых скрыптов. Калі крок не выконваецца, прычына неудачы должна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцюг задач. Старайцеся, каб стан графа быў простым і мелі тыпаванне. Вкладзеныя структуры дадзеных маскуюць інфармацыю пра тое, які вузел запісаў якое поле, і спакоююць продажчэнне выканання пасля перэрываў.
# 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()
Размышленні
Для размышлення, перш чым зменіць код, неабяжна візначыць вхідныя даны, адміністратара крока і крытэрыя завершэння. Аператары должны магчымаць перзапуск крока з вядомай точкі контролю, не спрабоўваючы здогадвацца пра схованы стан. Спрыяйце таму, каб гэты этап быў схожы на контракт межа вхіднымі данымі і перакананымі выходнымі рэзультатамі. Дайце назвы артыфактам, візначыце крытэрыя успеху і не праграмавайце беззвучна часткова завершэння задачы. Храніце стан графа у простам і тыпаваным фармаце. Вкладзеныя блокі маскуюць інфармацыю пра тое, які вузел запісаў якое поле, і спакошуюць возобнавленне працы пасля перэрываў. Для размышлення, перш чым зменіць код, неабяжна візначыць вхідныя даны, адміністратара крока і крытэрыя завершэння. Аператары должны магчымаць перзапуск крока з вядомай точкі контролю, не спрабоўваючы здагадвацца пра схованы стан. Храніце настройкі паза кодам прыемленае програмы. Файлы сераў, хранільнікі секрэтных даных і флагі функцыйяў должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць, не чытаяўшы весь граф.
# 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()
Адображэнне цялі
Для адаптавання меты неабяжна практычна апраначка: перш чым зменяць код, неабяжна візначыць інпуты, адпаведальнага за крок і крытэрыі завершэння. Аперацыяныя працавальнікі павінны магчымаць перзапуск кроку з вядомай точкі контролю, не падозрываючы прыхованы стан. Неабяжна задокументаваць як шлях успеху, так і шлях вярнення да нормальнага стану. Перапрыбуткі, людзкія перакрыцці і апрацоўка некорэктных паведамленняў ёсць часткай продукту, а не элементамі пазнейшага дапрацоўвання. Неабяжна прыменіць людзкую апраўдку для тых крокаў, якія выкарыстоўваюць грошы або зменяюць даны праўдзівай працы. Кампіляцыйныя наладкі не ёсць адпаведнікамі пачынковай готовасці продукту.
# 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}
Введэнне і перасмотраць меты пад час выканання
Для введэння і змены цялей падчас выканання неабходна адзначэнне вхідных дадзеных, адпаведальнага за крок і крэтарыяў завершэння працы перад зменай коду. Аператары должны магчымае перадзягнуць крок з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Лепш выбіраць маленькія, тэставаныя елементы замест большых скрыптов. Калі крок не выканаецца, прычына неудачы должна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцужок задач. Неабходна людская апраўда для тых крокаў, якія выкарыстоўваюць грошы чы зміняюць даны варабочага прыемлівача. Працэс складання коду не ўзроўнаважваецца з повнасцю бізнес-процэсаў.
# 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}"}
Выбіранне і сумешчанне шаблонаў цыклаў
Для выбору і адыянаціі шаблонаў цыкла неабходна перад змянай коду задаць вхідныя даны, адпаведальнага за крок і крэтыяры выходу. Аператары должны магчымае перайсці канкрэтны крок з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Спрыяйце цэй стадзіі як даговору межа вхіднымі данымі і перакананымі выходнымі рэзультатамі. Даць назвы артыфактам, задаць перакананні на успех і адмовіцца ад беззвучнага частковага завершэння. Забезпечыце людскія падтверджэння для тых крокаў, якія выкарыстоўваюць грошы чысткі зменяюць даны праўернай працы. Кампіляцыйныя налашчэння не ўзроўнаўцуюцься з пачатковай цэласнасцю бізнес-процэсаў. Для выбору і адыянаціі шаблонаў цыкла неабходна перад змянай коду задаць вхідныя даны, адпаведальнага за крок і крэтыяры выходу. Аператары должны магчымае перайсці канкрэтны крок з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Зберагачыце налашчэнні параду коду прыкладнення. Файлы сераўіса, хранільнікі секрэтных дадзеных і флагі функцыйяў должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць, не чытаючы весь код.
Граф.
+---------------------------------+--------------------------------------+
| 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
Перакантовка рэальнасці вырабоцтва
Для перакантовкі рэальнасці вырабоцтва неабяжна пазначыць вхідныя даны, адпаведальнага за крок і крытэрыі завершэння пры змяне коду. Аперацыйныя працавнікі должны магчымаць перапрацоўваць крок з вядомага пункту контролю без неабяжнай адгадванняя схованага стану. Неабяжна задокументаваць як шлях успеху, так і шлях вярнення да нормальнасці. Перапрыбуткі, людзкія контрольныя пункты і обработка некорэктных паведамленняў ёсць часткай продукту, а не элементамі пазнейшага дапрацоўкі. Пункт контролю пасля дорогіх крокаў. Функцыя вярнення не должна занова выклікаць той самы калектар LLM, калі аперацыйны працавнік перапрыбуе праз пазнейшы вузел.
Архітэктура для апавядання
Для архітэктуры нараджэння неабяцкова пазначыць вхідныя даны, адпаведальнага за кожны крок і крытэрыі завершэння пры змяне коду. Аперацыйныя працавнікі должны магчымае рэалізаваць павторную експансію крока з вядомага пункта контролю, не прыпускаючы стану, які застаецца незрозумелым. Лепш выбіраць маленькія, тэставаныя елементы замест большых скрыптов. Калі крок не выйшоў, прычына неудачы должна вказываць на адну конкрэтную адпаведальнасць, а не на заплутаны ланцужок задач. Пункт контролю пасля дорогіх крокаў. Функцыя продакцыі не должна занова ставіць плату за той самы вызыв LLM, калі аперацыйны працавнік прабуе зноў выконаць пазнейшы вузел.
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)
Стак інфраструктуры нараджэння
Для інфраструктурнага стака-праменя, пры змены коду, неабходна дэфініцыя вхідных даных, адпаведнага адпаведальнага за крок і крэтарыяў завершэння. Аперацыйныя працавнікі павінны магчымаецца перзапускаць крок з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Спрыяйце гэтаму этапу як кантракту межа вхіднымі данымі і падтвердзенымі выходнымі рэзультатамі. Даўце назвы артыфактам, задаць крэтарыяў успеху і адмовіцеся ад беззвучнага частковага завершэння. Пункт контролю пасля дорогіх крокаў. Сістэма адновлення не павінна занова сцягваць плата за той самы вызыв LLM, калі аперацыйны працавнік перапрыяўляе пазнейшы вузел.
+-----------------------------+---------------------+------------------------------+
| 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 |
+-----------------------------+---------------------+------------------------------+
Для інфраструктурнага стака-праменя неабяжна ўзначэнне вхідных дадзеных, адпаведальнага за практыку і крэтарыяў завершэння працы перад зменым коду. Аперацыйныя працавнікі должны магчымае перайсці практыку з вядомага контрольнага пункту, не спрабоўваючы здогадвацца пра схованы стан. Конфігурацыю трэба залічваць праза код аплікацыі. Файлы серавэсу, хранальнікі секрэтных дадзеных і флагі функцый належаць у аднам месца, якое працавнікі можу аўдытаваць, не чытаючы весь ланцуг.
Чэк-ліст для аперацый
Для чэк-ліста для аперацый неабяжна ўзначэнне вхідных дадзеных, адпаведальнага за практыку і крэтарыяў завершэння працы перад зменым коду. Аперацыйныя працавнікі должны магчымае перайсці практыку з вядомага контрольнага пункту, не спрабоўваючы здогадвацца пра схованы стан.
Запісваць час выканання і кост токена або запыту разам з функцыйнымі рэзультатамі. Відразлівасць костаў з самага пачатку запобегае неспадзяваным рахункам, калі ланцуг пераходзіць з дэмовай асяроддзя ў спяльнаныя асяроддзя.
Зберагаеце стан графа ў простам і тыпаваным фармаце. Вкладаныя блобы маскуюць інфармацію пра тое, які вузел запісаў якое поле, і спакошуюць продовжэнне працы пасля перэрываў.
Калі дозволяе бюджэт, дадзіце тэст на перакананне, які працюе з крытычным шляхам у CI за дапамою фіксатываў, а не з рэальнымі платнымі API.
Зберагаеце настройкі паза кодам прыемлівача. Файлы сераўнавання, хранільнікі секрэтных дадзеных і флагі функцыйяў должны знаходзіцца ў аднам месцы, куда аператары можуць адбавляць контроль без неабяжнага чытання всего графа.
Зберагаеце стан графа ў простам і тыпаваным фармаце. Вкладаныя блобы маскуюць інфармацію пра тое, які вузел запісаў якое поле, і спакошуюць продовжэнне працы пасля перэрываў.
Перад апраноўкай стэка заморозьце версіі, зафіксуйце ідэальны транскрыпт для крытычнага шляху і паказваце крокі для адворачэння. У спакульнаваных сераўнаваннях неабходны ліміты частоты запытоў, перакананні ў прыналежнасці і чыстае вялоўнік для змены секрэтных дадзеных. Валіце надзейнасць працы над красавім разовым дэманстрацыйным прыкладам.