state = init_state(user_goal)
for step in range(max_steps):
    decision = policy(state)        # LLM 决定下一步
    if terminator(decision):         # 是否结束?
        return final_answer(decision)
    observation = action(decision)   # 调工具 / 子 Agent
    state = update(state, observation)  # 更新状态
return forced_finish(state)          # 兜底:步数耗尽

Thought 1: 我需要先查一下用户的订单状态
Action 1:  get_order(order_id="12345")
Observation 1: {"status": "shipped", "tracking": "SF123..."}

Thought 2: 订单已发货,接下来需要根据物流信息估算送达时间
Action 2:  get_eta(tracking="SF123...")
Observation 2: {"eta": "2026-06-28"}

Thought 3: 信息齐全,可以回答用户了
Action 3:  finish(answer="您的订单预计 6 月 28 日送达")

import json, re
from openai import OpenAI

client = OpenAI()
TOOLS = {
    "get_weather": lambda city: f"{city} 今天晴, 25°C",
    "get_time":   lambda _: "现在是 2026-06-26 15:00",
}

SYSTEM = """你是一个 Agent。每轮必须严格输出:
Thought: <你的推理>
Action: <JSON,形如 {"name": "工具名", "args": {...}} 或 {"name": "finish", "args": {"answer": "..."}}>
"""

def react_loop(user_goal, max_steps=8):
    messages = [
        {"role": "system", "content": SYSTEM},
        {"role": "user",   "content": user_goal},
    ]
    for step in range(max_steps):
        resp = client.chat.completions.create(
            model="gpt-4o-mini", messages=messages
        )
        text = resp.choices[0].message.content
        messages.append({"role": "assistant", "content": text})

        # 解析 Action
        m = re.search(r"Action:\s*(\{.*\})", text, re.S)
        if not m:
            raise ValueError(f"step {step}: 模型未输出 Action")
        action = json.loads(m.group(1))

        if action["name"] == "finish":
            return action["args"]["answer"], messages

        # 执行工具 → Observation
        obs = TOOLS[action["name"]](**action["args"])
        messages.append({"role": "user", "content": f"Observation: {obs}"})

    raise TimeoutError("超过最大步数,未收敛")

[生成阶段]  LLM 生成初稿 answer_0
[反思阶段]  LLM(可以是同一个,也可以是更强的 critic)输出:
            - score: 0~10
            - critique: 具体问题清单
            - improved_answer: 改进版
[循环 N 次] 或 直至 score >= threshold

def reflection_loop(draft_prompt, max_rounds=3, threshold=8):
    draft = llm(draft_prompt)
    history = []
    for r in range(max_rounds):
        critique = critic_llm(f"""
            请评估以下回答,输出 JSON:
            {{"score": 0-10, "issues": [...], "improved": "..."}}
            原题: {draft_prompt}
            当前回答: {draft}
        """)
        score, improved = parse(critique)["score"], parse(critique)["improved"]
        history.append({"round": r, "score": score})
        if score >= threshold:
            return improved, history
        draft = improved
    return draft, history  # 兜底:返回最后一轮

Trajectory  → Reflector  → Self-Critique
                                    ↓
                          Memory Store (vector db)
                                    ↓
            下次同类任务 ← 检索相关反思 ←┘

外层 Planner Loop:   任务 → 计划(步骤列表)
内层 Executor Loop:  对计划中每个 step 反复 ReAct,直至完成
外层 Replanner:      若某个 step 失败,回到 Planner 重新规划

User Proxy  ──►  Assistant  ──►  User Proxy  ──►  Assistant ...
   ↑                                                    │
   └──────────── Critic/Inception Prompt ←──────────────┘

from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # 对话历史(自带 reducer,自动追加)
    messages: Annotated[List[dict], add_messages]
    # 当前计划(Plan-and-Execute 用)
    plan: List[str]
    # 已执行的步骤与结果
    past_steps: List[tuple[str, str]]
    # 反思/记忆
    reflections: List[str]
    # 控制字段
    step_count: int
    is_finished: bool

