返回文章列表

Python AI入门:调用大模型API实战

人工智能正在深刻改变软件开发的方式。大语言模型(LLM)的兴起让普通开发者也能在自己的项目中集成强大的AI能力。本文将从零开始,带你学会如何用Python调用大模型API,构建自己的AI应用。

一、AI与大模型概念简介

在动手写代码之前,我们先了解几个核心概念:

二、环境准备

我们需要安装Python的HTTP请求库来调用API。推荐使用requests库,它是Python中最流行的HTTP库。

安装依赖

# 使用pip安装requests库
pip install requests

# 如果你使用Python 3,也可以用pip3
pip3 install requests

配置API密钥

调用大模型API需要密钥。切勿将密钥硬编码在代码中,推荐使用环境变量来管理。

# config.py - 配置文件
import os

# 从环境变量读取API密钥
API_KEY = os.getenv("AI_API_KEY", "your-api-key-here")

# API基础地址(以智谱AI为例)
BASE_URL = "https://open.bigmodel.cn/api/paas/v4"

# 模型名称
MODEL_NAME = "glm-4-flash"
# 在终端中设置环境变量(Linux/Mac)
export AI_API_KEY="your-actual-api-key"

# Windows PowerShell
$env:AI_API_KEY="your-actual-api-key"

三、调用OpenAI兼容API(对话补全)

大多数国产大模型平台都兼容OpenAI的API格式。以下是一个最基本的调用示例:

# basic_chat.py - 基础对话示例
import requests
import json

def chat_completion(prompt, system_prompt="你是一个有帮助的AI助手。"):
    """
    发送对话请求到大模型API

    Args:
        prompt: 用户的输入文本
        system_prompt: 系统提示词,定义AI的角色和行为

    Returns:
        模型的回复文本
    """
    url = "https://open.bigmodel.cn/api/paas/v4/chat/completions"

    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {API_KEY}"
    }

    payload = {
        "model": MODEL_NAME,
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.7,  # 控制回复的随机性,0-1之间
        "max_tokens": 1024   # 最大生成token数量
    }

    try:
        response = requests.post(url, headers=headers, json=payload, timeout=30)
        response.raise_for_status()  # 检查HTTP错误

        data = response.json()
        reply = data["choices"][0]["message"]["content"]
        return reply

    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None

# 使用示例
if __name__ == "__main__":
    result = chat_completion("用Python写一个快速排序算法")
    print("AI回复:", result)

代码解析:

四、流式响应处理

普通请求需要等待模型完整生成后才能得到响应,用户体验不好。流式响应(Streaming)可以让模型边生成边返回,就像ChatGPT那样逐字显示。

# streaming_chat.py - 流式响应示例
import requests
import json

def chat_stream(prompt, system_prompt="你是一个有帮助的AI助手。"):
    """
    流式调用大模型API,逐字输出结果

    Args:
        prompt: 用户的输入文本
        system_prompt: 系统提示词
    """
    url = "https://open.bigmodel.cn/api/paas/v4/chat/completions"

    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {API_KEY}"
    }

    payload = {
        "model": MODEL_NAME,
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.7,
        "max_tokens": 1024,
        "stream": True  # 开启流式输出
    }

    try:
        response = requests.post(
            url,
            headers=headers,
            json=payload,
            timeout=60,
            stream=True  # requests库的流式请求
        )
        response.raise_for_status()

        # 逐行读取服务器返回的SSE(Server-Sent Events)数据
        for line in response.iter_lines():
            if line:
                line = line.decode("utf-8")

                # SSE格式以 "data: " 开头
                if line.startswith("data: "):
                    data_str = line[6:]  # 去掉 "data: " 前缀

                    # "data: [DONE]" 表示流式结束
                    if data_str == "[DONE]":
                        break

                    try:
                        data = json.loads(data_str)
                        content = data["choices"][0]["delta"].get("content", "")
                        if content:
                            print(content, end="", flush=True)
                    except json.JSONDecodeError:
                        continue

        print()  # 换行

    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")

