02 · Python 环境与第一个 Agent

本章目标：5 分钟内跑通第一个 LLM 调用，并在不依赖任何框架的前提下，亲手写出 Agent 的核心循环。

本章你将学到

搭好一套国内可直接联网调用的 Python LLM 开发环境；
完成第一次 LLM 调用、流式输出、结构化 JSON 输出；
亲手实现 Agent 的本质——"模型决策 → 宿主执行工具 → 结果回填 → 继续决策"的循环。

前置知识

了解什么是 token、上下文窗口、temperature（不熟先看第 01 章）；
会基本的命令行操作即可，不需要 Python 经验。

2.1 为什么先跑通"最小闭环"

老 Java 工程师上手新技术有个习惯：先写一个能跑的 HelloWorld，再谈架构。LLM 开发也一样——先跑通一次真实的模型调用，你对"上下文、token、延迟、成本"的体感就全回来了。

这一章我们不谈架构、不谈框架，只做三件事：环境、调用、手写 Agent 循环。跑通之后，后续每一章的示例你都能照搬这个环境。

2.2 Python 环境搭建

2.2.1 安装 Python 3.10+

到 python.org 下载安装 3.10 以上的版本。安装时务必勾选 "Add python.exe to PATH"。

验证：

python --version
# 期望输出类似：Python 3.12.x

Java 工程师类比：python 命令 ≈ java 命令；pip ≈ mvn；虚拟环境 venv ≈ Maven 里每个模块独立依赖的隔离效果。

2.2.2 创建虚拟环境

每个教程项目都建独立虚拟环境，避免污染全局（类似 Maven 每个工程独立 target）：

# 在教程根目录下
python -m venv .venv

# Windows（PowerShell）
.venv\Scripts\Activate.ps1

# macOS / Linux
source .venv/bin/activate

激活后命令行会出现 (.venv) 前缀。退出环境用 deactivate。

2.2.3 requirements.txt

在项目根目录新建 requirements.txt：

langchain==0.3.*
langchain-openai==0.3.*
langchain-community==0.3.*
python-dotenv
pydantic==2.*

安装：

pip install -r requirements.txt

版本说明：本系列统一用 langchain 0.3.x 稳定版（该版本生态与文档最齐全）。LangChain 已有 1.x，接口有变化，本系列代码按 0.3 写作，遇到新版本以官方文档为准，概念不变。

2.3 国内环境配置：OPENAI_BASE_URL

你可能没有 OpenAI 官方账号，或者网络访问受限。这不是障碍：OpenAI 的 API 协议已经成为事实标准，国内主流模型（DeepSeek、通义千问 Qwen、智谱 GLM、Kimi 等）都提供 "OpenAI 兼容端点"，只需换 base_url 和 model 名就能用同一套代码。

在项目根目录新建 .env 文件（不要提交到 git，.env 要加入 .gitignore）：

# 复制下面任一种配置即可

# --- 方案 A：OpenAI 官方 ---
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxx
# OPENAI_BASE_URL 不填，默认走官方

# --- 方案 B：DeepSeek ---
# OPENAI_API_KEY=sk-deepseek-xxxxxxxx
# OPENAI_BASE_URL=https://api.deepseek.com/v1
# MODEL_NAME=deepseek-chat

# --- 方案 C：阿里云百炼 Qwen ---
# OPENAI_API_KEY=sk-aliyun-xxxxxxxx
# OPENAI_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1
# MODEL_NAME=qwen-plus

# --- 方案 D：智谱 GLM ---
# OPENAI_API_KEY=glm-xxxxxxxx
# OPENAI_BASE_URL=https://open.bigmodel.cn/api/paas/v4
# MODEL_NAME=glm-4-flash

具体模型名与端点以各家官方文档为准。下面所有代码都读取 OPENAI_API_KEY / OPENAI_BASE_URL 环境变量，换模型只改 .env，代码不动。

用 python-dotenv 把 .env 加载进来（Java 类比：.env ≈ 本地 application-local.yml）。

2.4 第一个 LLM 调用

新建 hello_llm.py：

"""第一个 LLM 调用：让模型自我介绍。
用法：
    python hello_llm.py
依赖：见 requirements.txt，另需 .env 配置 API Key。
"""

import os

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()  # 读取 .env 里的 OPENAI_API_KEY / OPENAI_BASE_URL

# 兼容国产端点：base_url 为空时默认 OpenAI 官方
model = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0,          # 越低越确定，适合"回答事实型问题"
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)

