200 行 Python 代码,从零手搓极简 Agent,吃透智能体核心原理!

from openai import OpenAIclient = OpenAI(base_url='http://localhost:11434/v1/', api_key='ollama-local')  # 大家按需修改为自己的大模型MODEL ="qwen3.5:4b"# 对话主循环defchat_loop():# 维护对话上下文，实现短期记忆 messages = []    print("=== 极简Agent启动成功，输入/quit退出 ===")whileTrue:        user_input = input("\n你：").strip()# 退出指令if user_input == "/quit":            print("Agent：再见！")break# 加入用户输入到上下文        messages.append({"role": "user", "content": user_input})# 调用大模型        response = client.chat.completions.create(model=MODEL, messages=messages)        reply = response.choices[0].message.content# 加入模型回复到上下文，完成记忆闭环        messages.append({"role": "assistant", "content": reply})        print(f"Agent：{reply}")if __name__ == "__main__":    chat_loop()

import jsondefread_file(file_path: str) -> str:"""读取本地文件"""try:with open(file_path, "r", encoding="utf-8") as f:return f.read()except Exception as e:returnf"读取失败：{str(e)}"defwrite_file(file_path: str, content: str) -> str:"""写入本地文件"""try:with open(file_path, "w", encoding="utf-8") as f:            f.write(content)returnf"写入成功：{file_path}"except Exception as e:returnf"写入失败：{str(e)}"

TOOLS = [    {"type": "function","function": {"name": "read_file","description": "读取本地文件内容","parameters": {"type": "object", "properties": {"file_path": {"type": "string", "description": "要读取的文件路径"}}, "required": ["file_path"]}        }    },    {"type": "function","function": {"name": "write_file","description": "写入内容到本地文件","parameters": {"type": "object", "properties": {"file_path": {"type": "string", "description": "要写入的文件路径"}, "content": {"type": "string", "description": "要写入的内容"}}, "required": ["file_path", "content"]}        }    }]# 工具执行映射TOOL_MAP = {"read_file": read_file,"write_file": write_file}

defchat_loop():    print("=== 极简Agent启动成功，输入/quit退出 ===")# 维护对话上下文，实现短期记忆    messages = []whileTrue:        user_input = input("\n你：").strip()# 退出指令if user_input == "/quit":            print("Agent：再见！")break# 加入用户输入到上下文        messages.append({"role": "user", "content": user_input})# 🔥 内层循环：支持多轮工具调用whileTrue:# 调用大模型，判断是否需要调用工具            response = client.chat.completions.create(model=MODEL, messages=messages, tools=TOOLS, tool_choice="auto")            message = response.choices[0].message# 如果有工具调用，执行工具if hasattr(message, 'tool_calls') and message.tool_calls:                messages.append(message)# 遍历执行所有工具调用for tool_call in message.tool_calls:                    tool_name = tool_call.function.name                    tool_args = json.loads(tool_call.function.arguments)# 执行工具                    tool_result = TOOL_MAP[tool_name](**tool_args)# 把工具结果返回给LLM，开启下一次循环                    messages.append({"role": "tool", "tool_call_id": tool_call.id, "name": tool_name, "content": tool_result})                    print(f"Agent：调用工具{tool_name}({tool_args}) -> {tool_result}")else:# 无工具调用，直接取回复                reply = message.contentbreak# 加入模型回复到上下文，完成记忆闭环        messages.append({"role": "assistant", "content": reply})        print(f"Agent：{reply}")

# 压缩上下文defcompact_context(messages : list):    message_json = json.dumps(messages, ensure_ascii=False)if len(message_json) <= 10000:return messages    compact_prompt = f"将以下对话历史压缩为精简摘要，保留核心信息和关键事实，去除冗余（比如调用工具过程），压缩后不能超过100个字符：\n{message_json}"    compact_response = client.chat.completions.create(model=MODEL, messages=[{"role": "user", "content": compact_prompt}])return [{"role": "assistant", "content": f"【历史对话摘要】{compact_response.choices[0].message.content}"}]

# 对话主循环defchat_loop():    print("=== 极简Agent启动成功，输入/quit退出, 输入/compact压缩上下文, 输入/clear清空上下文 ===")# 维护对话上下文，实现短期记忆    messages = []whileTrue:        user_input = input("\n你：").strip()if user_input == "/compact":            messages = compact_context(messages)            print("Agent：上下文压缩完成！")continueif user_input == "/clear":            messages = []            print("Agent：上下文已清空！")continue# 调用大模型，判断是否需要自动压缩上下文        messages = compact_context(messages)

from sentence_transformers import SentenceTransformer, utilimport numpy as np# 长期记忆向量库LONG_MEMORY_DB = "long_memory_db.json"# 向量化模型（轻量级，CPU可跑）embedding_model = SentenceTransformer('all-MiniLM-L6-v2')# 加载记忆库defload_memory_db():try:with open(LONG_MEMORY_DB, "r", encoding="utf-8") as f:return json.load(f)except:return {"docs": [], "embeddings": []}# 保存记忆库defsave_memory_db(data):with open(LONG_MEMORY_DB, "w", encoding="utf-8") as f:        json.dump(data, f, ensure_ascii=False, indent=2)# 工具1：保存长期记忆（LLM自主调用）defsave_long_memory(content: str) -> str:    data = load_memory_db()    data["docs"].append(content)    data["embeddings"].append(embedding_model.encode(content).astype(np.float32).tolist())    save_memory_db(data)returnf"✅ 记忆已保存：{content[:30]}..."# 工具2：检索相关记忆（LLM自主调用）defquery_long_memory(query: str, top_k=2) -> str:    data = load_memory_db()    docs = data["docs"]    embs = data["embeddings"]ifnot docs:return"暂无长期记忆"    q_emb = embedding_model.encode(query).astype(np.float32)    db_embs = np.array(embs, dtype=np.float32)    scores = np.dot(q_emb, db_embs.T) / (np.linalg.norm(q_emb) * np.linalg.norm(db_embs, axis=1) + 1e-8)    print(scores)    top_idx = np.argsort(scores)[::-1][:top_k]    memories = [docs[i] for i in top_idx if scores[i] > 0.5]ifnot memories:return"未找到相关记忆"return"📌 相关记忆：\n" + "\n".join(memories)

# 工具注册：LLM需要的函数描述格式TOOLS = [    ...    {"type": "function","function": {"name": "save_long_memory","description": "当用户提到重要偏好、习惯、规则、秘密等需要长期记住的内容时，自动保存","parameters": {"type": "object","properties": {"content": {"type": "string", "description": "要记住的内容"}},"required": ["content"]            }        }    },    {"type": "function","function": {"name": "query_long_memory","description": "当回答用户问题需要历史偏好、习惯、规则时，先检索记忆","parameters": {"type": "object","properties": {"query": {"type": "string", "description": "检索关键词"}},"required": ["query"]            }        }    }]# 工具执行映射TOOL_MAP = { ..."save_long_memory": save_long_memory,"query_long_memory": query_long_memory}