def should_continue(state) -> Literal["continue", "end"]:
    # 1. 显式 finish(由 LLM 主动调用 finish 工具)
    if state["is_finished"]:
        return "end"
    # 2. 步数硬上限(防止 token 爆炸)
    if state["step_count"] >= 20:
        return "end"
    # 3. 步数软上限 + 强制收敛(兜底,走 LLM 总结)
    if state["step_count"] >= 15:
        return "force_summarize"
    # 4. 重复检测(防止模型在两个动作间反复横跳)
    if state["messages"][-1].content == state["messages"][-3].content:
        return "end"
    return "continue"

class LongTermMemory:
    def __init__(self, vector_store, embedder):
        self.vs, self.emb = vector_store, embedder

    def write(self, situation: str, lesson: str):
        self.vs.add(
            text=f"情境: {situation}\n教训: {lesson}",
            embedding=self.emb.embed(situation),
        )

    def recall(self, current_situation: str, k=3):
        q = self.emb.embed(current_situation)
        return self.vs.search(q, top_k=k)

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件(危险操作,需要 confirm=true 才真正发送)。"""
    if not confirm:
        return f"[DRY-RUN] 将发送邮件给 {to},主题: {subject}"
    return _smtp_send(to, subject, body)

from typing import Literal, Annotated, List
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode

@tool
def search(query: str) -> str:
    """模拟搜索工具。"""
    return f"关于 '{query}' 的搜索结果: ..."

tools = [search]
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)

class State(TypedDict):
    messages: Annotated[List, add_messages]
    step_count: int

def agent(state: State):
    return {
        "messages": [llm.invoke(state["messages"])],
        "step_count": state.get("step_count", 0) + 1,
    }

def should_continue(state: State) -> Literal["tools", END]:
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"
    return END

graph = (
    StateGraph(State)
    .add_node("agent", agent)
    .add_node("tools", ToolNode(tools))
    .add_edge(START, "agent")
    .add_conditional_edges("agent", should_continue)
    .add_edge("tools", "agent")
    .compile()
)

for chunk in graph.stream({"messages": [("user", "查一下 LangGraph 最新版本")]}):
    print(chunk)

def reflect(state: State):
    last_answer = state["messages"][-1].content
    critique = llm.invoke([
        {"role": "system", "content": "你是质检员,打分 0-10,并指出问题。"},
        {"role": "user",   "content": f"评价: {last_answer}"},
    ])
    score = parse_score(critique.content)
    if score >= 8 or state["step_count"] >= 10:
        return END
    return "agent"  # 不满意,把 critique 塞回 messages 让 agent 重写

graph.add_node("human_review", lambda s: s)  # 实际是中断点
graph.add_edge("agent", "human_review",
               condition=lambda s: "send_email" in str(s["messages"][-1]))

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "..."

# 之后每次 graph.invoke() 都会自动 trace 到 LangSmith

with tracer.start_as_current_span("agent_loop") as span:
    span.set_attribute("goal", user_goal)
    for step in range(max_steps):
        with tracer.start_as_current_span(f"step_{step}") as s:
            s.set_attribute("messages_count", len(state["messages"]))
            s.set_attribute("tokens_in",  usage.prompt_tokens)
            s.set_attribute("tokens_out", usage.completion_tokens)
            ...

# 反例
for step in itertools.count():  # 真的无限
    ...

# 反例
done = llm("请判断任务是否完成,只回答 yes/no") == "yes"

┌─────────────────────────────────────────────────┐
│                  主循环 (Agent Loop)              │
│                                                 │
│   ┌────────┐    ┌────────┐    ┌─────────────┐   │
│   │ Planner│───►│Executor│───►│  Reflector  │   │
│   └────────┘    └────┬───┘    └──────┬──────┘   │
│        ▲             │              │          │
│        │             ▼              ▼          │
│        │        ┌────────┐    ┌─────────────┐   │
│        └────────│Replanner│    │Memory Writer│   │
│                 └────────┘    └─────────────┘   │
│                                                 │
│   三重保险:Step 上限 / Token 上限 / 重复检测      │
│   三层记忆:State / Vector DB / External Store   │
└─────────────────────────────────────────────────┘