超越提示工程的目标导向智能体
目标导向的智能体需要目标、规划器以及反馈循环，而非更长的提示语。借助 LangGraph，可实现三种超越提示工程局限的循环模式。
来源: Agentic Architectures — Article 14: Beyond Prompts — Engineering Goal-Directed Agent Loops
本指南将逐步构建从原材料到可运行系统的完整流程，内容涉及：代理架构——第14篇：超越提示词——实现目标导向的代理循环。重点在于可操作的步骤、明确的检查点，以及可直接放入代码库而无需猜测其用途的代码。 在开始修改代码之前，应先明确输入参数、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需推测隐藏状态。 配置信息应与应用程序代码分开存放。环境文件、密钥存储以及功能开关都应集中管理，以便操作人员无需查看整个系统结构即可进行审核。
此处内容概览
对于此处的内容，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的功能。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复功能不应再次计费相同的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
三种循环模式
对于三种循环模式，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，杜绝默许的半完成状态。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。 对于三种循环模式，在修改代码之前需明确输入参数、该步骤的负责人以及终止条件。操作员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将配置信息置于应用程序代码之外。环境文件、密钥存储及功能开关应集中存放于一个操作员可审计的位置，无需查看整个流程结构。
+------------------+------------------------------------+---------------------------+
| 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
])
规划与执行
在“计划与执行”模式中，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 优先选择小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任主体，而非复杂的流程链。 保持图结构的状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会导致中断后无法继续执行。
# 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调用费用。
参考架构
在制定参考架构时，应在修改代码之前明确输入内容、各步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需猜测隐藏状态。相比庞大的脚本，更应采用小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向某个特定职责，而非整个复杂的流程。在成本较高的步骤之后设置检查点；当操作人员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
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)
参考基础设施栈
对于参考架构栈，在修改代码之前需明确输入内容、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行相应步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，设定成功检测标准，并拒绝默许的半完成状态。 在成本较高的步骤之后设置检查点。当操作人员重新尝试后续节点时，恢复流程不应再次计费相同的大型语言模型调用费用。
+-----------------------------+---------------------+------------------------------+
| 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 |
+-----------------------------+---------------------+------------------------------+
对于参考架构栈，在修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。配置信息应置于应用程序代码之外，环境文件、密钥存储以及功能标志都应集中存放于一个位置，这样操作人员无需查看整个系统结构即可进行审计。
运营检查清单
对于运营检查清单，同样需要在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。
应在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境切换到共享环境时出现意外费用。
保持图结构扁平且具有类型约束。嵌套的数据块会隐藏是哪个节点修改了哪个字段，还会在中断后导致无法继续执行。
只要预算允许，就在持续集成过程中使用测试用例而非真实的付费 API 来对关键路径进行压力测试。
将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，以便运维人员无需查看整个图结构即可进行审计。
保持图结构扁平且具有类型约束。嵌套的数据块会隐藏是哪个节点修改了哪个字段，还会在中断后导致无法继续执行。
在升级技术栈之前，先冻结现有版本，为关键路径生成标准操作记录，并确认回滚步骤。共享环境需要设置速率限制、租户验证机制，以及明确的密钥轮换负责人。与其追求花哨的一次性演示，不如注重扎实的可靠性。