很多人认为本地部署一个大模型，就实现 token 自由，就可以干活了，真的吗？

# 1. 安装并启动 Ollama，拉取本地模型（以 qwen2.5:7b 为例）
#    官网 https://ollama.com ，安装后执行：
ollama pull qwen2.5:7b

# 2. 安装 Python 依赖
pip install ollama requests

import os
import ollama
import requests
from requests.exceptions import RequestException


# 亿牛云代理（隧道代理 / 动态短效代理均可）
# 格式示例：http://<user>:<pass>@tunnel.16yun.cn:<port>
# 未配置时仍可运行，但在高频 / 批量采集场景下极易触发目标站点风控。
YINIU_PROXY = os.getenv("YINIU_PROXY")


def build_tools() -> list[dict]:
    """定义模型可调用的工具（OpenAI / Ollama 兼容的 function schema）。"""
    return [{
        "type": "function",
        "function": {
            "name": "web_fetch",
            "description": "抓取指定 URL 的网页正文，用于获取实时或外部信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "url": {"type": "string", "description": "目标网页地址 (http/https)"}
                },
                "required": ["url"],
            },
        },
    }]


def web_fetch(url: str, max_chars: int = 4000) -> str:
    """通过 requests 抓取网页；若配置 YINIU_PROXY 则经由代理出口。

    返回截断后的正文文本，供模型作为上下文消费。
    """
    proxies = (
        {"http": YINIU_PROXY, "https": YINIU_PROXY}
        if YINIU_PROXY else None
    )
    try:
        resp = requests.get(
            url,
            proxies=proxies,
            timeout=15,
            headers={"User-Agent": "Mozilla/5.0 (compatible; LocalAgent/1.0)"},
        )
        resp.raise_for_status()
        resp.encoding = resp.apparent_encoding or "utf-8"
        # 截断以控制上下文窗口占用；生产环境建议先做正文抽取（如 readability）
        return resp.text[:max_chars]
    except RequestException as e:
        return f"抓取失败：{e}"


def run_agent(question: str, model: str = "qwen2.5:7b", max_iter: int = 5) -> str:
    """基于 ReAct 思路的轻量 agent 循环。

    模型在每一轮可选择：(1) 调用工具并等待结果，或 (2) 直接给出最终答复。
    通过将 assistant 的工具调用与 tool 结果持续回填 messages，形成多轮推理链。
    """
    messages = [{"role": "user", "content": question}]
    tools = build_tools()

    for _ in range(max_iter):
        resp = ollama.chat(model=model, messages=messages, tools=tools)
        msg = resp["message"]

        if msg.get("tool_calls"):
            for call in msg["tool_calls"]:
                fn = call["function"]
                if fn["name"] == "web_fetch":
                    result = web_fetch(fn["arguments"].get("url", ""))
                    messages.append(msg)          # 保留模型的工具调用意图
                    messages.append({"role": "tool", "content": result})
            continue

        return msg["content"]

    return "（已达到最大迭代轮次，未得出最终结论，建议拆分问题或检查工具可用性）"


if __name__ == "__main__":
    q = "今天科技圈有哪些值得关注的动态？请联网查证后回答。"
    print(run_agent(q))

from sentence_transformers import SentenceTransformer
import numpy as np

embedder = SentenceTransformer("BAAI/bge-small-zh")

# 生产环境流水线：文档切分(chunks) -> 向量化 -> 写入向量库(faiss / chroma)
docs = ["公司 Q3 销售目标为 1200 万", "退款流程需经财务审批"]
doc_vecs = embedder.encode(docs, normalize_embeddings=True)

def search(query: str, top_k: int = 1) -> str:
    q_vec = embedder.encode([query], normalize_embeddings=True)[0]
    sims = doc_vecs @ q_vec          # 已归一化，点积即余弦相似度
    return docs[int(np.argmax(sims))]