从零构建 Agent Loop：从一个 while 循环到生产级 ReAct 框架

def agent_loop(user_input: str):
    messages = [
        {"role": "system", "content": "你是一个助手，可以调用工具。"},
        {"role": "user", "content": user_input},
    ]

    while True:
        response = call_llm(messages, tools=available_tools)

        if response.has_tool_calls():
            # 有工具调用 → 执行工具，把结果追加到上下文，继续循环
            for tool_call in response.tool_calls:
                result = execute_tool(tool_call)
                messages.append({"role": "tool", "content": result})
        else:
            # 没有工具调用 → 模型认为可以直接回答了
            return response.text

用户输入 → [LLM 思考] ──有工具调用──→ [执行工具] ──→ 回到 LLM
                │
                └──无工具调用──→ 输出最终回答 → 结束

MAX_ITERS = 20

def agent_loop(user_input: str):
    messages = build_initial_messages(user_input)

    for cur_iter in range(MAX_ITERS):
        response = call_llm(messages, tools=available_tools)

        if not response.has_tool_calls():
            return response.text  # 正常结束

        for tool_call in response.tool_calls:
            result = execute_tool(tool_call)
            messages.append({"role": "tool", "content": result})

    # 超限了
    return "抱歉，我尝试了很多次但没能完成任务。"

from enum import Enum

class EventType(Enum):
    THINKING = "thinking"       # 模型正在思考
    TEXT_DELTA = "text_delta"   # 文本片段
    TOOL_CALL = "tool_call"     # 工具调用开始
    TOOL_RESULT = "tool_result" # 工具返回结果
    REPLY_END = "reply_end"     # 最终回答完成

def agent_loop_stream(user_input: str):
    messages = build_initial_messages(user_input)

    for cur_iter in range(MAX_ITERS):
        yield {"type": EventType.THINKING, "iter": cur_iter}

        response = call_llm(messages, tools=available_tools)

        if not response.has_tool_calls():
            # 流式输出最终文本
            for chunk in response.stream_text():
                yield {"type": EventType.TEXT_DELTA, "text": chunk}
            yield {"type": EventType.REPLY_END}
            return

        for tool_call in response.tool_calls:
            yield {"type": EventType.TOOL_CALL, "name": tool_call.name}
            result = execute_tool(tool_call)
            yield {"type": EventType.TOOL_RESULT, "output": result}
            messages.append({"role": "tool", "content": result})

class Tool:
    name: str
    description: str
    input_schema: dict          # JSON Schema
    is_read_only: bool          # 只读工具（如搜索）→ 可以自动批准
    is_concurrency_safe: bool   # 并发安全 → 可以批量执行

    async def call(self, **kwargs) -> str:
        raise NotImplementedError

# AgentScope 的策略：根据工具属性自动决定
safe_calls = [c for c in tool_calls if tool.is_concurrency_safe]
unsafe_calls = [c for c in tool_calls if not tool.is_concurrency_safe]

# 安全的并发执行
safe_results = await asyncio.gather(*[tool.call(**c.args) for c in safe_calls])

# 不安全的顺序执行
for call in unsafe_calls:
    result = await tool.call(**call.args)

Toolkit
├── basic 组（始终可用）: search, calculator, datetime
├── code 组（按需激活）: run_python, run_shell, read_file
└── data 组（按需激活）: query_db, plot_chart, export_csv

请求进入 →
  [Middleware A: before] →
    [Middleware B: before] →
      [核心逻辑]
    [Middleware B: after] →
  [Middleware A: after] →
← 响应返回

class MiddlewareBase:
    async def on_reply(self, agent, input_kwargs, next_handler):
        """拦截整个 reply 过程"""
        async for event in next_handler():
            yield event

    async def on_reasoning(self, agent, input_kwargs, next_handler):
        """拦截推理阶段"""
        async for event in next_handler():
            yield event

    async def on_acting(self, agent, input_kwargs, next_handler):
        """拦截工具执行"""
        async for event in next_handler():
            yield event

    async def on_check_permission(self, agent, input_kwargs, next_handler):
        """拦截权限检查"""
        async for event in next_handler():
            yield event

    # ... 还有 on_model_call, on_compress_context, on_system_prompt

class LoggingMiddleware(MiddlewareBase):
    async def on_reasoning(self, agent, input_kwargs, next_handler):
        logger.info(f"[{agent.name}] 开始推理...")
        async for event in next_handler():
            yield event
        logger.info(f"[{agent.name}] 推理完成")

    async def on_acting(self, agent, input_kwargs, next_handler):
        logger.info(f"[{agent.name}] 执行工具...")
        async for event in next_handler():
            yield event

PENDING → ASKING（等待确认）→ ALLOWED → FINISHED
                            → DENIED

async def _execute_tool_call(self, tool_call):
    permission = await self._check_permission(tool_call)

    if permission == "ASK":
        # 暂停！yield 一个事件告诉外部"我需要确认"
        yield RequireUserConfirmEvent(tool_call=tool_call)
        return  # 退出当前循环，状态保存

    elif permission == "DENY":
        yield ToolResult(state="DENIED", error="权限不足")

    elif permission == "ALLOW":
        result = await self.toolkit.call_tool(tool_call)
        yield ToolResult(state="FINISHED", output=result)

