【大模型】如何写一个简单的agent

条件

需要能支持tool calling的模型，比如我用的qwen3.5

实现原理

1. 发请求：把 messages 发给 AI。
2. 看返回：

• 如果返回里有 tool_calls 说明 AI 需要帮忙，代码去执行真实函数，把结果追加到 messages，继续下一次循环。
• 如果返回里只有 content（没有 tool_calls）说明 AI 已经拿到了所有信息并整理好了答案，退出循环，打印结果。

代码

importrequestsimportjson# ==========================================# 1. 预制“真家伙”：真实的工具函数# ==========================================defget_weather(city:str)->str:""" 这是一个模拟查询天气的函数。 在真实场景中，这里会是调用第三方天气API的代码。 """# 实际开发中这里可以替换为真实的 HTTP 请求returnf"{city}现在的天气是晴天，气温 25℃。"# 工具注册表：把函数名和函数本身关联起来，方便后面通过名字调用TOOL_FUNCS={"get_weather":get_weather}# ==========================================# 2. 预制“说明书”：Ollama 格式的工具 Schema# ==========================================# 这是给 AI 看的“工具使用手册”，告诉它有哪些工具可以用，以及怎么用。OLLAMA_TOOLS=[{"type":"function",# 工具类型：函数"function":{"name":"get_weather",# 函数名，必须和 TOOL_FUNCS 里的键一致"description":"获取指定城市的当前天气情况",# 函数功能描述，AI 会根据这个决定何时调用"parameters":{# 函数的参数定义，遵循 JSON Schema 规范"type":"object","properties":{"city":{"type":"string","description":"需要查询天气的城市名称"}},"required":["city"]# 指明哪些参数是必须的}}}]# ==========================================# 3. 核心：Agent 循环逻辑# ==========================================defrun_ollama_agent(user_query:str):# 初始化对话上下文（messages）# 这是 Agent 的“记忆”，它会带着这份记忆和 AI 进行多轮对话messages=[{"role":"system","content":"你是一个有用的助手，需要时使用工具获取信息。"},{"role":"user","content":user_query}]# Ollama API 地址url="http://xxxxx/api/chat"# 开启 Agent Loop（防止死循环，最多循环10次）forstepinrange(10):print(f"\n--- 第{step+1}步：调用 Ollama 模型 ---")# 构建发送给 Ollama 的请求体payload={"model":"qwen3:8b",# 指定使用的模型"messages":messages,# 把完整的对话历史发给模型，让它基于上下文思考"keep_alive":"1h","tools":OLLAMA_TOOLS,# 把“工具说明书”也发给模型"stream":False# 设置为 False，让 API 一次性返回完整结果（但 Ollama 底层仍是流式）}# 发送请求给 Ollamatry:response=requests.post(url,json=payload,timeout=(60,1800))response.raise_for_status()# 检查 HTTP 请求是否成功# --- 开始处理 Ollama 的流式响应 ---# 即使 stream=False，Ollama 的 /api/chat 接口返回的仍是一个 JSON 流# 我们需要逐行读取并拼接，直到收到最后一块数据full_content=""assistant_msg={"role":"assistant","content":""}forlineinresponse.iter_lines():ifline:try:# 将每一行字节流解码为 JSON 对象json_line=json.loads(line.decode('utf-8'))# 拼接文本内容（content 字段在每个数据块里都有）if"message"injson_lineand"content"injson_line["message"]:full_content+=json_line["message"]["content"]# 检查是否是最后一个数据块 (done=True)ifjson_line.get("done"):assistant_msg["content"]=full_content# 保存最终拼接的完整文本# 关键逻辑：只在最后一个数据块中检查是否有工具调用请求# tool_calls 字段只会出现在流式响应的最后一个 JSON 对象中if"message"injson_lineand"tool_calls"injson_line["message"]:assistant_msg["tool_calls"]=json_line["message"]["tool_calls"]break# 收到最后一块数据，跳出循环exceptjson.JSONDecodeError:continue# 跳过无法解析的行# 将模型的完整回复（可能包含 tool_calls）追加到对话历史中messages.append(assistant_msg)exceptrequests.exceptions.ConnectionError:print("❌ 连接失败，请检查网络或Ollama服务")breakexceptrequests.exceptions.Timeout:print("❌ 请求超时")breakexceptExceptionase:print(f"❌ 发生未知错误:{e}")break# --- Agent 决策逻辑 ---# 判定1：如果回复中没有 tool_calls，说明 AI 认为任务已完成，可以直接回答用户if"tool_calls"notinassistant_msg:print("✅ 模型完成思考，输出最终回复。")returnassistant_msg["content"]# 判定2：如果回复中有 tool_calls，说明 AI 需要调用工具来获取信息fortool_callinassistant_msg["tool_calls"]:func_name=tool_call["function"]["name"]# 获取要调用的函数名func_args=tool_call["function"]["arguments"]# 获取函数参数print(f"🔧 执行工具:{func_name}，参数:{func_args}")# 在工具注册表中找到对应的真实函数并执行result=TOOL_FUNCS[func_name](**func_args)print(f"📦 工具返回结果:{result}")# 将工具的执行结果以 "tool" 角色追加到对话历史中# 这样，在下一次循环时，AI 就能看到工具返回的数据了messages.append({"role":"tool","content":result})# ==========================================# 4. 启动 Agent# ==========================================if__name__=="__main__":# 启动 Agent 并传入用户的问题final_answer=run_ollama_agent("南京和成都的天气怎么样？")print("\n最终答案：",final_answer)