# 使用示例
if __name__ == "__main__":
    print("AI: ", end="", flush=True)
    chat_stream("请用三句话介绍Python语言的特点")

流式响应的关键点:

  1. 请求体中加入"stream": True
  2. requests调用时设置stream=True
  3. 使用response.iter_lines()逐行读取SSE数据流
  4. 每行数据格式为data: {...},结束标志为data: [DONE]
  5. 增量数据在delta.content中(注意不是message.content)

五、多轮对话管理

真实的AI聊天需要记住上下文。实现多轮对话的核心思路是维护一个消息历史列表,每次请求时把完整的历史消息发给API。

# multi_turn_chat.py - 多轮对话管理
import requests

class ChatSession:
    """
    多轮对话管理器
    维护对话历史,支持上下文感知的连续对话
    """

    def __init__(self, system_prompt="你是一个有帮助的AI助手。", max_history=20):
        self.system_prompt = system_prompt
        self.max_history = max_history  # 最大保留的历史消息条数
        self.messages = []

        # 初始化系统消息
        if system_prompt:
            self.messages.append({
                "role": "system",
                "content": system_prompt
            })

    def chat(self, user_input, stream=False):
        """
        发送用户消息并获取AI回复

        Args:
            user_input: 用户输入
            stream: 是否使用流式输出

        Returns:
            AI的回复文本
        """
        # 添加用户消息到历史
        self.messages.append({
            "role": "user",
            "content": user_input
        })

        # 调用API
        url = "https://open.bigmodel.cn/api/paas/v4/chat/completions"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {API_KEY}"
        }
        payload = {
            "model": MODEL_NAME,
            "messages": self.messages,
            "temperature": 0.7,
            "max_tokens": 1024,
            "stream": stream
        }

        try:
            if stream:
                return self._handle_stream(url, headers, payload)
            else:
                response = requests.post(url, headers=headers, json=payload, timeout=30)
                response.raise_for_status()
                data = response.json()
                reply = data["choices"][0]["message"]["content"]

                # 保存AI回复到历史
                self.messages.append({
                    "role": "assistant",
                    "content": reply
                })
                self._trim_history()
                return reply

        except requests.exceptions.RequestException as e:
            # 请求失败时移除刚才添加的用户消息
            self.messages.pop()
            print(f"请求失败: {e}")
            return None

    def _handle_stream(self, url, headers, payload):
        """处理流式响应"""
        response = requests.post(url, headers=headers, json=payload,
                                 timeout=60, stream=True)
        response.raise_for_status()

        full_reply = ""
        for line in response.iter_lines():
            if line:
                line = line.decode("utf-8")
                if line.startswith("data: "):
                    data_str = line[6:]
                    if data_str == "[DONE]":
                        break
                    try:
                        import json
                        data = json.loads(data_str)
                        content = data["choices"][0]["delta"].get("content", "")
                        if content:
                            print(content, end="", flush=True)
                            full_reply += content
                    except json.JSONDecodeError:
                        continue

        print()
        # 保存完整回复到历史
        self.messages.append({"role": "assistant", "content": full_reply})
        self._trim_history()
        return full_reply

    def _trim_history(self):
        """修剪历史消息,防止超出上下文长度限制"""
        # 保留system消息 + 最近max_history条消息
        if len(self.messages) > self.max_history + 1:
            system_msg = self.messages[0]
            self.messages = [system_msg] + self.messages[-self.max_history:]

    def clear_history(self):
        """清除对话历史(保留系统提示)"""
        self.messages = self.messages[:1]
        print("对话历史已清除。")

    def show_history(self):
        """显示对话历史"""
        for msg in self.messages:
            role = msg["role"]
            content = msg["content"][:50] + "..." if len(msg["content"]) > 50 else msg["content"]
            print(f"[{role}] {content}")


