前四篇博客,我们像剥洋葱一样,把 Agent 的大脑袋(LLM)、手脚(Tools)、记忆(Memory)和灵魂(ReAct 规划)全给扒得底朝天。
理论听了这么多,是时候来点**“带泥土气息的工程实战”**了。
今天,咱们扔掉 LangChain、LlamaIndex 那些沉重的框架包袱。就用最纯粹的 Python 代码,手撸一个带多步推理、工具调用和自我运转闭环的 Agent。
哪怕你刚学会写 Python 的if-else,只要跟着这套逻辑走,你也能立刻明白市面上那些估值千万的 AI 创业项目,底层到底是怎么跑起来的。
一、 我们要造一个什么“打工人”?
为了化抽象为具象,我们要手搓一个**“出行小助手”**。
它只有两个极简的超能力(工具):
查天气:告诉它城市,它返回天气情况。
算数:给它一个数学公式,它给你算结果。
任务目标:当用户问“我在北京,想去上海出差,帮我查查上海的天气,顺便算算我带 5000 块钱,扣掉 1200 机票后还剩多少钱?”时,这个 Agent 必须能自主思考、按顺序调用工具、并给出最终回答。
二、 核心灵魂:老板的“SOP(标准作业程序)”
不用框架怎么让 LLM 听话?全靠 System Prompt(系统提示词)。 在这个实战里,我们必须强制 LLM 扮演一个“流水线工人”,每次只准输出特定格式的 JSON,绝对不能说废话。
咱们先把这份“SOP”写好:
Python
SYSTEM_PROMPT = """ 你是一个全能的 AI 助手。你可以使用以下工具来解决用户的问题: 【可用工具列表】 1. get_weather: 获取指定城市的天气。参数格式:{"city": "城市名"} 2. calculator: 计算数学表达式。参数格式:{"expression": "数学公式"} 【你的工作流程要求】 你必须且只能以 JSON 格式输出你的决策!绝不能输出任何多余的解释文字! JSON 必须严格包含以下 4 个字段: - "thought": 你的内心思考过程(下一步该干嘛?) - "action": 你要调用的工具名(如果任务已完成,请填 "FINISH") - "action_args": 你传给工具的参数字典(如果 action 是 FINISH,这里留空) - "final_answer": 如果任务完成,这里填最终回答(否则留空) 【示例】 {"thought": "用户问天气,我需要调用 get_weather 工具", "action": "get_weather", "action_args": {"city": "北京"}, "final_answer": ""} """有了这份极度严苛的 SOP,大模型就从一个“聊天机器”,变成了一个“决策引擎”。
三、 完整代码:100 行手捏 Agent
找个新建的 Python 文件,把下面的代码贴进去。这几乎就是当今所有 Agent 编排引擎最底层的骨架(你需要自备一个能调用的 LLM API,这里以最通用的 OpenAI 接口格式为例):
Python
import json from openai import OpenAI # 1. 准备大模型客户端 (请替换为你自己的 API_KEY 和 BASE_URL) client = OpenAI(api_key="sk-your-api-key", base_url="https://api.your-llm-provider.com/v1") # ========================================== # 第一步:外包团队(定义工具集) # ========================================== def get_weather(city: str) -> str: """模拟查询天气的工具""" print(f" [🛠️ 动作执行] 正在查询 {city} 的天气...") # 真实工程中,这里会调用第三方气象 API mock_data = {"北京": "晴天,20度", "上海": "暴雨,15度"} return mock_data.get(city, "未知天气") def calculator(expression: str) -> str: """模拟计算器工具""" print(f" [🛠️ 动作执行] 正在计算: {expression}") try: # 真实工程中绝不能直接用 eval,极其危险!这里仅作演示 return str(eval(expression)) except Exception as e: return f"计算错误: {str(e)}" # 工具路由表(Dispatcher) tools_map = { "get_weather": get_weather, "calculator": calculator } # ========================================== # 第二步:核心闭环(Agent 运行引擎) # ========================================== def run_agent(user_query: str): print(f"\n👨💻 用户提问: {user_query}\n" + "="*50) # 初始化短期记忆(Context Window) messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user_query} ] # 强制加上 max_steps 限制,防止 Agent 陷入死循环! max_steps = 5 for step in range(max_steps): print(f"\n--- 第 {step + 1} 轮思考 ---") # 1. 大脑决策:调用大模型 response = client.chat.completions.create( model="gpt-4o-mini", # 或者其他便宜好用的模型 messages=messages, temperature=0.1 # 调低温度,让机器人的决策更稳定 ) raw_reply = response.choices[0].message.content # 泥土气息的工程操作:清理 LLM 返回的 Markdown 标记 clean_json_str = raw_reply.replace('```json', '').replace('```', '').strip() try: decision = json.loads(clean_json_str) except json.JSONDecodeError: print(f"❌ 致命错误:大模型没有返回标准 JSON!内容是:\n{raw_reply}") break # 2. 打印 Agent 的内心戏 print(f"🧠 思考 (Thought): {decision.get('thought')}") # 3. 判断是否大功告成 action = decision.get("action") if action == "FINISH": print("\n🎉 任务完成!") print(f"🤖 最终回答: {decision.get('final_answer')}") break # 4. 执行动作 (Action) print(f"⚙️ 决定调用工具 (Action): {action}, 参数: {decision.get('action_args')}") if action in tools_map: # 拿到对应的函数去执行 func = tools_map[action] args = decision.get("action_args", {}) # 假设参数只有一个,这里做个简单拆包 observation = func(list(args.values())[0]) print(f"👀 观察结果 (Observation): {observation}") else: observation = f"错误:找不到名为 {action} 的工具" print(f"👀 观察结果 (Observation): {observation}") # 5. 将结果写入短期记忆,进入下一轮死循环! messages.append({"role": "assistant", "content": raw_reply}) # 存入它自己的思考 messages.append({"role": "user", "content": f"工具执行结果:{observation}"}) # 存入客观事实 else: print("\n⚠️ 警告:达到了最大思考步数,强制停机!") # ========================================== # 启动测试 # ========================================== if __name__ == "__main__": run_agent("我准备去上海出差。帮我查一下天气。另外我带了 5000 块钱,机票花了 1200,我还剩多少预算?")四、 见证奇迹:程序跑起来是什么样?
如果你运行这段代码,你会看到控制台打印出极其精彩的**“破案过程”**:
Plaintext
👨💻 用户提问: 我准备去上海出差。帮我查一下天气。另外我带了 5000 块钱,机票花了 1200,我还剩多少预算? ================================================== --- 第 1 轮思考 --- 🧠 思考 (Thought): 用户需要查询上海的天气,并且需要计算剩余的预算。我应该先查询上海的天气。 ⚙️ 决定调用工具 (Action): get_weather, 参数: {'city': '上海'} [🛠️ 动作执行] 正在查询 上海 的天气... 👀 观察结果 (Observation): 暴雨,15度 --- 第 2 轮思考 --- 🧠 思考 (Thought): 已经获取了上海的天气。接下来需要计算剩余预算,用 5000 减去 1200。 ⚙️ 决定调用工具 (Action): calculator, 参数: {'expression': '5000 - 1200'} [🛠️ 动作执行] 正在计算: 5000 - 1200 👀 观察结果 (Observation): 3800 --- 第 3 轮思考 --- 🧠 思考 (Thought): 天气和预算都计算完毕。现在我可以给用户一个最终答复了。 ⚙️ 决定调用工具 (Action): FINISH, 参数: {} 🎉 任务完成! 🤖 最终回答: 上海的天气是暴雨,15度。您的剩余预算是 3800 元。出行请注意带好雨具哦!你看,没有任何黑魔法!大模型并没有亲自去查天气,也没有亲自去算数。它只是在你设定的while/for循环里,不断地“思考 → 下发工单 → 等待你把结果塞回来”,直到收集齐所有拼图,交出最终答卷。
五、 带着泥土气息的工程复盘
如果你要把这 100 行代码用到生产环境里赚钞票,你立刻会面临几个硬核问题,这也是真实开发中极其痛的领悟:
Json 解析的血泪史:代码里那句
replace('```json', '')看似随意,却是无数工程师用凌晨的头发换来的。即便你千叮咛万嘱咐,LLM 依然有极大概率在 JSON 外面包裹 Markdown 标记,甚至在 JSON 里塞入未转义的换行符导致loads崩溃。在生产环境,你必须引入更强健的解析库(比如 Pydantic 或专门的 JSON Repair 工具)。死循环的刹车片:必须要有
max_steps。有时工具一直报错(比如网络不通),Agent 会不厌其烦地调用同一个工具几百次,把你的 API 额度瞬间刷干。安全的边界:我们为了演示使用了
eval()来做计算器,这在工程上是绝对禁止的,一旦用户输入import os; os.system("rm -rf /"),你的服务器就灰飞烟灭了。真实的工具必须要做极度严格的入参沙箱校验。
结语
恭喜你!到这里,你已经亲手拼装了一台 Agent 引擎。
回望这五篇博客,从抽象的概念拆解、到 ReAct 闭环的理论、再到长短记忆的管理,最后到今天这 100 行清晰可运行的代码。你会发现,打破 AI 幻觉、让大模型真正落地做事,靠的从来不是什么玄之又玄的魔法,而是极其清晰的系统架构和扎实的工程手艺。
现在,底座已经建好。接下来,你想给它装上连接数据库的手,还是连通微信服务器的腿?舞台交给你了!