特别说明

forlineinresponse.iter_lines():ifline:try:# 将每一行字节流解码为 JSON 对象json_line=json.loads(line.decode('utf-8'))# 拼接文本内容（content 字段在每个数据块里都有）if"message"injson_lineand"content"injson_line["message"]:full_content+=json_line["message"]["content"]# 检查是否是最后一个数据块 (done=True)ifjson_line.get("done"):assistant_msg["content"]=full_content# 保存最终拼接的完整文本# 关键逻辑：只在最后一个数据块中检查是否有工具调用请求# tool_calls 字段只会出现在流式响应的最后一个 JSON 对象中if"message"injson_lineand"tool_calls"injson_line["message"]:assistant_msg["tool_calls"]=json_line["message"]["tool_calls"]break# 收到最后一块数据，跳出循环exceptjson.JSONDecodeError:continue# 跳过无法解析的行中，详细解释下这个代码 # 拼接文本内容（content 字段在每个数据块里都有）if"message"injson_lineand"content"injson_line["message"]:full_content+=json_line["message"]["content"]

这段代码的核心作用是：在流式（Streaming）响应中，像拼图一样，把 AI 吐出来的一小块一小块的文本，拼凑成一句完整的话。

虽然在请求头里设置了stream: False，但 Ollama 的底层 API 实际上返回的仍然是一个换行符分隔的 JSON 流（JSON Lines）。这意味着 AI 不会一次性把一大段话发给你，而是像打字机一样，一个字或一个词地往外蹦，每蹦出一个片段，就包装成一个 JSON 发给你。

我们来逐行拆解这段代码：

1. 安全地“拆快递”（防御性编程）

if"message"injson_lineand"content"injson_line["message"]:

因为 AI 返回的每一个 JSON 数据块，结构可能不一样。有些数据块里只有进度信息，没有message字段；有些有message，但可能只有thinking（思考过程）而没有content（最终回复）。

2. 核心动作：拼接文本

full_content+=json_line["message"]["content"]

• 假设 AI 要回复“你好”，它可能会分三次发给你：
  * 第 1 个数据块：{"message": {"content": "你"}}
  * 第 2 个数据块：{"message": {"content": "好"}}
  * 第 3 个数据块：{"message": {"content": "！"}}
• 作用： 这行代码就是把这三个片段依次累加到full_content变量里。经过三次循环，full_content就会变成完整的"你好！"。

这段代码就是一个**“流式文本拼接器”**。它一边接收 AI 吐出的碎片，一边把它们拼成完整的句子，同时还能保证在遇到残缺数据时不崩溃。等最后拼完了，再把完整的full_content交给后面的逻辑去处理。