摘要：上一篇文章介绍了《从0开发大模型之实现Agent（Bash 到 SKILL）》，有些读者反馈文章太长了，所以这篇文章主要介绍 Agent Loop。 join(all_outputs)             history.append({"role": "user", "content": "Error: Tool call failed or produced no output."})

    return"达到最大执行步数限制，停止执行。

上一篇文章介绍了《从0开发大模型之实现Agent（Bash 到 SKILL）》，有些读者反馈文章太长了，所以这篇文章主要介绍 Agent Loop。

Agent Loop 是什么？

Agent Loop = 调用模型 → 判断是否要用工具 → 执行工具 → 把结果回喂给模型 → 重复
直到模型认为信息足够，输出最终答案为止。
它把大模型从“文本生成器”升级为“能完成任务的执行系统”。

还是用之前最简单执行 Bash 的 Agent 作为样例代码：

import sys
import os
import traceback
from llm_factory import LLMFactory, LLMChatAdapter
from util.mylog import logger
from utils import run_bash, BASH_TOOLS

# 初始化 API 客户端
# 使用 LLMFactory 创建 LLM 实例
llm = LLMFactory.create(
    model_type="openai",
    model_name="deepseek-v3.2", # 使用支持的模型
    temperature=0.0,
    max_tokens=8192
)
client = LLMChatAdapter(llm)

# 系统提示词
SYSTEM = f"""你是一个位于 {os.getcwd()} 的 CLI 代理，系统为 {sys.platform}。使用 bash 命令解决问题。

## 规则：
- 优先使用工具而不是文字描述。先行动，后简要解释。
- 读取文件：cat, grep, find, rg, ls, head, tail
- 写入文件：echo '...' > file, sed -i, 或 cat << 'EOF' > file
- 避免危险操作，如 rm -rf等删除或者清理文件, 或格式化挂载点，或对系统文件进行写操作

## 要求
- 不使用其他工具，仅使用 bash 命令或者 shell 脚本
- 子代理可以通过生成 shell 代码执行
- 如果当前任务超过 bash 的处理范围，则终止不处理
"""

def extract_bash_commands(text):
    """从 LLM 响应中提取 bash 命令"""
    import re
    pattern = r'```bash\n(.*?)\n```'
    matches = re.findall(pattern, text, re.DOTALL)
    return [cmd.strip() for cmd in matches if cmd.strip()]

def chat(prompt, history=None, max_steps=10):
    if history isNone:
        history = []
    
    # 检查历史记录中是否已有系统提示词（作为系统消息）
    has_system = any(msg.get("role") == "system"for msg in history)
    ifnot has_system:
         # 在开头添加系统提示词作为系统消息
         history.insert(0, {"role": "system", "content": SYSTEM})

    history.append({"role": "user", "content": prompt})

    step = 0
    while step < max_steps:
        step += 1
        # 1. 调用模型（传递 tools 参数）
        # 使用 chat_with_tools 接口，支持 function calling
        response = client.chat_with_tools(
            prompt=prompt,
            messages=history,
            tools=BASH_TOOLS
        )
        if step == 1:
            prompt = '继续'

        # 2. 解析响应内容
        assistant_text = []
        tool_calls = []
        
        logger.info(f"第 {step} 步响应: {response}")

        # chat_with_tools 返回的是 Response 对象，包含 content 列表
        for block in response.content:
            if getattr(block, "type", "") == "text":
                assistant_text.append(block.text)
            elif getattr(block, "type", "") == "tool_use":
                tool_calls.append(block)

        # 记录助手文本回复
        full_text = "\n".join(assistant_text)
        if full_text:
            logger.info(f"助手: {full_text}")
            history.append({"role": "assistant", "content": full_text})
        elif tool_calls:
            # 如果只有工具调用没有文本，添加一个占位文本到历史，保持对话连贯
            history.append({"role": "assistant", "content": "(Executing tools...)"})
        
        # 3. 如果没有工具调用，直接返回内容
        ifnot tool_calls:
            logger.info(f"第 {step} 步结束，无工具调用")
            if response.stop_reason == "end_turn":
                return full_text
            # 如果异常结束，也返回
            return full_text or"(No response)"

        # 4. 执行工具
        logger.info(f"第 {step} 步工具调用: {tool_calls}")
        all_outputs = []
        for tc in tool_calls:
            if tc.name == "bash":
                cmd = tc.input.get("command")
                if cmd:
                    logger.info(f"[使用工具] {cmd}")  # 黄色显示命令
                    output = run_bash(cmd)
                    all_outputs.append(f"$ {cmd}\n{output}")
                    # 如果输出太长则截断打印
                    if len(output) > 200:
                        logger.info(f"输出: {output[:200]}... (已截断)")
                    else:
                        logger.info(f"输出: {output}")
            else:
                logger.warning(f"Unknown tool: {tc.name}")
        
        # 5. 将命令执行结果添加到历史记录中
        if all_outputs:
            combined_output = "\n".join(all_outputs)
            history.append({"role": "user", "content": f"执行结果：\n{combined_output}\n\n请继续处理。"})
        else:
            # 有工具调用但没产生输出（可能是解析失败或空命令）
            history.append({"role": "user", "content": "Error: Tool call failed or produced no output."})

    return"达到最大执行步数限制，停止执行。"