# 使用示例
if __name__ == "__main__":
    session = ChatSession(
        system_prompt="你是一个编程导师,擅长用简洁的语言解释复杂概念。",
        max_history=10
    )

    print("=== AI编程导师(输入 quit 退出,clear 清除历史)===")
    while True:
        user_input = input("\n你: ").strip()
        if not user_input:
            continue
        if user_input.lower() == "quit":
            break
        if user_input.lower() == "clear":
            session.clear_history()
            continue

        print("AI: ", end="", flush=True)
        session.chat(user_input, stream=True)

六、Function Calling概念

Function Calling(函数调用/工具调用)让AI模型能够调用外部函数来获取信息或执行操作。例如,AI可以调用天气查询函数来回答"今天天气怎么样"。

# function_calling.py - Function Calling示例
import requests
import json

def function_calling_demo():
    """
    演示function calling:让AI调用本地函数获取当前时间
    """
    # 定义可用的工具(函数)
    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_current_time",
                "description": "获取当前的日期和时间",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "timezone": {
                            "type": "string",
                            "description": "时区,例如 Asia/Shanghai",
                            "default": "Asia/Shanghai"
                        }
                    },
                    "required": []
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "calculate",
                "description": "执行数学计算",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "expression": {
                            "type": "string",
                            "description": "数学表达式,例如 '2 + 3 * 4'"
                        }
                    },
                    "required": ["expression"]
                }
            }
        }
    ]

    # 本地函数实现
    def get_current_time(timezone="Asia/Shanghai"):
        from datetime import datetime
        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    def calculate(expression):
        try:
            result = eval(expression)
            return str(result)
        except Exception as e:
            return f"计算错误: {e}"

    # 函数名到实际函数的映射
    function_map = {
        "get_current_time": get_current_time,
        "calculate": calculate
    }

    url = "https://open.bigmodel.cn/api/paas/v4/chat/completions"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {API_KEY}"
    }

    user_input = "现在是几点了?再帮我算一下 123 * 456 等于多少"

    payload = {
        "model": MODEL_NAME,
        "messages": [
            {"role": "user", "content": user_input}
        ],
        "tools": tools,
        "tool_choice": "auto"
    }

    response = requests.post(url, headers=headers, json=payload, timeout=30)
    data = response.json()
    assistant_msg = data["choices"][0]["message"]

    # 检查AI是否决定调用函数
    if assistant_msg.get("tool_calls"):
        print("AI决定调用以下工具:")
        for tool_call in assistant_msg["tool_calls"]:
            func_name = tool_call["function"]["name"]
            func_args = json.loads(tool_call["function"]["arguments"])
            print(f"  函数: {func_name}, 参数: {func_args}")

            # 调用本地函数
            if func_name in function_map:
                result = function_map[func_name](**func_args)
                print(f"  结果: {result}")

                # 将函数结果返回给AI,让它生成最终回复
                payload["messages"].append(assistant_msg)
                payload["messages"].append({
                    "role": "tool",
                    "tool_call_id": tool_call["id"],
                    "content": result
                })

        # 第二次请求,让AI根据函数结果生成人类可读的回复
        second_response = requests.post(url, headers=headers,
                                        json=payload, timeout=30)
        second_data = second_response.json()
        final_reply = second_data["choices"][0]["message"]["content"]
        print(f"\n最终回复: {final_reply}")
    else:
        print(f"AI回复: {assistant_msg['content']}")

if __name__ == "__main__":
    function_calling_demo()

七、提示工程基础

提示工程(Prompt Engineering)是优化与大模型交互效果的关键技能。好的提示词能显著提升输出质量。

提示词设计原则

