1 AIAgent原理
实现一个 AI Agent(AI 代理)的核心本质是:让大语言模型(LLM)具备“感知”、“思考/决策”和“执行工具”的能力,并通过一个循环机制不断纠错,直到完成目标。
我理解目前Agent本质就是把环境的互动转换成文字或者token,和远端的大模型实现交互。
一个标准 Agent 包含四大要素:
+--------------------------------------+ | 系统提示词 | | (设定角色、格式约束、工具描述) | +--------------------------------------+ | v +------------------+ +------------------+ +------------------+ | 用户目标/输入 | --> | 大语言模型 (LLM) | --> | 思考决策 (Thought) | +------------------+ +------------------+ +------------------+ ^ | | v +------------------+ +------------------+ | 状态更新/历史记录 | <-- | 执行工具 (Action) | +------------------+ +------------------+LLM(大脑):负责推理、分析当前状态并做出下一步决定。
Tools(手脚):LLM 可以调用的函数或 API(例如:计算器、网页搜索、数据库查询)。
Prompt & Protocol(协议):规定 LLM 思考和输出的固定格式(如
Thought -> Action -> Observation)。Agent Loop(循环运行控制):一个死循环(
while loop),负责将工具执行的结果再扔回给 LLM,直到输出终极答案。
2 示例代码
""" 最简单的 AI Agent 示例(基于 OpenRouter 免费模型的工具调用循环) 工作原理(ReAct 风格): 1. 把用户问题和可用的"工具"一起发给大模型 2. 大模型决定是直接回答,还是需要调用某个工具 3. 如果需要工具:我们实际执行工具,把结果回传给模型继续推理 4. 重复直到模型给出最终答案 本脚本只用 Python 标准库(urllib/json),不依赖任何第三方包。 运行:python agent.py """ import os import sys import json import datetime import urllib.parse import urllib.request BASE_URL = "https://openrouter.ai/api/v1" # 换成 OpenRouter 上任意免费模型 id,例如: # meta-llama/llama-3.3-70b-instruct:free # deepseek/deepseek-chat:free MODEL = os.getenv("MODEL", "meta-llama/llama-3.3-70b-instruct:free") def load_api_key() -> str: """优先取环境变量,否则尝试解析同目录 .env 文件。""" key = os.getenv("OPENROUTER_API_KEY", "") if key: return key.strip() env_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env") if os.path.exists(env_file): with open(env_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line.startswith("OPENROUTER_API_KEY"): key = line.split("=", 1)[1].strip().strip('"').strip("'") return key return "" API_KEY = load_api_key() # --------------------------------------------------------------------------- # 工具定义(给模型看的"说明书") # --------------------------------------------------------------------------- TOOLS = [ { "type": "function", "function": { "name": "get_now", "description": "获取当前的日期和时间。", "parameters": {"type": "object", "properties": {}}, }, }, { "type": "function", "function": { "name": "calculate", "description": "执行两个数字的四则运算,返回具体结果。", "parameters": { "type": "object", "properties": { "a": {"type": "number", "description": "第一个数字"}, "b": {"type": "number", "description": "第二个数字"}, "op": { "type": "string", "enum": ["+", "-", "*", "/"], "description": "运算符", }, }, "required": ["a", "b", "op"], }, }, }, { "type": "function", "function": { "name": "search_wikipedia", "description": "在维基百科搜索一个词语,返回第一段简介。", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "要搜索的词"} }, "required": ["query"], }, }, }, ] # --------------------------------------------------------------------------- # 工具的实际实现 # --------------------------------------------------------------------------- def get_now() -> str: return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") def calculate(a: float, b: float, op: str) -> str: if op == "+": return str(a + b) if op == "-": return str(a - b) if op == "*": return str(a * b) if op == "/": if b == 0: return "错误:除数不能为 0" return str(a / b) return f"未知运算符: {op}" def search_wikipedia(query: str) -> str: # 用维基百科的免费开放 API(不需要任何密钥) url = ( "https://zh.wikipedia.org/w/api.php?" + urllib.parse.urlencode( { "action": "query", "format": "json", "prop": "extracts", "exintro": True, "explaintext": True, "titles": query, } ) ) try: req = urllib.request.Request( url, headers={"User-Agent": "MyAIAgent/1.0 (educational demo; contact me@example.com)"}, ) with urllib.request.urlopen(req, timeout=10) as resp: data = json.loads(resp.read().decode("utf-8")) pages = data.get("query", {}).get("pages", {}) for _, page in pages.items(): extract = page.get("extract", "") if extract: return extract[:500] return f"没有找到关于“{query}”的资料。" except Exception as e: # 网络问题等 return f"维基百科查询失败: {e}" # 工具名 -> 实际函数 的映射表 TOOL_IMPLEMENTATIONS = { "get_now": lambda: json.dumps({"result": get_now()}), "calculate": lambda **kw: json.dumps({"result": calculate(kw["a"], kw["b"], kw["op"])}), "search_wikipedia": lambda **kw: json.dumps({"result": search_wikipedia(kw["query"])}), } # --------------------------------------------------------------------------- # 调用模型的辅助函数(纯标准库,OpenAI 兼容接口) # --------------------------------------------------------------------------- def chat_completion(messages, tools): """POST 到 OpenRouter 的 /chat/completions,返回 message 字典。""" url = BASE_URL + "/chat/completions" body = json.dumps( {"model": MODEL, "messages": messages, "tools": tools, "tool_choice": "auto"} ).encode("utf-8") req = urllib.request.Request( url, data=body, headers={ "Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json", "HTTP-Referer": "http://localhost", "X-Title": "My First AI Agent", }, method="POST", ) with urllib.request.urlopen(req, timeout=120) as resp: data = json.loads(resp.read().decode("utf-8")) return data["choices"][0]["message"] def run_agent(prompt: str, max_steps: int = 5): # 消息历史:第一句是系统提示,之后是用户的提问 messages = [ { "role": "system", "content": ( "你是一个乐于助人的 AI Agent。你可以调用工具来获取真实信息," "然后基于工具返回的结果回答用户。回答请使用中文。" ), }, {"role": "user", "content": prompt}, ] print(f"\n=== 用户问题: {prompt} ===") for step in range(1, max_steps + 1): print(f"\n▶ 第 {step} 步:调用模型 ...") msg = chat_completion(messages, TOOLS) messages.append(msg) # 保留模型这一步的发言/工具调用 # 模型没有要求调用工具 => 说明它给出了最终答案 tool_calls = msg.get("tool_calls") if not tool_calls: print("\n✅ 最终回答:") print(msg.get("content")) return msg.get("content") # 模型要求调用工具:逐个执行,把结果塞回对话 for call in tool_calls: fn = call["function"] name, args = fn["name"], json.loads(fn["arguments"] or "{}") print(f" 🔧 调用工具 [{name}],参数: {args}") result = TOOL_IMPLEMENTATIONS[name](**args) messages.append( { "role": "tool", "tool_call_id": call["id"], "content": result, } ) print("\n⚠ 达到最大步骤数,未能完成。") return None # --------------------------------------------------------------------------- # 入口 # --------------------------------------------------------------------------- if __name__ == "__main__": # 让中文/特殊字符在任何 Windows 终端都能正常打印 if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="replace") if not API_KEY: print("缺少 API Key。请先在官网 https://openrouter.ai 注册并获取密钥,") print("然后把它写到同目录的 .env 文件里:OPENROUTER_API_KEY=你的密钥") print("示例见 .env.example") raise SystemExit(1) print(f"使用模型: {MODEL}") questions = [ "现在几点钟了?今天是几月几号?", "帮我计算 12345 * 678 等于多少?", "用中文简单介绍一下“人工智能”。", ] # 你可以改成任意自己想问的中文问题,例如: # questions = ["今天的天气怎么样?", "帮我写一首关于秋天的五行诗"] for q in questions: run_agent(q) print("\n" + "-" * 60)从这个代码可以看出,一个Agent的核心作用就是那个for,不停的循环环境和大模型之间的交护,实现AI自动化。仅此而已了。。。
3 进阶:如何让 Agent 更实用?
如果希望构建能用于生产环境的复杂 Agent,可以在极简框架的基础上扩展以下模块:
1. 使用原生 Function Calling / Tool Call
上面的示例通过文本正则匹配解析 Action,容易因为格式错乱出错。主流 LLM API(OpenAI、Anthropic、DeepSeek)都支持原生的Tool Calling API,可以直接传 JSON Schema 格式的函数定义,由大模型原生地返回结构化的工具调用参数。
2. 增加持久化记忆 (Memory)
短期记忆:维护一个消息队列或滑动窗口,截断过长的历史 Context。
长期记忆:引入向量数据库(如 Chroma、Qdrant),将用户偏好或历史知识嵌入(Embedding)后按需检索(RAG)。
3. 多 Agent 协作 (Multi-Agent Systems)
对于极其复杂的工作流(如自动写软件工程、市场调研报告),单个 Agent 容易迷失。可以采用多 Agent 协作设计模式:
Supervisor 模式:由一个 Leader Agent 负责任务拆解分发,子 Agent 各司其职(如:写代码、测试、审查)。
Peer-to-Peer 模式:Agent 之间互相传达消息与协同谈判。
4 常用开源框架生态
自己编写底层代码能助你透彻理解原理。当逻辑变得庞大时,可以借助开源生态快速搭建:
LangGraph / LangChain:适合构建有向无环图(DAG)和复杂状态调度的企业级 Agent。
CrewAI:高度面向角色扮演和团队协作(Multi-Agent)的开箱即用框架。
AutoGen (Microsoft):微软主导的高扩展性多 Agent 讨论与代码生成框架。
Dify / FastGPT:支持可视化 Flow 画布调优的工作流 Agent 平台(适合低代码部署)。