本地大模型验证

春风映月

244人浏览 · 2026-06-17 23:13:43

春风映月 · 2026-06-17 23:13:43 发布

Python验证ollama部署的本地大模型API

预定义

# 绕过代理访问 localhost
os.environ['NO_PROXY'] = 'localhost,127.0.0.1'

# Ollama API 配置
OLLAMA_BASE_URL = "http://localhost:11434"

# 禁用代理的 session
session = requests.Session()
session.trust_env = False  # 不使用系统代理
MODEL_NAME = "qwen2.5-coder:7b"

检查模型状态

def check_ollama_status():
    """检查 Ollama 服务是否运行"""
    try:
        response = session.get(OLLAMA_BASE_URL, timeout=5)
        if response.status_code == 200:
            print("Ollama 服务正在运行")
            return True
    except requests.exceptions.RequestException as e:
        print(f"Ollama 服务连接失败: {e}")
        return False
    return False

列出本地可用模型（ollama list）

def list_models():
    """列出本地可用的模型"""
    try:
        response = session.get(f"{OLLAMA_BASE_URL}/api/tags", timeout=10)
        if response.status_code == 200:
            data = response.json()
            models = data.get("models", [])
            if models:
                print("\n 本地可用模型:")
                for model in models:
                    print(f"  - {model['name']} (大小: {model['size'] / (1024**3):.2f} GB)")
                return models
            else:
                print(" 没有找到本地模型")
                return []
    except requests.exceptions.RequestException as e:
        print(f" 获取模型列表失败: {e}")
        return []

大模型的输出

非流式输出

模型在后台把整个回答全部写完，再一口气打包返回给你
像发短信。你发送问题后，界面会“卡住”或显示加载中，等待几秒后，一大段文字突然同时弹出来
等待时间长，首字延迟

流式输出

模型每生成一个词或一个字，就立刻通过网络通道传回给你的程序
像打字机。你发送问题后，界面几乎瞬间开始逐字显示内容
不需要用户长时间等待，就像目前常见大模型聊天时的效果

非流式输出聊天

"stream": False

def chat_simple(message: str, model: str = MODEL_NAME):
    """简单聊天（非流式输出）"""
    url = f"{OLLAMA_BASE_URL}/api/chat"

    # "role": "user" 表示这是用户消息（还可以有 "system" 系统提示和 "assistant" 助手回复）
    # "stream": False — 关键参数！设为 `False` 表流式输出，模型会处理完整个请求后一次性返回完整响应

    payload = {
        "model": model,
        "messages": [
            {"role": "user", "content": message}
        ],
        "stream": False
    }
    
    try:
        print(f"\n 模型: {model}")
        print(f" 问题: {message}")
        print("-" * 50)
        
        response = session.post(url, json=payload, timeout=120)
        if response.status_code == 200: #HTTP 200 表示成功，解析 JSON 响应
            data = response.json()
            # 响应结构为 `{"message": {"role": "assistant", "content": "回答内容"}, ...}`，通过链式 `.get()` 安全提取回答文本
            answer = data.get("message", {}).get("content", "")
            print(f" 回答:\n{answer}")
            return answer
        else:
            print(f" 请求失败: HTTP {response.status_code}")
            return None
    except requests.exceptions.RequestException as e:
        print(f" 请求异常: {e}")
        return None

响应的数据结构

{
  "model": "qwen2.5-coder:7b",
  "created_at": "2026-06-16T08:00:00.000000Z",
  "message": {
    "role": "assistant",
    "content": "你好！我是Qwen，一个AI助手。"
  },
  "done": true,
  "done_reason": "stop",
  "total_duration": 5000000000,
  "load_duration": 1000000000,
  "prompt_eval_count": 15,
  "prompt_eval_duration": 500000000,
  "eval_count": 30,
  "eval_duration": 3500000000
}

提取数据拆解

answer = data.get("message", {}).get("content", "")

第一步 `data.get("message", {})`