# prompt_engineering.py - 提示工程示例
def build_code_review_prompt(code, language="python"):
    """
    构建高质量的代码审查提示词
    """
    prompt = f"""请审查以下{language}代码,并从以下几个方面给出建议:

1. **代码质量**:命名是否规范、结构是否清晰
2. **潜在Bug**:是否存在逻辑错误或边界条件问题
3. **性能优化**:是否有可以改进的地方
4. **安全风险**:是否存在安全隐患

请用中文回答,每个方面给出具体建议和修改后的代码片段。

```{language}
{code}
```"""
    return prompt

def build_translation_prompt(text, target_lang="英文"):
    """
    构建翻译提示词,使用few-shot方式
    """
    prompt = f"""请将以下中文翻译为{target_lang}。

翻译要求:
- 保持原文的语气和风格
- 专业术语使用行业标准译法
- 不要添加原文没有的内容

示例:
输入: "这个bug导致内存泄漏"
输出: "This bug causes a memory leak"

输入: "我们需要重构这个模块"
输出: "We need to refactor this module"

现在请翻译以下内容:
输入: "{text}"
输出: """
    return prompt

# 使用示例
if __name__ == "__main__":
    code = """
def get_data(id):
    d = {}
    result = db.query("select * from users where id=" + id)
    for row in result:
        d[row[0]] = row[1]
    return d
"""
    prompt = build_code_review_prompt(code)
    print("代码审查提示词:")
    print(prompt)
    print("\n" + "="*50 + "\n")

    translation_prompt = build_translation_prompt("Python是一种解释型、面向对象的编程语言")
    print("翻译提示词:")
    print(translation_prompt)

八、实战案例:构建一个简单的AI聊天助手

综合运用以上知识,我们来构建一个功能完整的命令行AI聊天助手。它支持多轮对话、流式输出、历史记录和多种预设角色。

# ai_assistant.py - 完整的AI聊天助手
import requests
import json
import os
from datetime import datetime

# ========== 配置 ==========
API_KEY = os.getenv("AI_API_KEY", "your-api-key-here")
BASE_URL = "https://open.bigmodel.cn/api/paas/v4/chat/completions"
MODEL = "glm-4-flash"

# ========== 预设角色 ==========
PRESETS = {
    "default": {
        "name": "通用助手",
        "system": "你是一个有帮助的AI助手,能够回答各种问题。"
    },
    "coder": {
        "name": "编程专家",
        "system": "你是一个资深程序员,精通Python、JavaScript等多种编程语言。"
                   "回答代码问题时,请给出完整可运行的代码示例,并附上详细注释。"
    },
    "teacher": {
        "name": "英语老师",
        "system": "你是一位耐心的英语老师。当用户发送中文时,将其翻译为英文并解释"
                   "关键语法点;当用户发送英文时,纠正语法错误并给出改进建议。"
    },
    "writer": {
        "name": "写作助手",
        "system": "你是一位专业的写作助手,擅长撰写文章、润色文字、提供写作建议。"
                   "回复时注意文字的流畅性和逻辑性。"
    }
}

