基于ollama的deepseek基础入门代码示例

李昊哲-小课

2431人浏览 · 2025-05-21 12:07:31

李昊哲-小课 · 2025-05-21 12:07:31 发布

基于ollama的deepseek基础入门代码示例

Ollama 是一个开源的 AI 代理框架，用于与各种语言模型进行交互。
DeepSeek 是一个基于 Ollama 的模型或服务。
在 Ollama 的 API 中，
http://localhost:11434/api/chat和http://localhost:11434/api/generate是两个不同的接口，
它们的主要区别如下：

1.功能用途

• http://localhost:11434/api/chat

• 用途：主要用于对话式交互。它通常用于实现聊天机器人或类似的应用场景，能够处理自然语言输入，并以自然语言的形式返回回复。

• 特点：通常会维护对话状态（context），能够根据上下文生成连贯的对话内容。例如，用户可以与模型进行多轮对话，模型会记住之前的对话内容，从而生成更符合语境的回复。

• 应用场景：适合聊天机器人、客服系统、智能助手等需要与用户进行交互的场景。

• http://localhost:11434/api/generate

• 用途：主要用于文本生成任务。它接收一个输入提示（prompt），并根据提示生成文本内容。

• 特点：不维护对话状态，每次请求都是独立的。它更侧重于生成高质量的文本，而不是对话的连贯性。生成的文本可以是文章、故事、代码片段等。

• 应用场景：适合写作辅助、内容生成、代码生成等场景。

2.输入和输出

• http://localhost:11434/api/chat

• 输入：通常是一个用户的自然语言输入，可能包含上下文信息（如之前的对话历史）。

• 输出：返回一个自然语言的回复，通常是对话式的回答。

• http://localhost:11434/api/generate

• 输入：是一个提示（prompt），可以是一个句子、段落或更复杂的文本结构。

• 输出：返回生成的文本内容，内容的长度和格式根据提示和模型的配置而定。

3.内部处理逻辑

• http://localhost:11434/api/chat

• 内部会处理对话的上下文信息，可能使用特定的对话管理机制来维护对话状态。

• 生成的回复会考虑对话的连贯性和语境。

• http://localhost:11434/api/generate

• 主要基于输入的提示进行文本生成，不考虑对话状态。

• 更注重生成文本的质量和相关性。

4.配置和参数

• http://localhost:11434/api/chat

• 可能支持对话相关的参数，如对话历史长度、上下文窗口大小等。

• 参数配置更倾向于优化对话的流畅性和连贯性。

• http://localhost:11434/api/generate

• 支持文本生成相关的参数，如生成长度、温度（temperature）、Top-K、Top-P 等。

• 参数配置更侧重于控制生成文本的质量和多样性。

总结

• 如果你需要实现一个聊天机器人或需要与模型进行多轮对话，建议使用 http://localhost:11434/api/chat

• 如果你需要生成文本内容（如文章、故事、代码等），建议使用http://localhost:11434/api/generate

升级pip

python -m pip install --upgrade pip

安装依赖

pip install requests

chat对话式交互

通过 requests 向 ollama 发生http请求，ollama 调用本地 deepseek 进行推理并将推理结果返回

deepseek_chat.py

import requests

# 使用 requests.post 方法向 Ollama 的本地服务发送 POST 请求
response = requests.post(
  url="http://192.168.34.61:11434/api/chat",  # 指定请求的 URL，Ollama 服务运行在本地的 11434 端口
  json={  # 请求体的内容，以 JSON 格式发送
    "model": "deepseek-r1:1.5b",  # 指定要使用的模型名称和版本
    "messages": [  # 包含用户的消息
      {"role": "user", "content": "帮我写首打油诗送给临近毕业大大学生"}  # 用户发送的消息，角色为 "user"，内容为 "帮我写首打油诗送给临近毕业大大学生"
    ],
    "stream": False  # 设置是否以流式响应返回结果，这里设置为 False，表示一次性返回完整结果
  }
)