resp = model.invoke("用一句话介绍你自己")
print(resp.content)          # 模型回复文本
# print(resp)                # 取消注释可看完整返回结构

运行：

python hello_llm.py

2.4.1 理解返回结构

resp 是 AIMessage 对象，最重要的字段：

字段	含义
`resp.content`	模型生成的文本
`resp.tool_calls`	模型请求调用的工具列表（第 2.7 节会用）
`resp.usage_metadata`	输入/输出 token 数（算成本的依据）
`resp.response_metadata`	原始响应元数据（如 finish_reason）

Java 类比：一次 invoke ≈ 一次 HTTP 请求。模型是无状态的，你这次发了什么它看到什么，上次说了什么它记不住——这是整个 LLM 工程最需要记住的一点。

2.4.2 最常见的坑

认证失败：.env 没被加载或 key 写错。先 print(os.getenv("OPENAI_API_KEY")) 确认。
模型名不存在：国产端点不支持 gpt-4o-mini，换成该厂商文档里有的模型名。
网络超时：默认 60s，网络慢可以给 ChatOpenAI(..., timeout=120)。
rate limit（限流）：高频调用会报 429。生产环境要加退避重试，第 12 章详讲。

2.5 流式输出

LLM 是逐个 token 生成的，流式输出能显著改善用户体验（类似网页请求的 chunked 传输，首字快、体感"边想边说"）。修改 hello_llm.py：

from langchain_core.messages import HumanMessage

# 流式：for chunk in model.stream(...)
for chunk in model.stream([HumanMessage(content="写一首关于飞机的五言绝句")]):
    print(chunk.content, end="", flush=True)
print()

每个 chunk 是增量片段，拼起来就是完整回复；
注意这里传的是 messages 列表（HumanMessage），而不是裸字符串——与真实生产一致。

2.6 结构化输出：JSON + pydantic 校验

模型返回文本，但下游程序要的是结构化数据。最可靠的方式：定义 pydantic 模型 → with_structured_output → 模型直接输出符合 schema 的 JSON。

Java 类比：pydantic 模型 ≈ Java 的 POJO/DTO + Bean Validation 注解；with_structured_output ≈ 让模型按你定义的契约返回，客户端直接反序列化。

"""结构化输出示例：从一段航班描述里提取结构化字段。"""

import os

from dotenv import load_dotenv
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

load_dotenv()

model = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)


class FlightInfo(BaseModel):
    """模型必须输出的结构（类似 DTO）。"""

    flight_no: str = Field(description="航班号，如 CA1234")
    departure: str = Field(description="出发城市")
    arrival: str = Field(description="到达城市")
    date: str = Field(description="日期，格式 YYYY-MM-DD")


# 绑定结构：模型只输出符合 FlightInfo 的 JSON
structured_model = model.with_structured_output(FlightInfo)

text = "请帮我查一下下周五从北京飞上海的 CA1501 航班"
result = structured_model.invoke([HumanMessage(content=text)])
print(result)          # FlightInfo 对象
print(result.flight_no, result.departure, "->", result.arrival)

输出字段不齐或类型错误时，模型会尝试自我修正，最终返回的对象保证符合 FlightInfo；
若仍失败，可加一层"校验失败重试"——把错误信息回传给模型再抽一次（完整模式见第 03 章）。

2.7 手写第一个 Agent 循环（核心）

这是全教程最重要的一个示例。它就是 Agent 的本质。

什么是 Agent？就是一个 while 循环：

while 模型还没有给出最终答案:
    1. 把消息列表发给模型
    2. 模型要么直接回答，要么请求调用某个工具(返回 tool_calls)
    3. 如果请求调工具: 宿主执行该工具, 把结果以"工具消息"回填给模型
    4. 如果直接回答: 循环结束, 输出答案

下面实现一个"航班助手"：模型可以在需要时调用 get_flight_status（查航班状态）或 get_flight_price（查机票价格），真正执行查询的是你的 Python 代码——模型只负责"决定调哪个工具、传什么参数"。

"""手写 Agent 循环：航班助手（不依赖任何 Agent 框架）。

原理：
    - 模型根据对话决定调用哪个工具（tool_calls）；
    - 宿主（本脚本）执行真正的工具函数；
    - 结果以 ToolMessage 回填，模型基于结果继续推理；
    - 直到模型直接输出最终回答。

用法：
    python flight_agent.py
"""

import os
from datetime import date

from dotenv import load_dotenv
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

load_dotenv()

