您的位置: 首页> AI模型

再谈Agent Loop：大模型 “能做事” 的核心机制

匿名上传

发布时间:2026-02-09 10:12:01

上一篇文章介绍了《从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'```bashn(.*?)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}nn请继续处理。"})
        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）：

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

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

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

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

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

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

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

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

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

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

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

工具执行发生了什么？

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

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

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