if __name__ == "__main__":
    if len(sys.argv) > 1:
        logger.info(chat(sys.argv[1]))
    else:
        # 交互模式
        logger.info("Bash 代理已启动。输入 'exit' 退出。")
        history = []
        whileTrue:
            try:
                user_input = input("> ")
                if user_input.lower() in ['exit', 'quit']:
                    break
                chat(user_input, history)
            except KeyboardInterrupt:
                logger.info("\n正在退出...")
                break
            except Exception as e:
                logger.info(f"\n错误: {e}")
                traceback.print_exc()

Agent Loop 怎么工作？（核心循环）

可以把 Agent Loop 理解成一个递归的工作流（对应上面的代码就是 while 循环，不断将历史数据和执行结果信息输入给LLM）：

1. 模型思考：基于当前上下文理解任务、制定下一步；
2. 选择工具：如果缺信息或需要执行动作，就发起工具调用请求；
3. 执行工具：系统根据工具定义（schema）校验参数并运行工具；
4. 结果回传：把工具结果作为新消息加入对话历史；
5. 回到第 1 步，继续迭代，直到输出最终回答；

核心的东西：上下文会累积，模型不仅看到用户最初的问题，还能看到自己调用过哪些工具、拿到了哪些结果，从而完成多步推理与决策。

一个具体例子：扫描代码库安全漏洞

用户说：“帮我分析这个代码库有没有安全漏洞。”

单次回答不可能完成，因为需要读代码、搜索、归纳，Agent Loop 会这样跑：

1. 模型先需要了解项目结构 → 调用目录列举工具
2. 看到目录结构后，定位入口文件 → 调用文件读取工具
3. 发现有数据库查询 → 再读数据库模块 → 再次调用文件读取工具
4. 识别出疑似 SQL 注入（用户输入直接拼接 SQL）→ 为确认影响面，调用代码搜索工具找调用点
5. 工具返回 12 处调用 → 模型认为信息足够 → 输出最终报告（漏洞说明、位置、修复建议）

每一轮都遵循同一个模式：拿到新信息 → 决定继续行动还是结束输出，而且这些决定是模型基于当前上下文“自主做的”。

Agent Loop 里消息长什么样？（对话历史=工作记忆）

Agent Loop 主要维护一份“对话历史”，它是模型的临时工作记忆。消息通常分两类角色：

• user：用户输入、后续指令、以及（常见实现里）工具结果也可能以用户/系统形式注入
• assistant：模型输出，可能是：
• • 给用户的文本回答
  • 工具调用请求（要用哪个工具、参数是什么） -（部分模型支持）推理痕迹

对话历史会越积越多，因此通常需要 “对话管理” 策略来避免超出上下文窗口（后面会提到常见问题）。

工具执行发生了什么？

当模型请求用工具时，执行系统一般会做这些事：

1. 对工具进行：校验参数，一般工具遵循如下格式：

{
    "type": "tool_use",
    "id": "toolu_01A09q90qw90lq917835123",
    "name": "my_function_name",
    "input": {
        "query": "Latest developments in quantum computing"
    }
}

2. 在工具注册表里找到对应工具 3. 执行工具，并做好错误处理 4. 把成功结果或失败信息，统一封装成 “工具结果消息” 回传给模型

重点：工具失败不会直接让循环崩掉，而是把错误返回给模型，让它有机会调整策略、换工具或重试。

什么时候结束循环？（Stop Reasons）

每次调用模型都会带一个 “停止原因”，决定 Loop 下一步怎么走，常见包括：

• end turn：模型已完成 → 正常结束并返回最终答案
• tool use：模型要用工具 → 执行工具后继续下一轮
• max tokens：输出被 token 上限截断 → 通常不可恢复，需要报错或拆分任务
• stop sequence：遇到预设停止符 → 正常结束
• content filtered / guardrail intervention：触发安全/策略拦截 → 按产品规则处理

常见坑与解决思路

1）上下文窗口耗尽

循环次数多、工具输出长，会把对话历史撑爆，导致输入过长或模型表现变差。

应对方法：

• 让工具返回摘要/关键片段，不要一股脑返回全量
• 简化工具 schema（复杂嵌套也很吃 token）
• 用对话管理策略：滑动窗口、总结压缩等
• 把大任务拆成多个子任务，分段跑、分段总结

2）模型老选错工具

通常是工具描述不清或重叠，模型不知道怎么选。

应对方法：

• 写清楚工具 “适用场景、输入输出、边界”
• 避免多个工具描述高度相似

3）MaxTokensReached（输出太长）

可能是回答太长、或上下文太满导致留给输出的空间不够。

应对方法：

• 增加 token 上限（如果可控）
• 缩短上下文/工具输出
• 拆分任务为子任务，让子任务执行减少上下文窗口

4）复杂的任务不要全部依赖大模型决策

可以参考这篇文章：mp.weixin.qq.com/s/Zhc-GDTJS… ，讲的是为什么大模型不能准确执行所有的 skills (这里其实就是对应工具)，主要原因如下：

• 工具太多，会造成噪声，大模型自主决策不一定能按照你想要的方式调用工具链；
• 没有决策点，大模型会找到最相近的工具，但是有些时候相近的工具不一定是最佳工具；
• 没有排序，大模型自己决策：A -> B -> C，但是不一定符合你调用的要求；

那该怎么做？

• 在提示词或者 rules.md 中增加你当前任务大概需要用到哪些工具，或者强制使用哪些工具；
• 指出具体任务调用的工具顺序可能是怎么样的？虽然看起来像人工决策，但是往往能有效的解决你想要解决的问题；

参考

（1）strandsagents.com/latest/docu…
（2）mp.weixin.qq.com/s/Zhc-GDTJS…