# ---------- 1. 定义工具：真正干活的函数 ----------
# @tool 装饰器把 Python 函数转成"模型可理解的工具"(自动生成 JSON Schema)
@tool
def get_flight_status(flight_no: str) -> str:
    """查询航班实时状态。

    Args:
        flight_no: 航班号，如 CA1501
    """
    # 真实场景这里应调用航司 API / 内部服务
    table = {
        "CA1501": "已起飞，预计 11:35 抵达上海虹桥",
        "MU5102": "延误约 40 分钟，预计 14:20 起飞",
        "CZ3901": "准点，正点起飞",
    }
    return table.get(flight_no.upper(), f"未查询到航班 {flight_no} 的信息")


@tool
def get_flight_price(flight_no: str, travel_date: str) -> str:
    """查询指定航班在指定日期的经济舱票价。

    Args:
        flight_no: 航班号，如 CA1501
        travel_date: 出行日期，格式 YYYY-MM-DD
    """
    price = {"CA1501": 780, "MU5102": 650, "CZ3901": 920}.get(flight_no.upper())
    if price is None:
        return f"未查询到航班 {flight_no} 的票价"
    return f"{flight_no} 在 {travel_date} 的经济舱票价约 {price} 元"


TOOLS = [get_flight_status, get_flight_price]
# 工具名 -> 函数 的注册表，方便分发执行
TOOL_MAP = {t.name: t for t in TOOLS}


# ---------- 2. 初始化模型并绑定工具 ----------
model = ChatOpenAI(
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    temperature=0,
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)
model_with_tools = model.bind_tools(TOOLS)  # 把工具 schema 告诉模型


def run_agent(user_input: str, max_iterations: int = 5) -> str:
    """Agent 主循环。

    Args:
        user_input: 用户问题
        max_iterations: 最大推理轮数保护，防止模型死循环烧钱
    """
    messages = [
        SystemMessage(
            content=(
                "你是智能客服助手。你可以查询航班状态和票价。"
                "回答时语气友好、简洁。若需要最新数据，请调用工具。"
            )
        ),
        HumanMessage(content=user_input),
    ]

    for i in range(max_iterations):
        print(f"\n--- 第 {i + 1} 轮推理 ---")
        resp: AIMessage = model_with_tools.invoke(messages)
        messages.append(resp)

        # 情况一：模型没有请求调用工具 → 输出最终回答，结束
        if not resp.tool_calls:
            return resp.content

        # 情况二：模型请求调用工具 → 宿主执行，结果回填
        for tc in resp.tool_calls:
            name = tc["name"]
            args = tc["args"]
            call_id = tc["id"]
            print(f"[工具调用] {name}({args})")

            func = TOOL_MAP.get(name)
            if func is None:
                result = f"错误：不存在工具 {name}，请换一种方式"
            else:
                try:
                    result = func.invoke(args)  # 执行真正的工具函数
                except Exception as e:  # 工具异常也要回填，让模型自己换策略
                    result = f"工具执行失败：{e}"

            # 关键：用 tool_call_id 把工具结果"挂回"对应的这次调用
            messages.append(
                ToolMessage(content=str(result), tool_call_id=call_id)
            )
            print(f"[工具结果] {result}")

    # 轮数耗尽保护
    return "抱歉，处理次数过多仍未完成，请稍后再试。"


if __name__ == "__main__":
    q1 = "帮我查一下 CA1501 现在什么状态？"
    print("用户：", q1)
    print("助手：", run_agent(q1))
    print("\n" + "=" * 60)
    q2 = f"下周五 {date.today().isoformat()} CA1501 的票价是多少？"
    print("用户：", q2)
    print("助手：", run_agent(q2))

2.7.1 跑起来看输出

--- 第 1 轮推理 ---
[工具调用] get_flight_status({'flight_no': 'CA1501'})
[工具结果] CA1501 已起飞，预计 11:35 抵达上海虹桥
--- 第 2 轮推理 ---
助手： CA1501 目前已起飞，预计 11:35 抵达上海虹桥，请您合理安排接机时间。

2.7.2 这段代码里的工程要点

工具是模型 schema + 宿主实现：@tool 装饰器只负责把函数签名转成模型能理解的 JSON Schema；真正执行的是你的代码，权限与数据校验都在你这端（重要安全点，第 12 章展开）。
tool_call_id 回填：工具结果必须用本次调用的 id 挂回，模型才能把"结果"和"这次调用"对上。
轮数保护 max_iterations：模型可能陷入"反复调同一个工具"的循环，必须有硬上限（类比：给递归加深度限制）。
异常回填：工具抛异常时，把错误信息作为工具结果回传，让模型换一种做法——而不是让整个流程崩掉。
无状态：每一轮都把 messages 完整传回，这就是 Agent 的"短期记忆"。