MAX_TOOL_RESULT = 4000  # 字符

def truncate_tool_result(result: str, limit: int = MAX_TOOL_RESULT) -> str:
    if len(result) <= limit:
        return result
    return result[:limit] + f"\n... [已截断，原始长度 {len(result)} 字符]"

class ContextConfig:
    trigger_ratio: float = 0.8    # 上下文使用率超过 80% 时触发
    reserve_ratio: float = 0.1    # 压缩后保留 10% 的空间

async def compress_context(self, messages):
    """当上下文过长时，对早期消息做摘要"""
    # 保留最近 N 条消息不动
    # 对更早的消息调用 LLM 生成摘要
    # 用摘要替换原始消息
    return compressed_messages

class InjectionConfig:
    inject_time: bool = True         # 注入当前时间
    inject_task_state: bool = True   # 注入任务进度
    inject_context_length: bool = True # 注入上下文长度感知

reply_stream(user_input)
  │
  ├── 1. 检查输入事件（新消息 / HITL 恢复）
  ├── 2. 构建 ReplyContext（中间件链、状态、上下文）
  │
  └── 3. ┌──── ReAct 主循环 ────────────────────────────┐
          │                                               │
          │  while True:                                  │
          │    action = _next_action()   ← 决策器         │
          │                                               │
          │    if action == Exit:                         │
          │      yield ReplyEndEvent → return             │
          │                                               │
          │    if action == Reasoning:                    │
          │      async for event in _reasoning():         │
          │        yield event     ← 流式推理事件          │
          │                                               │
          │    if action == Acting:                       │
          │      async for event in _batch_tool_calls():  │
          │        yield event     ← 流式工具执行事件       │
          │                                               │
          │    cur_iter += 1                              │
          └───────────────────────────────────────────────┘

def _next_action(self) -> Reasoning | Acting | Exit:
    # 1. 有待执行的 tool calls（已授权）→ Acting
    if self._has_pending_tool_calls():
        return Acting()

    # 2. 有等待确认/外部执行的 tool calls → Exit（暂停）
    if self._has_waiting_tool_calls():
        return Exit(reason="waiting_for_confirmation")

    # 3. 结构化输出检查 → 完成则 Exit，否则继续 Reasoning
    if self._needs_structured_output():
        if self._structured_output_complete():
            return Exit(reason="structured_output_done")
        return Reasoning(hint="请输出结构化数据", tool_choice="required")

    # 4. 已有最终文本回复 → Exit
    if self._has_final_message():
        return Exit(reason="final_message")

    # 5. 超出迭代上限 → Exit
    if self.cur_iter >= self.max_iters:
        return Exit(reason="exceed_max_iters")

    # 6. 默认 → 继续推理
    return Reasoning()

async def _reasoning(self, tool_choice=None):
    # 1. 准备模型输入（messages + tools schema + 状态注入）
    model_input = self._prepare_model_input(tool_choice)

    # 2. 调用 LLM（经过 middleware 链包装）
    async for chunk in self._call_model_through_middleware(model_input):
        # 3. 流式处理每个 chunk → 转换为事件
        if chunk.is_text:
            yield TextBlockDeltaEvent(text=chunk.text)
        elif chunk.is_thinking:
            yield ThinkingBlockDeltaEvent(text=chunk.text)
        elif chunk.is_tool_call:
            yield ToolCallStartEvent(tool_call=chunk.tool_call)

    # 4. 保存到上下文
    self._save_to_context(response)

async def _batch_tool_calls(self):
    tool_calls = self._get_pending_tool_calls()

    # 按并发安全性分组
    concurrent = [c for c in tool_calls if c.tool.is_concurrency_safe]
    sequential = [c for c in tool_calls if not c.tool.is_concurrency_safe]

    # 并发执行安全的工具
    if concurrent:
        tasks = [self._execute_tool_call(c) for c in concurrent]
        async for event in merge_async_generators(*tasks):
            yield event

    # 顺序执行不安全的工具
    for call in sequential:
        async for event in self._execute_tool_call(call):
            yield event

# 传统方式：显式编排
pipeline = SequentialPipeline([
    ResearchAgent(),
    WritingAgent(),
    ReviewAgent(),
])
pipeline.run(input)

# Leader Agent 的 toolkit 里有特殊的"团队工具"
leader = Agent(
    name="项目经理",
    system_prompt="你是一个项目经理，负责协调团队完成任务...",
    toolkit=Toolkit(tools=[
        spawn_agent_tool,      # 创建一个 Worker Agent
        assign_task_tool,      # 给 Worker 分配任务
        collect_result_tool,   # 收集 Worker 的结果
    ]),
)

第1层  while True 循环          → 最小可用的 ReAct
第2层  迭代上限                 → 防止死循环（循环决策的雏形）
第3层  yield 事件流             → 流式输出，实时可观察
第4层  工具协议 + 并发策略       → 从裸函数到标准化工具系统
第5层  中间件（洋葱模式）        → 横切逻辑的优雅解耦
第6层  HITL 状态机              → 可中断、可恢复的循环
第7层  上下文管理               → 防止 token 爆炸
第8层  整合为完整 ReAct 系统     → 决策器 + 推理 + 执行 + 事件流
第9层  多 Agent 编排            → Leader-Worker + Message Bus