# 定义一个函数，用于提取思考过程和推理结果
def extract_thinking_and_response(data):
  """
  提取思考过程和推理结果。

  参数:
      data (dict): Ollama 返回的 JSON 数据。

  返回:
      tuple: 包含思考过程和推理结果的元组。
  """
  # 初始化思考过程和推理结果变量
  thinking_process = ""
  response_content = ""

  # 检查返回数据中是否包含消息内容
  if "message" in data and "content" in data["message"]:
    # 提取消息内容
    content = data["message"]["content"]

    # 检查是否包含思考过程
    if "<think>" in content and "</think>" in content:
      # 提取思考过程，去除标签
      start = content.find("<think>") + len("<think>")  # 找到 "<think>" 标签的起始位置
      end = content.find("</think>")  # 找到 "</think>" 标签的结束位置
      thinking_process = content[start:end].strip()  # 提取标签内的内容并去除首尾空格

    # 提取推理结果（即消息的其余部分）
    response_content = content.split("</think>")[-1].strip()  # 从 "</think>" 标签之后的内容中提取推理结果

  # 返回思考过程和推理结果
  return thinking_process, response_content


# 检查 HTTP 请求的返回状态码
if response.status_code != 200:  # 如果状态码不是 200（表示请求成功）
  print("请求失败，请重试！")  # 输出错误提示
else:
  print(response.json())  # 如果请求成功，打印返回的 JSON 数据
  thinking_process, response_content = extract_thinking_and_response(response.json())
  print("\n=== 模型思考过程 ===")
  print(thinking_process)
  print("\n=== 模型推理结果 ===")
  print(response_content)

generate 文本生成任务

deepseek_genetate.py

import requests

# 使用 requests.post 方法向 Ollama 的本地服务发送 POST 请求
response = requests.post(
  url="http://192.168.34.61:11434/api/generate",  # 指定请求的 URL，Ollama 服务运行在本地的 11434 端口
  json={  # 请求体的内容，以 JSON 格式发送
    "model": "deepseek-r1:1.5b",  # 指定要使用的模型名称和版本
    "prompt": "帮我写首打油诗送给临近毕业大大学生",  # 提示文本，用于引导模型生成内容
    "stream": False  # 设置是否以流式响应返回结果，这里设置为 False，表示一次性返回完整结果
  }
)


# 定义一个函数，用于提取思考过程和推理结果
def extract_thinking_and_response(data):
  """
  提取思考过程和推理结果。

  参数:
      data (dict): Ollama 返回的 JSON 数据。

  返回:
      tuple: 包含思考过程和推理结果的元组。
  """
  # 初始化思考过程和推理结果变量
  thinking_process = ""
  response_content = ""

  # 检查返回数据中是否包含响应内容
  if "response" in data:
    # 提取响应内容
    response = data["response"]

    # 检查是否包含思考过程
    if "<think>" in response and "</think>" in response:
      # 提取思考过程，去除标签
      start = response.find("<think>") + len("<think>")
      end = response.find("</think>")
      thinking_process = response[start:end].strip()

    # 提取推理结果（即响应的其余部分）
    response_content = response.split("</think>")[-1].strip()

  # 返回思考过程和推理结果
  return thinking_process, response_content


# 检查 HTTP 请求的返回状态码
if response.status_code != 200:  # 如果状态码不是 200（表示请求成功）
  print("请求失败，请重试！")  # 输出错误提示
else:
  print(response.json())  # 如果请求成功，打印返回的 JSON 数据
  thinking_process, response_content = extract_thinking_and_response(response.json())
  print("\n=== 模型思考过程 ===")
  print(thinking_process)
  print("\n=== 模型推理结果 ===")
  print(response_content)

解析 chat 和 generate 推理过程和推理结果函数

thinking_and_response.py

# 定义一个函数，用于提取思考过程和推理结果
def extract_langchain_thinking_and_response(data):
  """
    提取langchain思考过程和推理结果。

    参数:
        data (dict): Ollama 返回的 JSON 数据。

    返回:
        tuple: 包含思考过程和推理结果的元组。
   """
  # 初始化思考过程和推理结果变量
  thinking_process = ""
  response_content = ""

  # 检查是否包含思考过程
  if "<think>" in data and "</think>" in data:
    # 提取思考过程，去除标签
    start = data.find("<think>") + len("<think>")  # 找到 "<think>" 标签的起始位置
    end = data.find("</think>")  # 找到 "</think>" 标签的结束位置
    thinking_process = data[start:end].strip()  # 提取标签内的内容并去除首尾空格

  # 提取推理结果（即消息的其余部分）
  response_content = data.split("</think>")[-1].strip()  # 从 "</think>" 标签之后的内容中提取推理结果
  # 返回思考过程和推理结果
  return thinking_process, response_content