2.8 总结

环境：venv + requirements.txt + .env，一套环境贯穿整个教程；
调用：ChatOpenAI(...).invoke(...) 是"无状态 HTTP 调用"；.stream() 改善体验；with_structured_output 拿结构化数据；
Agent 的本质就是本章的手写循环：模型决策 → 宿主执行 → 结果回填 → 循环直至给出答案；
后续所有框架（LangChain 第 7-8 章、LangGraph 第 9-10 章）都只是在为这个循环加上更复杂的控制流、状态持久化、人工介入、多智能体——底层原理你已经在今天亲手实现过了。

练习

加一个 get_weather(city) 工具（随便返回 mock 数据），注册进 TOOLS 和 TOOL_MAP，让模型回答"北京明天适不适合坐飞机"。
把 max_iterations 改成 2，问一个需要连续调两个工具的问题，观察触发轮数保护时会发生什么。
故意把 .env 里的模型名改错，观察报错，再改回来——提前熟悉排查路径。
统计一次问答花了多少 token（resp.usage_metadata），建立成本体感。

参考答案

<details>
<summary>练习1（get_weather 工具）</summary>

@tool
def get_weather(city: str) -> str:
    """查询指定城市今天的天气。

    Args:
        city: 城市名，如 北京
    """
    table = {"北京": "晴，最高 8℃，风力 3 级", "上海": "小雨，最高 12℃",
             "广州": "多云，最高 22℃", "成都": "阴，最高 10℃"}
    return f"{city}：{table.get(city, '暂无数据')}"

TOOLS = [get_flight_status, get_flight_price, get_weather]
TOOL_MAP = {t.name: t for t in TOOLS}

问"北京明天适不适合坐飞机"时，模型会调 get_weather("北京") 拿到天气，再结合航空常识给出建议（晴天适合，暴雨需留意）。注意：get_weather 返回的是"今天"的数据——如果想让模型区分日期，把参数改成 get_weather(city, date)，这也体现了工具参数设计决定 Agent 能力边界。
</details>

<details>
<summary>练习2（max_iterations 轮数保护）</summary>

把 max_iterations 改成 2，问一个需要 3 次工具调用的问题（如"CA1501 什么状态？票价多少？天气适不适合飞？"）。运行结果：

--- 第 1 轮推理 ---   调 get_flight_status
--- 第 2 轮推理 ---   调 get_flight_price
（第 3 轮还没开始，循环已到上限）

最终返回兜底文案"抱歉，处理次数过多仍未完成，请稍后再试。"，没有最终答案。观察点：

轮数保护是防"模型死循环烧钱"的硬保险（每轮都是真金白银的 token）；
生产环境把"达到轮数上限"当作可观测事件记录日志（第 12 章），因为它往往意味着"工具反复失败"或"任务太复杂"，需要人工介入；
上线前要用真实场景估好上限（常见 5~15 轮）。
</details>

<details>
<summary>练习3（改错模型名）</summary>

把 .env 里 MODEL_NAME 改成不存在的名字（如 gpt-9）再运行，预期报错类型是 NotFound / 404 / model_not_found（具体文案各端点不同，国产端点通常是"模型不存在"或返回 400）。排查路径：

print(os.getenv("MODEL_NAME")) 确认环境变量真的读到了（.env 是否被 load_dotenv() 加载、文件名是否拼对）；
确认 base_url 指向的端点里有没有这个模型（各家模型列表以官方文档为准）；
改回来后重跑。

这条练习的价值：熟悉"模型名写错"这种最高频的配置错误长什么样，上生产时能 30 秒定位。

</details>

<details>
<summary>练习4（token 统计）</summary>

resp = model.invoke("...")
print(resp.usage_metadata)
# 输出示例：{'input_tokens': 18, 'output_tokens': 42, 'total_tokens': 60}

input_tokens 会随历史变长而增长——这就是为什么"多轮对话成本更高"；
成本估算：total_tokens / 1_000_000 × 单价(每百万 token)（各模型价格看官方文档）；
生产建议：每会话累计 token，按用户/天汇总，出现异常峰值就是"死循环"或"有人灌超长输入"的信号（第 12 章成本监控）。
</details>

▕┃