企业官网建设流程全解析

day_3

Tool Calling

User | v +----------------+ | Agent | | | | LLM Reasoning | +----------------+ | 判断是否需要工具 | ------------------- | | | v v v ​ Search Database API Tool Tool Tool ​ \ | / \ | / ​ Tool Result ​ | v ​ LLM生成结果

RAG

Retrieval-Augment Generation---->检索增强生成，让LLM在回答问题的时候，先检索外部知识，再基于检索结果生成答案

RAG=搜索知识+LLM生成回答

用户问题 | v 知识库检索 | v 找到相关资料 | v LLM结合资料生成答案

需要RAG？

LLM的知识有限

• 知识截止

  • 训练出的数据没有涉及到

• 无法访问私有数据

• 容易产生幻觉

  • LLM可能不存在的接口

核心思想

LLM+外部知识库

LLM

• 理解问题

• 推理

• 组织语言

• 生成答案

RAG

• 找资料

• 提供上下文

用户问题 | v +---------------+ | Query理解 | +---------------+ | v +---------------+ | Retriever | | 检索模块 | +---------------+ | v +---------------+ | Vector DB | | 知识库 | +---------------+ | v 相关文档片段 | v +---------------+ | LLM | | 生成答案 | +---------------+ | v 最终回答

关键组件

Document Loader(文档加载)

• 读取数据（PDF,Word,Excel,HTML,数据库、API）

Text Splitter(文本切分)

• 文件可能太，不能一次全部放入模型

• Chunk大小设计---->300~1000 tokens(更具场景调整)

Embedding(向量化)--->核心

• 把文字--->数字向量

• 可以比较向量距离

  • A 如何重置密码？ B 忘记密码怎么办？ A向量≈B向量

Vector Database(向量数据库)

• 存储---->文本+向量+元数据

Retriever(检索器)

• 问题---->找到对应的Chunk

Generatot(生成器)

• 通常是LLM

RAG 是一种让大模型连接外部知识库的方法，通过检索相关信息增强上下文，再让 LLM 基于真实资料生成答案，从而解决知识过时、私有数据不可见和幻觉问题。

Python subprocess库

• 基本用法：执行命令

  • subprocess.run()

    • improt subprocess ​ result=subprocess.run(["ls","-l"]) ​ print(result)

    • completedProcess(args=['ls','-l'],returncode=0)

      • args:执行的命令

      • returncode:退出码（0表示成功）

• 获取命令输出

  • 默认情况下输出会直接显示在终端，如果想获取

    • import subprocess ​ result=subprocess.run( ["ls","-l"], capture_output=True, text=True ) print(result.stdout);

    • 输出

      total 20 -rw-r--r-- file.txt

    • 参数

      • capture_output=True

        • 捕获stdout和stderr

      • text=True

        • 返回字符串而不是bytes

• 执行Shell命令

  • subprocess.run("ls -l | grep txt",shell=True)

    • shell=True--->通过系统Shell执行

    • 支持管道|

    • 支持重定向 >

  • name = input() ​ subprocess.run( f"rm {name}", shell=True )

    • 注意不要把用户输入直接放进shell=True,可能会导致命令注入

    • 什么时候才必须写shell=True？

      在 Python 中，只有下面这几种极端情况你才不得不用shell=True（且前提是参数不是用户输入的）：

      1. 使用 Shell 管道符或重定向（如|、>、&&）：

         • 例如：subprocess.run("cat a.txt | grep hello", shell=True)

      2. 使用 Shell 的内置命令（Built-in commands）：

         • 在 Windows 下：dir、copy、cls等（它们不是.exe文件，而是cmd.exe内置的命令）。

         • 在 Linux 下：cd、export、alias等。

      3. 需要使用 Shell 的环境变量展开：

         • 例如：subprocess.run("echo $HOME", shell=True)

      总结：像ls、ping、git、python这类真实的外部可执行程序，直接用列表形式["ls", "/not_exist"]即可，不需要写shell=True。check=True则是负责在命令报错时及时拦截抛出 Python 异常。

• 检查命令执行结果

  • result = subprocess.run( ["python", "--version"], capture_output=True, text=True ) ​ if result.returncode == 0: print("成功") else: print("失败")

• 出错自动抛出异常

  • subprocess.run( ["ls", "/not_exist"], check=True )

    • 如果失败

      subprocess.CalledProcessError

• 捕获错误输出std::err

  • redsult=subprocess.run( ["python","error.py"], capture_output=True, text=True ) pritf(result.std.err)