从响应字典 `data` 中取 `"message"` 键
如果 `"message"键存在，返回其值（即 `{"role": "assistant", "content": "回答内容"}`）
如果 `"message"` 键不存在（如 API 返回异常格式），返回默认值 `{}`（空字典），而不是抛出 `KeyError`

第二步 .get("content", "")

对上一步返回的字典取 `"content"` 键
如果 `"content"` 键存在，返回助手的回答文本
如果 `"content"` 键不存在，返回默认值 `""`（空字符串）

聊天中的角色

Chat API 的 `messages` 数组中，每条消息都有一个 `role` 字段，标识这条消息的"说话者"。三种角色各有明确分工

1. `system` — 系统角色（导演/规则制定者）

作用：设定模型的行为规则和人格，不参与对话内容本身
特点：
- 通常放在 `messages` 数组最前面，且只出现一次
- 用户看不到这条消息，但深刻影响模型的所有后续回答
- 类似于给演员的"剧本设定"——规定了角色该怎么演
- 可以控制：语言风格、专业领域、回答格式、禁止事项等

2. user — 用户角色（提问者）

作用：代表人类用户发送的消息/问题
特点：
- 就是用户的实际输入水电费
- 在一次对话中可以出多次多轮对话中用户多次提问）
- 模型会针对 `user` 消息生成回答

3. `assistant` — 助手角色（模型/回答者）

作用：代表模型生成的回答
特点：
- 是模型的回复，不是用户输入的
- 在多轮对话中，需要把历史的 `assistant` 回复也放入 `messages`，模型才能"记住"之前说过什么
- 如果不传历史 `assistant` 消息，模型就不知道自己之前回答了什么

流式输出聊天

主要看下流式接收的代码


def chat_stream(message: str, model: str = MODEL_NAME):
    """流式聊天（实时输出）"""
    url = f"{OLLAMA_BASE_URL}/api/chat"
    payload = {
        "model": model,
        "messages": [
            {"role": "user", "content": message}
        ],
        "stream": True
    }
    
    try:
        print(f"\n 模型: {model}")
        print(f" 问题: {message}")
        print("-" * 50)
        print(" 回答: ", end="", flush=True)
        
        response = session.post(url, json=payload, stream=True, timeout=120)
        if response.status_code == 200:
            for line in response.iter_lines():
                if line:
                    data = json.loads(line)
                    content = data.get("message", {}).get("content", "")
                    print(content, end="", flush=True)
            print("\n")
            return True
        else:
            print(f"\n 请求失败: HTTP {response.status_code}")
            return False
    except requests.exceptions.RequestException as e:
        print(f"\n 请求异常: {e}")
        return False

非流式 vs 流式的 HTTP 响应差异：

stream: False（非流式）
─────────────────────────
客户端发送请求
    ... 等待 10 秒 ...
服务器一次性返回完整 JSON：
{"message":{"content":"你好！我是Qwen，一个AI助手。"},"done":true}
    ↑ 一次性收到全部内容

stream: True（流式）
─────────────────────────
客户端发送请求
服务器立即开始返回数据，逐行推送：
{"message":{"content":"你"},"done":false}       ← 0.5秒后
{"message":{"content":"好"},"done":false}        ← 0.6秒后
{"message":{"content":"！"},"done":false}       ← 0.7秒后
{"message":{"content":"我是"},"done":false}     ← 1.0秒后
{"message":{"content":"Qwen"},"done":false}     ← 1.2秒后
...
{"message":{"content":""},"done":true}          ← 10秒后，结束
    ↑ 边生成边返回，用户能实时看到

打印收到的消息

print("💡 回答: ", end="", flush=True)

参数	作用	为什么需要
`" 回答: "`	打印提示文字	让用户知道回答开始了
`end=""`	末尾不换行	默认 `end="\n"` 会换行，这里需要后续内容接在同一行
`flush=True`	立即刷新输出缓冲区	默认 Python 输出有缓冲，可能不会立即显示

代码中有两个 stream=True

`payload` 中的 `"stream": True` 告诉 Ollama 服务器用流式格式返回
`session.post()` 的 `stream=True` — 告诉 requests模块读取响应，不要一次性下载整个响应体，而是保持连接，允许逐块读取。

stream=False（requests 默认）：
  客户端 → 服务器：给我数据
  服务器 → 客户端：[等10秒] 这是全部数据（100MB）
  客户端收到完整响应后才开始处理

stream=True：
  客户端 → 服务器：给我数据
  服务器 → 客户端：第1块数据
  客户端可以立即处理第1块
  服务器 → 客户端：第2块数据
  客户端可以立即处理第2块
  ...

for line in response.iter_lines():
`iter_lines()` 是 `requests` 库提供的方法，按行分割响应体，每次迭代返回一行。

Ollama 流式响应的格式是 NDJSON（Newline Delimited JSON）

{"message":{"content":"你"},"done":false}\n
{"message":{"content":"好"},"done":false}\n
{"message":{"content":"！"},"done":false}\n
\n
{"message":{"content":"我是"},"done":false}\n

iter_lines() 的处理过程：

第1次迭代: line = b'{"message":{"content":"你"},"done":false}'
第2次迭代: line = b'{"message":{"content":"好"},"done":false}'
第3次迭代: line = b'{"message":{"content":"！"},"done":false}'
第4次迭代: line = b''  ← 空行，被 ⑤ 过滤
第5次迭代: line = b'{"message":{"content":"我是"},"done":false}'
...

`data = json.loads(line)` 解析单行 JSON

生成文本

def generate(prompt: str, model: str = MODEL_NAME):
    """使用 generate API 生成文本"""
    url = f"{OLLAMA_BASE_URL}/api/generate"
    payload = {
        "model": model,
        "prompt": prompt,
        "stream": False
    }
    
    try:
        print(f"\n 模型: {model}")
        print(f" 提示词: {prompt}")
        print("-" * 50)
        
        response = session.post(url, json=payload, timeout=120)
        if response.status_code == 200:
            data = response.json()
            answer = data.get("response", "")
            print(f" 生成结果:\n{answer}")
            return answer
        else:
            print(f" 请求失败: HTTP {response.status_code}")
            return None
    except requests.exceptions.RequestException as e:
        print(f" 请求异常: {e}")
        return None

聊天和生成文本的区别

对应不同的 Ollama API 端点

	`chat_stream`	`generate`
API 端点	`/api/chat`	`/api/generate`
API 设计目的	对话式交互	纯文本补全/生成

请求数据的结构体不同

聊天中结构体 - 对话消息格式

payload = {
    "model": model,
    "messages": [                          # ← message 被放入 messages 列表
        {"role": "user", "content": message}  # ← 包装成结构化对话消息
    ],
    "stream": True
}

聊天可支持多轮对话

"messages": [
    {"role": "system", "content": "你是一个翻译助手"},
    {"role": "user", "content": "翻译这句话"},
    {"role": "assistant", "content": "上一轮回答"},
    {"role": "user", "content": "新的问题"}  # ← message 在这里
]

generate中的提示词直接传入,不支持对话历史，只是一个纯粹的"给前文，续后文"的补全接

payload = {
    "model": model,
    "prompt": prompt,    # ← prompt 直接作为顶层字段，无包装
    "stream": False
}

响应数据的结构体不同

聊天

{"message": {"role": "assistant", "content": "回答内容"}, ...}

生成

{"response": "生成的内容", ...}

交互聊天

多轮对话 + 流式输出 + 上下文记忆的交互式聊天

主要就是多轮对话的实现，每次对话都要把历史消息传一遍，这样大模型才会知道本轮会话的历史信息。

代码

def interactive_chat(model: str = MODEL_NAME):
    """交互式聊天模式"""
    print(f"\n{'='*50}")
    print(f" 交互式聊天模式 (模型: {model})")
    print("输入 'quit' 或 'exit' 退出")
    print(f"{'='*50}\n")
    
    messages = []
    
    while True:
        try:
            user_input = input("👤 你: ").strip()
            
            if user_input.lower() in ['quit', 'exit', 'q']:
                print(" 再见！")
                break
            
            if not user_input:
                continue
            
            # 添加用户消息到历史
            messages.append({"role": "user", "content": user_input})
            
            # 调用 API（流式输出）
            url = f"{OLLAMA_BASE_URL}/api/chat"
            payload = {
                "model": model,
                "messages": messages,
                "stream": True
            }
            
            print("🤖 助手: ", end="", flush=True)
            
            response = session.post(url, json=payload, stream=True, timeout=120)
            assistant_message = ""
            
            if response.status_code == 200:
                for line in response.iter_lines():
                    if line:
                        data = json.loads(line)
                        content = data.get("message", {}).get("content", "")
                        assistant_message += content
                        print(content, end="", flush=True)
                print("\n")
                
                # 添加助手回复到历史
                messages.append({"role": "assistant", "content": assistant_message})
            else:
                print(f"\n 请求失败: HTTP {response.status_code}")
                
        except KeyboardInterrupt:
            print("\n\n 再见！")
            break
        except Exception as e:
            print(f"\n 发生错误: {e}")

完整数据流图

用户输入 "我叫小明"
  │
  ▼
messages.append({"role": "user", "content": "我叫小明"})
messages = [user: "我叫小明"]
  │
  ▼
POST /api/chat  {messages: [user: "我叫小明"], stream: true}
  │
  ▼
流式接收 → 拼接 → assistant_message = "你好小明！很高兴认识你！"
  │
  ▼
messages.append({"role": "assistant", "content": "你好小明！..."})
messages = [user: "我叫小明", assistant: "你好小明！..."]
  │
  ▼
用户输入 "我叫什么名字？"
  │
  ▼
messages.append({"role": "user", "content": "我叫什么名字？"})
messages = [user: "我叫小明", assistant: "你好小明！", user: "我叫什么名字？"]
  │
  ▼
POST /api/chat  {messages: [完整历史], stream: true}
  │
  ▼
模型看到完整上下文 → 回答 "你叫小明" ✅

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

AI Agent编排：复杂工作流的自动化实现

在深入探讨Agent编排之前，我们首先需要明确什么是AI Agent。AI Agent是一个能够感知环境、做出决策并采取行动的自主实体。感知模块：接收和处理环境信息推理/决策模块：基于感知信息进行思考和决策行动模块：执行具体的操作（如调用工具、生成文本）记忆模块：存储历史信息和知识通信模块：与其他Agent或系统进行交互Agent编排（Agent Orchestration）是指对多个Agent的