class AIAssistant:
    """完整的AI聊天助手"""

    def __init__(self):
        self.history = []
        self.system_prompt = PRESETS["default"]["system"]
        self.current_preset = "default"
        self.max_history = 20

    def set_preset(self, name):
        """切换预设角色"""
        if name in PRESETS:
            self.current_preset = name
            self.system_prompt = PRESETS[name]["system"]
            self.history = []
            print(f"\n已切换到角色: {PRESETS[name]['name']}")
        else:
            print(f"\n未知角色: {name}")
            print(f"可用角色: {', '.join(PRESETS.keys())}")

    def list_presets(self):
        """列出所有预设角色"""
        print("\n可用角色:")
        for key, preset in PRESETS.items():
            marker = " > " if key == self.current_preset else "   "
            print(f"{marker}{key}: {preset['name']}")

    def chat(self, user_input):
        """发送消息并获取流式响应"""
        self.history.append({"role": "user", "content": user_input})

        messages = [{"role": "system", "content": self.system_prompt}]
        messages.extend(self.history[-self.max_history:])

        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {API_KEY}"
        }
        payload = {
            "model": MODEL,
            "messages": messages,
            "temperature": 0.7,
            "max_tokens": 2048,
            "stream": True
        }

        try:
            response = requests.post(BASE_URL, headers=headers,
                                    json=payload, timeout=60, stream=True)
            response.raise_for_status()

            full_reply = ""
            for line in response.iter_lines():
                if line:
                    line = line.decode("utf-8")
                    if line.startswith("data: "):
                        data_str = line[6:]
                        if data_str == "[DONE]":
                            break
                        try:
                            data = json.loads(data_str)
                            content = data["choices"][0]["delta"].get("content", "")
                            if content:
                                print(content, end="", flush=True)
                                full_reply += content
                        except json.JSONDecodeError:
                            continue

            print()
            self.history.append({"role": "assistant", "content": full_reply})
            self._trim_history()

        except requests.exceptions.RequestException as e:
            self.history.pop()
            print(f"\n[错误] 请求失败: {e}")

    def _trim_history(self):
        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history:]

    def save_history(self, filename="chat_history.json"):
        """保存对话历史到文件"""
        data = {
            "timestamp": datetime.now().isoformat(),
            "preset": self.current_preset,
            "messages": self.history
        }
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        print(f"对话历史已保存到 {filename}")

    def show_help(self):
        """显示帮助信息"""
        print("""
=== 命令列表 ===
  /preset [名称]   - 切换角色(coder/teacher/writer/default)
  /presets         - 列出所有角色
  /save            - 保存对话历史
  /clear           - 清除对话历史
  /help            - 显示帮助
  /quit            - 退出程序
==================
        """)

    def run(self):
        """运行主循环"""
        print("=" * 50)
        print("  AI 聊天助手")
        print(f"  当前角色: {PRESETS[self.current_preset]['name']}")
        print("  输入 /help 查看命令列表")
        print("=" * 50)

        while True:
            user_input = input("\n你: ").strip()
            if not user_input:
                continue

            if user_input.startswith("/"):
                cmd = user_input.lower()
                if cmd.startswith("/preset "):
                    self.set_preset(cmd.split(maxsplit=1)[1].strip())
                elif cmd == "/presets":
                    self.list_presets()
                elif cmd == "/save":
                    self.save_history()
                elif cmd == "/clear":
                    self.history = []
                    print("对话历史已清除。")
                elif cmd == "/help":
                    self.show_help()
                elif cmd in ("/quit", "/exit", "/q"):
                    print("再见!")
                    break
                else:
                    print(f"未知命令: {user_input},输入 /help 查看帮助")
            else:
                print("\nAI: ", end="", flush=True)
                self.chat(user_input)


if __name__ == "__main__":
    assistant = AIAssistant()
    assistant.run()

总结

本文带你从零开始学习了Python调用大模型API的核心知识:

  1. 了解了AI大模型和API调用的基本概念
  2. 学会了使用requests库发送HTTP请求调用API
  3. 掌握了流式响应的处理方式,实现逐字输出效果
  4. 实现了多轮对话管理,支持上下文感知
  5. 了解了Function Calling的概念和用法
  6. 学习了提示工程的基础原则和技巧
  7. 综合运用以上知识构建了一个完整的AI聊天助手

下一步建议:

动手挑战

学到这里,不妨动手试一试以下练习,巩固你的理解:

  1. 基础练习:回顾本文核心概念,用自己的话总结关键知识点。
  2. 进阶实践:将文中的示例代码运行一遍,尝试修改参数观察变化。
  3. 拓展思考:想一想这个技术/方法还能应用在哪些场景中?

小贴士:遇到问题时,先独立思考,再查阅资料,最后请教他人——这是成长最快的学习方式。

赞赏支持

本文更新于 2026-08-22,环境 Python 3.12