# 定义一个函数，用于提取思考过程和推理结果
def extract_chat_thinking_and_response(data):
  """
  提取对话交互思考过程和推理结果。

  参数:
      data (dict): Ollama 返回的 JSON 数据。

  返回:
      tuple: 包含思考过程和推理结果的元组。
  """
  # 初始化思考过程和推理结果变量
  thinking_process = ""
  response_content = ""

  # 检查返回数据中是否包含消息内容
  if "message" in data and "content" in data["message"]:
    # 提取消息内容
    content = data["message"]["content"]

    # 检查是否包含思考过程
    if "<think>" in content and "</think>" in content:
      # 提取思考过程，去除标签
      start = content.find("<think>") + len("<think>")  # 找到 "<think>" 标签的起始位置
      end = content.find("</think>")  # 找到 "</think>" 标签的结束位置
      thinking_process = content[start:end].strip()  # 提取标签内的内容并去除首尾空格

    # 提取推理结果（即消息的其余部分）
    response_content = content.split("</think>")[-1].strip()  # 从 "</think>" 标签之后的内容中提取推理结果

  # 返回思考过程和推理结果
  return thinking_process, response_content


# 定义一个函数，用于提取思考过程和推理结果
def extract_generate_thinking_and_response(data):
  """
  提取文本生成思考过程和推理结果。

  参数:
      data (dict): Ollama 返回的 JSON 数据。

  返回:
      tuple: 包含思考过程和推理结果的元组。
  """
  # 初始化思考过程和推理结果变量
  thinking_process = ""
  response_content = ""

  # 检查返回数据中是否包含响应内容
  if "response" in data:
    # 提取响应内容
    response = data["response"]

    # 检查是否包含思考过程
    if "<think>" in response and "</think>" in response:
      # 提取思考过程，去除标签
      start = response.find("<think>") + len("<think>")
      end = response.find("</think>")
      thinking_process = response[start:end].strip()

    # 提取推理结果（即响应的其余部分）
    response_content = response.split("</think>")[-1].strip()

  # 返回思考过程和推理结果
  return thinking_process, response_content


import re


def extract_ai_chat(response_data):
  """从DeepSeek API响应中提取并格式化推理结果"""
  try:
    # 从响应数据中提取模型返回的原始内容
    content = response_data["message"]["content"]

    # 使用正则表达式移除<think>标签及其包含的所有内容（非贪婪模式）
    cleaned_content = re.sub(r'<think>.*?</think>', '', content, flags=re.DOTALL).strip()

    # 使用正则表达式按###标题分割内容为多个章节
    sections = re.split(r'###\s*', cleaned_content)

    # 提取主体内容（第一个部分）
    main_content = sections[0].strip()
    chapters = []  # 用于存储各章节内容的列表

    # 处理每个章节（从第二个部分开始）
    for section in sections[1:]:
      if not section.strip():  # 跳过空部分
        continue
      # 分割章节标题和内容（按第一个换行符分割）
      lines = section.split('\n', 1)
      if len(lines) == 2:  # 确保有标题和内容
        title, content = lines
        chapters.append({  # 将章节信息添加到列表
          "title": title.strip(),  # 章节标题
          "content": content.strip()  # 章节内容
        })

    # 返回结构化的提取结果
    return {
      "main_content": main_content,  # 主体内容
      "chapters": chapters  # 章节列表
    }
  except Exception as e:
    print(f"提取内容时出错: {e}")  # 处理异常
    return None


def extract_ai_generate(response_data):
  """从DeepSeek API响应中提取并格式化推理结果"""
  try:
    # 从响应数据中提取模型返回的原始内容
    content = response_data["response"]

    # 使用正则表达式移除<think>标签及其包含的所有内容（非贪婪模式）
    cleaned_content = re.sub(r'<think>.*?</think>', '', content, flags=re.DOTALL).strip()

    # 使用正则表达式按###标题分割内容为多个章节
    sections = re.split(r'###\s*', cleaned_content)

    # 提取主体内容（第一个部分）
    main_content = sections[0].strip()
    chapters = []  # 用于存储各章节内容的列表

    # 处理每个章节（从第二个部分开始）
    for section in sections[1:]:
      if not section.strip():  # 跳过空部分
        continue
      # 分割章节标题和内容（按第一个换行符分割）
      lines = section.split('\n', 1)
      if len(lines) == 2:  # 确保有标题和内容
        title, content = lines
        chapters.append({  # 将章节信息添加到列表
          "title": title.strip(),  # 章节标题
          "content": content.strip()  # 章节内容
        })

    # 返回结构化的提取结果
    return {
      "main_content": main_content,  # 主体内容
      "chapters": chapters  # 章节列表
    }
  except Exception as e:
    print(f"提取内容时出错: {e}")  # 处理异常
    return None