• 启动长时间运行程序

  • 使用Popen()

  • inmport subprocess p=subprocess.Popen( ["python"，"server.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 强行杀死

• 与进程交互

  • p=subprocess.Popen( ["pyhon"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True ) out,err=p.communicate( "print('hello')\n" ) print(out)

• 管道Pipe

  • Linux

    • cat file.txt | grep hello

  • Python

    • import subprocess ​ # 1. 启动第一个进程 p1（读取文件） p1 = subprocess.Popen( ["cat", "file.txt"], stdout=subprocess.PIPE # 开启输出管道，准备把数据传给下一个进程 ) ​ # 2. 启动第二个进程 p2（过滤文本） p2 = subprocess.Popen( ["grep", "hello"], stdin=p1.stdout, # 👈 关键点：把 p1 的输出直接作为 p2 的输入！ stdout=subprocess.PIPE, # 开启输出管道，准备把最终结果拿回 Python text=True ) ​ # 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"], env=env )

• 设置超时时间

  • import subprocess ​ try: subprocess.run( ["python", "sleep.py"], timeout=5 ) ​ except subprocess.TimeoutExpired: print("程序超时")

• 常见的API对比

  • 方法	用途
    subprocess.run()	最常用，一次执行命令
    subprocess.Popen()	高级控制，持续运行
    subprocess.call()	老接口，不推荐
    subprocess.check_call()	执行失败抛异常
    subprocess.check_output()	获取输出

• 实操

  • 执行Git

    • import subprocess ​ result = subprocess.run( ["git", "status"], capture_output=True, text=True ) ​ print(result.stdout)

  • 批量执行系统命令

    • commands = [ ["echo", "hello"], ["python", "--version"], ["pip", "--version"] ] ​ for cmd in commands: result = subprocess.run( cmd, capture_output=True, text=True ) print(result.stdout)

temperature

model_dump

把模型对象----->Python字典（dict ）

from pydantic import BaseModel ​ class User(BaseModel): name: str age: int user=User(name="cjj",age=20) ​ data=user.model_dump()

结果

{ "name": "Alice", "age": 20 }

• 发JSON请求

  • request.post( "https://example.com", json=user.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_key=API_KEY ) ​ ​ # ========================== # 工具函数 # ========================== ​ def run_shell_command(command: str) -> str: """ 执行shell命令 """ ​ try: result = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=20 ) ​ 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( ​ model=MODEL_NAME, ​ messages=history, ​ tools=TOOLS, ​ temperature=0.7, ​ max_tokens=4096 ) ​ ​ 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/New+York

• 指定中文输出

  • curl wttr.in/Beijing?lang=zh-cn curl wttr.in/北京?lang=zh-cn

• 温度单位——>默认根据地区单位选择

  • 摄氏度

    • curl wttr.in/Tokyo?m

  • 华氏度

    • curl wttr.in/New+York?u

  • 风速m/s

    • curl wttr.in/Tokyo?M

• 控制显示内容

  • 只显示当前天气

    • curl wttr.in/Tokyo?0

  • 今天预报

    • curl wttr.in/Tokyo?1

  • 今天+明天预报

    • curl wttr.in/Tokyo?2

• 简洁一行输出（适合脚本）

  • curl wttr.in/Tokyo?format=3

  • Tokyo: ☀️ +28°C

  • 常用状态栏格式

    • curl wttr.in/Tokyo?format=“%c+%t+%h”

    • 参数	含义
      %c	天气图标
      %t	温度
      %h	湿度
      %w	风速
      %l	地点

• 关闭颜色

  • curl wttr.in/Tokyo?T

  • 纯文本输出

• 获取JSON

  • curl wttr.in/Tokyo?format=j1

• 图片天气图

  • 生成PNG wget wttr.in/Tokyo.png 透明背景 wget wttr.in/Tokyo_t.png

• 常用组合

  • 中文+摄氏度+简洁

    • curl “wttr.in/shanghai?m&lang=zh-cn&format=3”

quote

URL编码

• 将字符串转换成合适放在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?q=python%20%E6%95%99%E7%A8%8B

quote_plus()

from urllib.parse import quote, quote_plus ​ print(quote("hello world")) print(quote_plus("hello world")) hello%20world hello+world

• quote()----->空格变%20

  • URL路径

• quote_plus()---->空格变+

  • 查询参数

curl

• Client URL---->命令行工具

• 用于在终端里通过各种协议向服务器发送请求、获取数据

常见用途

访问网页

• curl https://example.com

• 把网页的HTML内容直接打印到终端

查看天气(wttr.in)

curl wttr.in/Tokyo

• curl:发起网络请求

• 服务器返回天气信息

下载文件

• 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

常用参数

curl和wget的区别

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、金山灵犀）：预置办公场景专用工具，如文档协同编辑、会议日程管理、内部数据合规查询等 。‌‌