4.1 环境准备与基础工具定义

配套代码:code/chapter4

在开始构建之前,我们需要先搭建好开发环境并定义一些基础组件。这能帮助我们在后续实现不同范式时,避免重复劳动,更专注于核心逻辑。

4.1.1 安装依赖库

本书的实战部分将主要使用 Python 语言,建议使用 Python 3.10 或更高版本。首先,请确保你已经安装了 openai 库用于与大语言模型交互,以及 python-dotenv 库用于安全地管理我们的 API 密钥。

在你的终端中运行以下命令:

pip install openai python-dotenv

4.1.2 配置 API 密钥

为了让我们的代码更通用,我们将模型服务的相关信息(模型ID、API密钥、服务地址)统一配置在环境变量中。

  1. 在你的项目根目录下,创建一个名为 .env 的文件。
  2. 在该文件中,添加以下内容。你可以根据自己的需要,将其指向 OpenAI 官方服务,或任何兼容 OpenAI 接口的本地/第三方服务。
  3. 如果实在不知道如何获取,可以参考 环境配置
# .env file
LLM_API_KEY="YOUR-API-KEY"
LLM_MODEL_ID="YOUR-MODEL"
LLM_BASE_URL="YOUR-URL"

我们的代码将从此文件自动加载这些配置。

4.1.3 封装基础 LLM 调用函数

为了让代码结构更清晰、更易于复用,我们来定义一个专属的LLM客户端类。这个类将封装所有与模型服务交互的细节,让我们的主逻辑可以更专注于智能体的构建。

import os
from openai import OpenAI
from dotenv import load_dotenv
from typing import List, Dict

# 加载 .env 文件中的环境变量
load_dotenv()

class HelloAgentsLLM:
    """
    为本书 "Hello Agents" 定制的LLM客户端。
    它用于调用任何兼容OpenAI接口的服务,并默认使用流式响应。
    """
    def __init__(self, model: str = None, apiKey: str = None, baseUrl: str = None, timeout: int = None):
        """
        初始化客户端。优先使用传入参数,如果未提供,则从环境变量加载。
        """
        self.model = model or os.getenv("LLM_MODEL_ID")
        apiKey = apiKey or os.getenv("LLM_API_KEY")
        baseUrl = baseUrl or os.getenv("LLM_BASE_URL")
        timeout = timeout or int(os.getenv("LLM_TIMEOUT", 60))
        
        if not all([self.model, apiKey, baseUrl]):
            raise ValueError("模型ID、API密钥和服务地址必须被提供或在.env文件中定义。")

        self.client = OpenAI(api_key=apiKey, base_url=baseUrl, timeout=timeout)

    def think(self, messages: List[Dict[str, str]], temperature: float = 0) -> str:
        """
        调用大语言模型进行思考,并返回其响应。
        """
        print(f"🧠 正在调用 {self.model} 模型...")
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=messages,
                temperature=temperature,
                stream=True,
            )
            
            # 处理流式响应
            print("✅ 大语言模型响应成功:")
            collected_content = []
            for chunk in response:
                if not chunk.choices:
                    continue
                content = chunk.choices[0].delta.content or ""
                print(content, end="", flush=True)
                collected_content.append(content)
            print()  # 在流式输出结束后换行
            return "".join(collected_content)

        except Exception as e:
            print(f"❌ 调用LLM API时发生错误: {e}")
            return None

# --- 客户端使用示例 ---
if __name__ == '__main__':
    try:
        llmClient = HelloAgentsLLM()
        
        exampleMessages = [
            {"role": "system", "content": "You are a helpful assistant that writes Python code."},
            {"role": "user", "content": "写一个快速排序算法"}
        ]
        
        print("--- 调用LLM ---")
        responseText = llmClient.think(exampleMessages)
        if responseText:
            print("\n\n--- 完整模型响应 ---")
            print(responseText)

    except ValueError as e:
        print(e)


>>>
--- 调用LLM ---
🧠 正在调用 xxxxxx 模型...
✅ 大语言模型响应成功:
快速排序是一种非常高效的排序算法...