def format_ai_history(extracted_data):
  """将提取的内容格式化为易读的文本"""
  if not extracted_data:  # 检查提取数据是否有效
    return "未能提取有效内容"

  # 初始化格式化文本，先添加主体内容
  formatted = extracted_data["main_content"] + "\n\n"

  # 遍历每个章节，按序号和标题格式添加
  for i, chapter in enumerate(extracted_data["chapters"], 1):
    formatted += f"### {chapter['title']}\n{chapter['content']}\n\n"

  return formatted  # 返回格式化后的完整文本


def save_to_file(content, filename="ai_history.md"):
  """将内容保存到文件"""
  try:
    # 以写入模式打开文件（自动创建或覆盖）
    with open(filename, 'w', encoding='utf-8') as file:
      file.write(content)  # 写入内容
    print(f"内容已保存到 {filename}")  # 提示保存成功
  except Exception as e:
    print(f"保存文件时出错: {e}")  # 处理保存异常

chat对话交互模式调用DeepSeek API获取AI发展历程的内容

import requests
import json


def call_deepseek_api(prompt):
  """调用DeepSeek API获取AI发展历程的内容"""
  try:
    # 发送POST请求到DeepSeek API
    response = requests.post(
      url="http://192.168.34.61:11434/api/chat",  # API端点URL
      json={
        "model": "deepseek-r1:1.5b",  # 指定使用的模型
        "messages": [
          {"role": "user", "content": prompt}  # 用户输入的问题
        ],
        "stream": False  # 不使用流式响应
      }
    )
    response.raise_for_status()  # 检查请求是否成功，不成功则抛出异常
    return response.json()  # 将响应内容解析为JSON并返回
  except requests.exceptions.RequestException as e:
    print(f"API请求错误: {e}")  # 处理请求异常
    return None
  except json.JSONDecodeError as e:
    print(f"JSON解析错误: {e}")  # 处理JSON解析异常
    return None


from thinking_and_response import extract_ai_chat, format_ai_history, save_to_file

# 主程序入口
if __name__ == "__main__":
  # 定义用户问题
  user_prompt = "请介绍一下人工智能的发展历程"

  # 调用API获取内容
  print("正在调用DeepSeek API获取内容...")
  response_data = call_deepseek_api(user_prompt)

  if response_data:  # 检查API响应是否成功
    # 提取内容
    print("正在提取并格式化内容...")
    extracted = extract_ai_chat(response_data)

    # 格式化内容
    formatted_content = format_ai_history(extracted)

    # 打印结果
    print("\n=== 人工智能发展历程 ===")
    print(formatted_content)

    # 保存到文件
    save_to_file(formatted_content)
  else:
    print("获取内容失败，请检查API连接和设置")

generate文本生成任务调用DeepSeek API获取AI发展历程的内容

import requests
import json


def call_deepseek_api(prompt):
  """调用DeepSeek API获取AI发展历程的内容"""
  try:
    # 发送POST请求到DeepSeek API
    response = requests.post(
      url="http://192.168.34.61:11434/api/generate",  # API端点URL
      json={
        "model": "deepseek-r1:1.5b",  # 指定使用的模型
        "prompt": prompt,
        "stream": False  # 不使用流式响应
      }
    )
    response.raise_for_status()  # 检查请求是否成功，不成功则抛出异常
    return response.json()  # 将响应内容解析为JSON并返回
  except requests.exceptions.RequestException as e:
    print(f"API请求错误: {e}")  # 处理请求异常
    return None
  except json.JSONDecodeError as e:
    print(f"JSON解析错误: {e}")  # 处理JSON解析异常
    return None


from thinking_and_response import extract_ai_generate, format_ai_history, save_to_file

# 主程序入口
if __name__ == "__main__":
  # 定义用户问题
  user_prompt = "请介绍一下人工智能的发展历程"

  # 调用API获取内容
  print("正在调用DeepSeek API获取内容...")
  response_data = call_deepseek_api(user_prompt)

  if response_data:  # 检查API响应是否成功
    # 提取内容
    print("正在提取并格式化内容...")
    extracted = extract_ai_generate(response_data)

    # 格式化内容
    formatted_content = format_ai_history(extracted)

    # 打印结果
    print("\n=== 人工智能发展历程 ===")
    print(formatted_content)

    # 保存到文件
    save_to_file(formatted_content)
  else:
    print("获取内容失败，请检查API连接和设置")