day_3Tool CallingUser | v ---------------- | Agent | | | | LLM Reasoning | ---------------- | 判断是否需要工具 | ------------------- | | | v v v ​ Search Database API Tool Tool Tool ​ \ | / \ | / ​ Tool Result ​ | v ​ LLM生成结果Tool CallingRAG目的执行动作获取知识对象函数/API文档输出操作结果文本信息例子退款查询产品说明RAGRAGRetrieval-Augment Generation----检索增强生成让LLM在回答问题的时候先检索外部知识再基于检索结果生成答案RAG搜索知识LLM生成回答用户问题 | v 知识库检索 | v 找到相关资料 | v LLM结合资料生成答案需要RAGLLM的知识有限知识截止训练出的数据没有涉及到无法访问私有数据容易产生幻觉LLM可能不存在的接口核心思想LLM外部知识库LLM理解问题推理组织语言生成答案RAG找资料提供上下文用户问题 | v --------------- | Query理解 | --------------- | v --------------- | Retriever | | 检索模块 | --------------- | v --------------- | Vector DB | | 知识库 | --------------- | v 相关文档片段 | v --------------- | LLM | | 生成答案 | --------------- | v 最终回答关键组件Document Loader(文档加载)读取数据PDF,Word,Excel,HTML,数据库、APIText Splitter(文本切分)文件可能太不能一次全部放入模型Chunk大小设计----300~1000 tokens(更具场景调整)Embedding(向量化)---核心把文字---数字向量可以比较向量距离A 如何重置密码 B 忘记密码怎么办 A向量≈B向量Vector Database(向量数据库)存储----文本向量元数据Retriever(检索器)问题----找到对应的ChunkGeneratot(生成器)通常是LLM搜索RAG目标找到网页回答问题输出链接/文档自然语言答案理解弱强使用LLM通常没有有RAG 是一种让大模型连接外部知识库的方法通过检索相关信息增强上下文再让 LLM 基于真实资料生成答案从而解决知识过时、私有数据不可见和幻觉问题。Python subprocess库基本用法执行命令subprocess.run()improt subprocess ​ resultsubprocess.run([ls,-l]) ​ print(result)completedProcess(args[ls,-l],returncode0)args:执行的命令returncode:退出码0表示成功获取命令输出默认情况下输出会直接显示在终端如果想获取import subprocess ​ resultsubprocess.run( [ls,-l], capture_outputTrue, textTrue ) print(result.stdout);输出total 20 -rw-r--r-- file.txt参数capture_outputTrue捕获stdout和stderrtextTrue返回字符串而不是bytes执行Shell命令subprocess.run(ls -l | grep txt,shellTrue)shellTrue---通过系统Shell执行支持管道|支持重定向 name input() ​ subprocess.run( frm {name}, shellTrue )注意不要把用户输入直接放进shellTrue,可能会导致命令注入什么时候才必须写shellTrue在 Python 中只有下面这几种极端情况你才不得不用shellTrue且前提是参数不是用户输入的使用 Shell 管道符或重定向如|、、例如subprocess.run(cat a.txt | grep hello, shellTrue)使用 Shell 的内置命令Built-in commands在 Windows 下dir、copy、cls等它们不是.exe文件而是cmd.exe内置的命令。在 Linux 下cd、export、alias等。需要使用 Shell 的环境变量展开例如subprocess.run(echo $HOME, shellTrue)总结像ls、ping、git、python这类真实的外部可执行程序直接用列表形式[ls, /not_exist]即可不需要写shellTrue。checkTrue则是负责在命令报错时及时拦截抛出 Python 异常。检查命令执行结果result subprocess.run( [python, --version], capture_outputTrue, textTrue ) ​ if result.returncode 0: print(成功) else: print(失败)出错自动抛出异常subprocess.run( [ls, /not_exist], checkTrue )如果失败subprocess.CalledProcessError捕获错误输出std::errredsultsubprocess.run( [python,error.py], capture_outputTrue, textTrue ) pritf(result.std.err)启动长时间运行程序使用Popen()inmport subprocess psubprocess.Popen( [pythonserver.py] ) print(p.pid)Popen不会等待程序结束---异步非阻塞import subprocess ​ print(准备启动服务器...) # 启动子进程后Python 【立刻向下走】一毫秒都不停留 p subprocess.Popen([python, server.py]) ​ print(p.pid) # 立刻打印出 server.py 的进程 ID (PID) print(我是主程序我不需要等服务器结束我已经继续往下运行了) ​ # 此时你的主程序和 server.py 正在【同时运行】并发执行import subprocess import time ​ # 1. 启动后台程序不阻塞 p subprocess.Popen([python, server.py]) ​ # 2. 主程序做点别的事情 print(主程序正在忙别的事情...) time.sleep(5) ​ # 3. 轮询检查后台程序死了没有 if p.poll() is None: print(服务器还在正常运行中...) ​ # 4. 手动等待我现在想卡在这里等它结束了 p.wait() ​ # 5. 或者强行杀死它 # p.terminate() # 相当于发送 SIGTERM # p.kill() # 相当于 kill -9 强行杀死与进程交互psubprocess.Popen( [pyhon], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, textTrue ) out,errp.communicate( print(hello)\n ) print(out)管道PipeLinuxcat file.txt | grep helloPythonimport subprocess ​ # 1. 启动第一个进程 p1读取文件 p1 subprocess.Popen( [cat, file.txt], stdoutsubprocess.PIPE # 开启输出管道准备把数据传给下一个进程 ) ​ # 2. 启动第二个进程 p2过滤文本 p2 subprocess.Popen( [grep, hello], stdinp1.stdout, # 关键点把 p1 的输出直接作为 p2 的输入 stdoutsubprocess.PIPE, # 开启输出管道准备把最终结果拿回 Python textTrue ) ​ # 3. 允许 p1 在 p2 读取时接收 SIGPIPE 信号最佳实践补充防止死锁 p1.stdout.close() ​ # 4. 从 p2 拿到最终过滤后的输出 # p2.communicate() 返回一个元组 (stdout_data, stderr_data) # [0] 拿到的就是 stdout_data即最终的过滤结果 output p2.communicate()[0] ​ print(output)环境变量的设置import subprocess import os ​ env os.environ.copy() ​ env[MODE] test ​ subprocess.run( [python, app.py], envenv )设置超时时间import subprocess ​ try: subprocess.run( [python, sleep.py], timeout5 ) ​ except subprocess.TimeoutExpired: print(程序超时)常见的API对比方法用途subprocess.run()最常用一次执行命令subprocess.Popen()高级控制持续运行subprocess.call()老接口不推荐subprocess.check_call()执行失败抛异常subprocess.check_output()获取输出实操执行Gitimport subprocess ​ result subprocess.run( [git, status], capture_outputTrue, textTrue ) ​ print(result.stdout)批量执行系统命令commands [ [echo, hello], [python, --version], [pip, --version] ] ​ for cmd in commands: result subprocess.run( cmd, capture_outputTrue, textTrue ) print(result.stdout)temperaturemodel_dump把模型对象-----Python字典dict from pydantic import BaseModel ​ class User(BaseModel): name: str age: int userUser(namecjj,age20) ​ datauser.model_dump()结果{ name: Alice, age: 20 }发JSON请求request.post( https://example.com, jsonuser.model_dump() )存数据库db.insert(user.model_dump()) ​调试和打印print(user.model_dump())适用场景mudle_dump():把数据传给Python的其他函数使用数据库驱动支持直接接收Python字典mudel_dump_json把数据直接写成.json文件需要写入Redis(Redis只能存字符串)模型包含datetime、UUID等复杂最短准备发往HTTP API代码from zai import ZhipuAiClient from dotenv import load_dotenv ​ import os import json import subprocess ​ ​ load_dotenv() ​ ​ API_KEY os.getenv(ZHIPUAI_API_KEY) MODEL_NAME os.getenv(ZHIPUAI_MODEL_NAME) ​ ​ client ZhipuAiClient( api_keyAPI_KEY ) ​ ​ # # 工具函数 # ​ def run_shell_command(command: str) - str: 执行shell命令 ​ try: result subprocess.run( command, shellTrue, capture_outputTrue, textTrue, timeout20 ) ​ if result.stdout: return result.stdout ​ return result.stderr ​ except Exception as e: return f执行失败: {str(e)} ​ ​ ​ # # Function Calling 描述 # ​ TOOLS [ { type: function, function: { name: run_shell_command, description: 在Linux终端执行shell命令并返回结果, parameters: { type: object, properties: { command: { type: string, description: 需要执行的shell命令 } }, required: [ command ] } } } ] ​ ​ ​ # # System Prompt # ​ system_prompt # 角色设定 你是大内太监总管侍奉皇上多年。 ​ # 行为要求 ​ 1. 必须称呼用户为“皇上”。 ​ 2. 自称“奴才”。 ​ 3. 回复必须以 奉天承运皇帝诏曰 ​ 作为开头。 ​ 4. 全程使用中文。 ​ 5. 如果需要执行终端操作调用提供的工具。 ​ ​ history [ { role: system, content: system_prompt } ] ​ ​ ​ # # 对话循环 # ​ while True: ​ try: ​ user_input input(\n皇上: ).strip() ​ ​ if not user_input or user_input 退下: print( 总管: 奉天承运皇帝诏曰 奴才告退恭候皇上再次召见。 ) break ​ ​ history.append( { role:user, content:user_input } ) ​ ​ while True: ​ ​ response client.chat.completions.create( ​ modelMODEL_NAME, ​ messageshistory, ​ toolsTOOLS, ​ temperature0.7, ​ max_tokens4096 ) ​ ​ message response.choices[0].message ​ ​ # # 有工具调用 # ​ if message.tool_calls: ​ ​ history.append( { role:assistant, content:message.content, tool_calls:[ { id:tc.id, type:function, function:{ name:tc.function.name, arguments:tc.function.arguments } } for tc in message.tool_calls ] } ) ​ ​ for tool_call in message.tool_calls: ​ ​ name tool_call.function.name ​ ​ if name run_shell_command: ​ ​ args json.loads( tool_call.function.arguments ) ​ ​ command args[command] ​ ​ print( f\n执行命令: {command} ) ​ ​ result run_shell_command( command ) ​ ​ history.append( { role:tool, tool_call_id:tool_call.id, content:result } ) ​ ​ # 工具执行完继续让模型总结 continue ​ ​ ​ # # 普通回答 # ​ reply message.content ​ ​ # 防止模型漏掉固定前缀 prefix 奉天承运皇帝诏曰 ​ if not reply.startswith(prefix): reply prefix reply ​ ​ print( \n总管:, reply ) ​ ​ history.append( { role:assistant, content:reply } ) ​ ​ break ​ ​ ​ except Exception as e: ​ print( 发生错误:, e )Wttr.in基本用法查看当前位置天气curl wttr.in查看指定城市curl wttr.in/yanan城市名中有空格用curl wttr.in/NewYork指定中文输出curl wttr.in/Beijing?langzh-cn curl wttr.in/北京?langzh-cn温度单位——默认根据地区单位选择摄氏度curl wttr.in/Tokyo?m华氏度curl wttr.in/NewYork?u风速m/scurl wttr.in/Tokyo?M控制显示内容只显示当前天气curl wttr.in/Tokyo?0今天预报curl wttr.in/Tokyo?1今天明天预报curl wttr.in/Tokyo?2简洁一行输出适合脚本curl wttr.in/Tokyo?format3Tokyo: ☀️ 28°C常用状态栏格式curl wttr.in/Tokyo?format“%c%t%h”参数含义%c天气图标%t温度%h湿度%w风速%l地点关闭颜色curl wttr.in/Tokyo?T纯文本输出获取JSONcurl wttr.in/Tokyo?formatj1图片天气图生成PNG wget wttr.in/Tokyo.png 透明背景 wget wttr.in/Tokyo_t.png常用组合中文摄氏度简洁curl “wttr.in/shanghai?mlangzh-cnformat3”quoteURL编码将字符串转换成合适放在URL里面的格式from urllib.parse import quote ​ text hello world result quote(text) ​ print(result) hello%20word from urllib.parse import quote ​ url https://example.com/search?q quote(python 教程) ​ print(url) https://example.com/search?qpython%20%E6%95%99%E7%A8%8Bquote_plus()from urllib.parse import quote, quote_plus ​ print(quote(hello world)) print(quote_plus(hello world)) hello%20world helloworldquote()-----空格变%20URL路径quote_plus()----空格变查询参数curlClient URL----命令行工具用于在终端里通过各种协议向服务器发送请求、获取数据常见用途访问网页curl https://example.com把网页的HTML内容直接打印到终端查看天气(wttr.in)curl wttr.in/Tokyocurl:发起网络请求服务器返回天气信息下载文件curl -0 https://example/file.zip-0-----保持服务器上的文件名查看HTTP请求结果curl -I https://example.com-I-----只显示响应头HTTP/2 200 content-type: text/html server: nginx发送POST请求curl -X POST https://apoi.example.com/login常用参数参数作用-o 文件名保存输出到文件-O按服务器文件名保存-I只看响应头-L跟随跳转-s静默模式不显示进度-v显示详细通信过程-X GET/POST指定请求方法-d发送数据参数记忆-vverbose 调试-IHTTP HEAD-Llocation 跳转-Xrequest 方法-Hheader-ddata-Fform 上传-ooutput 文件-Ooriginal 文件名-uuser 登录-bcookie-xproxy-k忽略 SSL-ssilent-wwrite 输出curl和wget的区别curlwget主要用途请求、API测试下载文件支持协议很多主要 HTTP/FTPAPI 调试⭐⭐⭐⭐⭐⭐⭐递归下载网页较弱强wget web get ↓ 获取网页/文件 ​ ​ curl client URL ↓ 操作 URL 请求数据 调 API作业Agent 应用开发中高频调用的工具核心为‌联网搜索、代码执行、文件/数据操作、API 集成及终端控制‌五大类‌联网与信息获取‌通用搜索引擎Web Search、垂直领域检索金融/法律/学术数据库、网页抓取Scraping、实时新闻/天气/股价查询 。‌计算与代码执行‌Python/SQL 代码解释器Code Interpreter、计算器、数据清洗与统计分析脚本、沙箱环境执行 。文件与文档处理‌本地/云存储读写Read/Write、PDF/Word/Excel 解析与生成、OCR 识别、格式转换 。外部系统/API 集成‌企业 ERP/CRM/OA 系统接口、邮件发送、日历调度、消息推送Slack/钉钉/企微、数据库直连查询 。‌终端与环境操控‌Shell 命令执行、Git 版本控制、容器/Docker 管理、远程服务器部署、GUI 自动化操作 。‌‌企业级商用‌如腾讯 WorkBuddy、金山灵犀预置办公场景专用工具如文档协同编辑、会议日程管理、内部数据合规查询等 。‌‌