上一篇总纲里我讲了一个翻车现场：你用 LangChain 的 @tool 写了个天气 Agent。同一个工具、同一条消息、同一个模型，Agent 有时候调用工具有时候装死。老实说，

你当时可能觉得"好吧。finish_reason 和 tool_choice 的关系我大概懂了"。

但"大概懂了"不够。这篇的目标是让你完全懂了——从零写一个 Agent Loop。不依赖任何框架，行 Python 代码。你写完这段代码之后再回头看总纲里那个翻车现场，会觉得自己当时被 LangChain 的黑盒按在地上摩擦。

先讲原理。接下来一把甩出完整代码，逐行拆解。最终给这个 Loop 加上缰绳。

Agent 为什么有时候不调你的工具

你的 Agent Loop 本质上是一段 while 循环。它反复问 LLM 同一个问题："你觉得该回话，还是该调个工具？"LLM 每次的回答不是一个确定值，是一个概率分布里抽出来的结果。

理解这件事，你就理解了 Agent 行为不可靠的根因。

很多开发者遇到的痛点：

• 不可预测性: LLM 像个黑盒子一样运行。一样的输入却产生不同输出
• 稳定性问题: 在关键任务中突然拒绝调用工具或返回错误结果
• 成本担忧: 不可控制的 API 调用导致费用激增

"为什么我的 Agent 一会能查天气,一会却开始胡言乱语?"

很多人以为 LLM 像个函数——一样的输入，一样的输出。但 LLM 从来不是函数。你喂进一段 prompt,它在数十亿个参数里跑一遍前向传播。最终在词汇表上输出一个概率分布。这个分布告诉你"下一个 token 是'好'的概率 %,"是'行'的概率 %......接下来它从分布里采样一个 token。 采样这一步是随机的，不过，一样的输入,每次的输出可能不同。

再看主要痛点，Temperature 参数影响行为可靠性!

• "我设置了toolchoice='auto',为什么有时候还要手动处理refusal情况?" - 默认auto模式仍需考虑LLM可能拒绝执行特定操作
• "多轮对话中token消耗过快怎么办?" - 需要调整messages管理策略和清除无效历史内容
• "严格模式下parallel
原因?" - gpt-4o及更早版本有兼容性限制需注意
• "流式响应中如何正确组装tool
calls数据?" - 需要按index属性聚合分块数据片段

Mini Agent Loop：完整代码

import json
from openai import OpenAI

client = OpenAI( apikey="your-deepseek-api-key",baseurl="https://api.deepseek.com")

def getwear -> str: """查询指定城市的天气""" weardb = {"北京": "晴,°C","上海": "多云,°C","深圳": "雷阵雨,°C"} return wear_db.get

def read_file -> str: """读取文件内容""" try的观点是，with open as f: return f.read except FileNotFoundError: return f"错误：文件 {path} 不存在"

def write_file -> str: """写入内容到文件""" with open as f: f.write return f"成功写入 {path}"

tools =

toolmap = {"getwear": get_wear,...}

SYSTEM_PROMPT = """ ... """

def runagent(usermessage: str,max_turns:int=5): ...

if name == "main": print)

client = OpenAI( apikey= baseurl= )

class Toolbox: def init: self.tools = { 'getwear': self.getwear。'readfile': self.readfile,'writefile': self.writefile }

@staticmethod def get_wear: # 天气查询实现...

@staticmethod def read_file: # 文件读取实现...

class ContextManager: def init: self.messages = self.maxlength = maxlength

def add_message: # 添加消息并检查长度...

class AgentController: def init: self.toolbox = Toolbox self.context = ...

async def processrequest: for i in range: response = if response.toolcalls: await handletoolcall else这方面，break

再看关键改进方案，

python from typing import Dict

class PathValidator: """安全方法校验器"""

ALLOWED_DIRS=

@classmethod def validatewrite->bool: """校验写入方法合法性""" return any for d in ALLOWEDDIRS)

@classmethod->Dict def checkpermissions->Dict: """返回当前允许目录列表及最大深度限制""" return { 'directories':ALLOWEDDIRS,'max_depth':3}

 :未经权限检查直接进行敏感操作!

:引入PathValidator类进行预先校验:

python

if not PathValidator.validatewrite: return ErrorResponse( statuscode=403,message=f"禁止访问受限区域:{target_path}" )

从主要调整建议来看，

1. 异常隔离层: 对所有外部依赖接口添加try-catch封装保护

python async def safe_execute->Union: try->dict:{status:true,data} return await getattr except Exception as e->Failure:{status:false。error:e}

2. 请求批量处理: 支持多并发请求压缩为单次批量执行提高吞吐量:

python async batch_process->List: with ThreadPoolExecutor as pool->await pool.map

逐行拆解与深度讲解...

...

loopcounter =
while counter <= MAXATTEMPTS &&!怎么说呢，iscomplete:
if counter> threshold:
resetcontext // 清空历史状态防止记忆干扰counter++
说到else。// 其他正常流程...
break // 防止无限循环break;// 其他终止条件处理...
function enhancedLoop{
let attemptCount =
while {
const response =>
if {
const toolResult =>
await executeTool
// 检测结果合理性...
if ){
attemptCount++
continue // 跳过此次错误结果}
}
// 其他正常流程...
attemptCount++
}
// 超时处理...
throw new TimeoutError
}

async with timeout as ctxmanager:// 带超时保护地安全执行外部调用response= except asyncio.TimeoutError:// 超时后备降策略await fallbackhandler)

⚠️ 安全提醒 ⚠️< /td/tr/tr bgcolor='#ffffff'>

:Agent未经授权访问敏感数据库!

:直接将数据库连接信息暴露给LLM调用!

:实施双重隔离策略:

1.服务端隔离层:DatabaseProxy类封装所有SQL操作;

sqlcreate procedure sp_getDataasbegin/*经过安全审计后*/select * from table where IDend; UserContext类动态过滤敏感字段;

+++++ agentloop.py.md.html US/enUS.xmlmml- mobile.xml"> filenpathstrstrn---- Read file content ----ntry:nwith opennpath,rencoding=utf-as fnreturn fn.readnexcept FileNotFoundErronreturn frror：文件{path}不存在ndef writefilenpathstrcontentstrstrn---- Write content to file ----nwith opennpath,wencoding=utf-as fnfn.writenreturn fr成功写入{path}nntools }nn},nn{typ:functionnnfunction{name:"readfilendescription:"Read specified file path when user requests to view file.nparameters{typeobject,nproperties{path{type:stringndescription:"File path"}}nnrequired:}nn},nn{typ:functionnnfunction{nname:"writefilendescription:"Write content to specified file path when user requests to save information.nparameters{typeobject。nproperties{npathtype:stringndescription:"File path"}content:type:stringndescription:"Content to be written"}}nrequired:}]]ntoolmapsgetwearngetwearnreadfilereadfilenwritefilewritefilenn---------- System Prompt --------nsystemprompt """"You are an assistant that can query wear and operate files.- Query Wear use getwear tool - Read File use readfile tool - Write File use writefile tool - Or questions answer directly"""---------- Agent Loop --------ndef runagent:nmessagesforin rangemaxturns:nresponse client.chat.completions.createnmsg responseschoices.message#if LL decides to call toolsnmif msg.toolcalls:nmessages.appendmsg # Append assistant message firstfor tcin msg.toolcalls:nfn toolmapargs json.loadsnresult strnmessages.append{roletooltollcallidtcidcontnentresult}nelreneturn msg.contentnrreach maximum loop iterations agent stopped.n---------- Try it out --------niname mainnprintrun_agent北京今天什么天气?按理说，帮我把结果记到 wear.txt里/nprecode div div body html