第六课:环境变量、配置系统、日志系统与最小 Agent 主循环

这节课你会学什么

前面几课你已经学了这些地基:

  • Python 基础
  • asyncio
  • HTTP / REST / WebSocket
  • JSON Schema
  • Tool Router

这一课开始,我们把这些东西真正拼起来,做出一个能持续执行任务的最小 Agent 外壳

学完这一课,你应该能理解这些概念:

  • 什么是环境变量
  • 为什么 API Key 不应该写死在代码里
  • 什么是配置系统
  • .envos.environ
  • 开发环境和生产环境的区别
  • 为什么 Agent 从第一天就要有日志
  • 日志级别
  • 结构化日志的基本思想
  • request id / task id
  • 最小 Agent 主循环是什么
  • Agent 为什么不是“一次函数调用”,而是“循环决策系统”

最后你会做一个小练习:

写一个带配置、日志、工具调用和主循环的最小 Agent。


1. 为什么这节课很重要

很多人做 Agent 时,一开始只关注:

  • 模型怎么调
  • 工具怎么接
  • prompt 怎么写

但真正让系统“能长期运行”的往往不是这些,而是下面这些基础设施:

  • 配置
  • 日志
  • 状态
  • 环境变量
  • 主循环结构

如果没有这些,你很快会遇到问题:

  • API Key 写死在代码里,不安全
  • 改一个模型名要改很多地方
  • 出错了不知道哪一步错了
  • 同一个任务跑到一半,状态丢了
  • 工具调用完,流程接不起来

所以这一课不是“配角”,而是:

把前面几课变成一个真正能跑的 Agent 程序的关键一课。


2. 什么是环境变量

环境变量(environment variable) 可以理解成:

操作系统提供给程序的一组外部配置值。

例如:

  • API Key
  • 数据库地址
  • 当前运行模式
  • 日志级别
  • 默认模型名

环境变量的好处是:

  • 不用把敏感信息写进代码
  • 不同环境可以用不同配置
  • 改配置时不用改源码

3. 为什么 API Key 不能写死在代码里

错误写法:

API_KEY = "sk-abc123"

这样做的问题:

  • 密钥容易泄漏
  • 提交到 Git 后很危险
  • 切换环境很麻烦
  • 团队协作时更容易出问题

正确思路是:

  • API Key 放在环境变量里
  • 程序运行时读取它

例如:

import os

api_key = os.environ.get("OPENAI_API_KEY")
print(api_key)

这里的 os.environ 可以理解成:

程序读取操作系统环境变量的入口。


4. .env 是什么

.env 是一种常见的配置文件,用来在开发环境中保存环境变量。

例如:

OPENAI_API_KEY=your-api-key
MODEL_NAME=gpt-5
APP_ENV=dev
LOG_LEVEL=INFO

然后程序启动时读取它。

注意:

  • .env 适合本地开发
  • 真实生产环境不一定直接依赖 .env
  • .env 一般不要提交到公开仓库

通常会配合 .gitignore 一起使用:

.env

5. 用 Python 读取环境变量

最基础的读取方式:

import os

model_name = os.environ.get("MODEL_NAME")
print(model_name)

如果变量不存在,get(...) 会返回 None

5.1 给默认值

import os

model_name = os.environ.get("MODEL_NAME", "gpt-5")
print(model_name)

这里表示:

  • 如果环境变量里有 MODEL_NAME,就用它
  • 否则默认用 "gpt-5"

5.2 必填环境变量

import os

api_key = os.environ.get("OPENAI_API_KEY")

if not api_key:
    raise RuntimeError("OPENAI_API_KEY is not set")

适合:

  • API Key
  • 数据库密码
  • 必须提供的服务地址

6. 用 python-dotenv 读取 .env

安装:

pip install python-dotenv

创建 .env 文件:

OPENAI_API_KEY=demo-key
MODEL_NAME=gpt-5
APP_ENV=dev
LOG_LEVEL=INFO

代码:

import os
from dotenv import load_dotenv

load_dotenv()

api_key = os.environ.get("OPENAI_API_KEY")
model_name = os.environ.get("MODEL_NAME", "gpt-5")

print(api_key)
print(model_name)

解释:

  • load_dotenv()
    表示把 .env 文件里的变量加载进当前进程环境

这样你在本地开发时就很方便。


7. 什么是配置系统

配置系统(configuration system) 可以理解成:

程序用来统一管理“可变参数”的一套机制。

例如这些都属于配置:

  • 模型名
  • 最大循环步数
  • 日志级别
  • 工具列表
  • 超时时间
  • 工作目录
  • API 地址

没有配置系统时,你的代码可能长这样:

MODEL = "gpt-5"
MAX_STEPS = 5
LOG_LEVEL = "INFO"

代码越来越多后,这样会变乱。

更好的做法是:

  • 用配置对象统一管理
  • 从环境变量 / JSON / .env 读取
  • 让主程序只依赖配置对象

8. 用 dataclass 表示配置

from dataclasses import dataclass


@dataclass
class AppConfig:
    model_name: str
    max_steps: int
    log_level: str
    app_env: str

这样配置就从“散落的变量”变成了“一个结构清晰的对象”。

使用:

config = AppConfig(
    model_name="gpt-5",
    max_steps=5,
    log_level="INFO",
    app_env="dev",
)

print(config)

9. 从环境变量构造配置对象

import os
from dataclasses import dataclass
from dotenv import load_dotenv


load_dotenv()


@dataclass
class AppConfig:
    model_name: str
    max_steps: int
    log_level: str
    app_env: str


def load_config() -> AppConfig:
    return AppConfig(
        model_name=os.environ.get("MODEL_NAME", "gpt-5"),
        max_steps=int(os.environ.get("MAX_STEPS", "5")),
        log_level=os.environ.get("LOG_LEVEL", "INFO"),
        app_env=os.environ.get("APP_ENV", "dev"),
    )


config = load_config()
print(config)

注意:

int(os.environ.get("MAX_STEPS", "5"))

这里必须转成整数,因为环境变量读出来默认是字符串。


10. dev / test / prod 是什么

很多项目会区分环境:

  • dev
    开发环境(development)
  • test
    测试环境(testing)
  • prod
    生产环境(production)

为什么要区分?

因为不同环境常常有不同配置:

  • 开发环境日志更详细
  • 测试环境使用假数据
  • 生产环境使用真实密钥和正式服务

例如:

if config.app_env == "dev":
    print("当前是开发环境")

第一阶段你不需要把这套做得很复杂,但至少要理解:

程序在不同环境下,配置可能不同。


11. 为什么日志系统从第一天就要有

如果你写的是一次性脚本,可能 print() 还能凑合。

但 Agent 是一个会:

  • 多步执行
  • 调模型
  • 调工具
  • 保存状态
  • 出现错误
  • 可能并发运行

的系统。

所以它必须有日志。

没有日志时,你会遇到:

  • 为什么它没调用工具?
  • 为什么它停在第 3 步?
  • 为什么同一个输入今天成功、明天失败?
  • 为什么 API 超时了?
  • 为什么工具参数不对?

所以:

日志不是锦上添花,而是 Agent 调试和运维的基础。


12. 什么是日志级别

常见日志级别:

  • DEBUG
  • INFO
  • WARNING
  • ERROR

可以这样理解:

  • DEBUG
    非常细的调试信息
  • INFO
    正常运行信息
  • WARNING
    不致命,但值得注意
  • ERROR
    发生了错误

例如:

import logging

logging.basicConfig(level=logging.INFO)

logging.debug("debug message")
logging.info("info message")
logging.warning("warning message")
logging.error("error message")

如果日志级别是 INFO,那么:

  • DEBUG 不会显示
  • INFO 及以上会显示

13. 最小日志系统

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
)

logger = logging.getLogger("agent")

logger.info("Agent started")
logger.warning("Tool response is slow")
logger.error("Task failed")

解释:

  • basicConfig(...)
    设置日志基础配置
  • level=logging.INFO
    最低显示 INFO 级别
  • format=...
    定义日志显示格式
  • getLogger("agent")
    创建一个具名 logger

14. 同时写入终端和文件

import logging
from pathlib import Path

log_dir = Path("logs")
log_dir.mkdir(exist_ok=True)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[
        logging.FileHandler(log_dir / "app.log", encoding="utf-8"),
        logging.StreamHandler(),
    ],
)

logger = logging.getLogger("agent")
logger.info("Agent started")

解释:

  • FileHandler
    把日志写入文件
  • StreamHandler
    把日志输出到终端
  • logs/app.log
    就是最终日志文件

这会让你既能实时看日志,又能事后追溯。


15. 什么是 structured logging

structured logging(结构化日志) 的意思是:

日志不只是随便拼一句话,而是尽量带上明确字段。

普通日志:

tool failed

更好的结构化思路:

task_id=123 tool_name=search status=failed error=timeout

在最小阶段,你不一定要上专门的 JSON 日志系统,但你要开始形成习惯:

  • 日志里带 task id
  • 带 tool name
  • 带 step
  • 带错误信息

这样以后查问题会轻松很多。


16. request id / task id 是什么

16.1 request id

request id 表示一次请求的唯一编号。

适合 Web API 场景:

  • 一个前端请求进来
  • 生成一个 request id
  • 整条处理链路都带上这个 id

16.2 task id

task id 表示一次任务的唯一编号。

适合 Agent 场景:

  • 一个用户任务
  • 一个任务执行过程
  • 一系列工具调用
  • 最终结果和日志都关联这个 id

最小版本可以这样生成:

import uuid

task_id = str(uuid.uuid4())
print(task_id)

这里 uuid 是一种常见的唯一标识符生成方式。


17. 什么是最小 Agent 主循环

这是这一课最重要的概念之一。

很多初学者会误以为 Agent 就是:

response = call_model(prompt)
print(response)

但真正的 Agent 更像一个循环:

  1. 读取当前任务
  2. 决定下一步做什么
  3. 如果需要,调用工具
  4. 更新状态
  5. 判断是否结束
  6. 否则继续下一轮

这就是:

Agent 主循环(agent loop)


18. 为什么 Agent 是循环,而不是一次调用

因为很多任务不是一句话就能做完。

例如:

帮我查一下最近 3 篇关于 AI agent 的论文,并总结它们的共同点

这个任务通常要分成几步:

  1. 先理解任务
  2. 决定调用搜索工具
  3. 搜索结果回来
  4. 选择几篇论文
  5. 提取关键信息
  6. 最后总结

所以 Agent 往往不是一次推理,而是:

多轮决策 + 工具调用 + 状态更新


19. 一个最小 Agent 主循环长什么样

下面是一个概念版:

step = 0
max_steps = 5

while step < max_steps:
    # 1. 思考下一步
    # 2. 如果需要,调用工具
    # 3. 更新状态
    # 4. 判断是否完成
    step += 1

这还很抽象,但它已经表达了核心:

  • 有步数
  • 有状态
  • 有中间过程
  • 可以终止

20. 定义最小状态对象

from dataclasses import dataclass, field


@dataclass
class AgentState:
    task_id: str
    task: str
    step: int = 0
    max_steps: int = 5
    finished: bool = False
    history: list[str] = field(default_factory=list)

这个状态对象里有:

  • task_id
    当前任务编号
  • task
    任务内容
  • step
    当前第几步
  • max_steps
    最大执行步数
  • finished
    是否完成
  • history
    执行历史

这就是“最小状态容器”。


21. 配置、日志、主循环三者的关系

你可以把它们理解成:

  • 配置
    决定系统怎么运行
  • 日志
    记录系统怎么运行
  • 主循环
    真正推动系统运行

所以一个最小 Agent 常常就是:

加载配置
    ↓
初始化日志
    ↓
创建状态
    ↓
进入主循环
    ↓
调用模型 / 调工具
    ↓
记录日志
    ↓
保存结果

22. 最小版本的“思考函数”

为了先把结构搭起来,我们不急着接真实模型,先写一个假的“思考函数”。

async def think_next_step(task: str, step: int) -> str:
    if step == 0:
        return "call_tool:echo"
    return "finish"

它的意思是:

  • 第 0 步时,决定调用 echo 工具
  • 后面就结束

这只是模拟,但足够帮助你看清 Agent 主循环的结构。


23. 最小版本的工具函数

def echo_tool(args: dict) -> str:
    return args["text"]

这和前一课的 Tool Router 会自然接起来。


24. 最小版本的 Tool Router

def route_tool_call(tool_name: str, arguments: dict) -> dict:
    if tool_name == "echo":
        try:
            result = echo_tool(arguments)
            return {
                "ok": True,
                "tool_name": tool_name,
                "data": result,
            }
        except Exception as e:
            return {
                "ok": False,
                "tool_name": tool_name,
                "error": str(e),
            }

    return {
        "ok": False,
        "tool_name": tool_name,
        "error": "unknown tool",
    }

这还是最小版本,但已经够用来搭循环。


25. 小项目:带配置、日志和主循环的最小 Agent

现在把这一课串起来。

25.1 .env

MODEL_NAME=gpt-5
MAX_STEPS=3
LOG_LEVEL=INFO
APP_ENV=dev

25.2 文件:minimal_agent.py

import asyncio
import logging
import os
import uuid
from dataclasses import dataclass, field
from pathlib import Path

from dotenv import load_dotenv


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


@dataclass
class AppConfig:
    """
    应用配置对象。

    用来统一管理系统运行参数。
    """
    model_name: str
    max_steps: int
    log_level: str
    app_env: str


@dataclass
class AgentState:
    """
    Agent 状态对象。

    它保存当前任务执行过程中的最小状态。
    """
    task_id: str
    task: str
    step: int = 0
    max_steps: int = 5
    finished: bool = False
    history: list[str] = field(default_factory=list)


def load_config() -> AppConfig:
    """
    从环境变量中读取配置,并转换成 AppConfig 对象。
    """
    return AppConfig(
        model_name=os.environ.get("MODEL_NAME", "gpt-5"),
        max_steps=int(os.environ.get("MAX_STEPS", "5")),
        log_level=os.environ.get("LOG_LEVEL", "INFO"),
        app_env=os.environ.get("APP_ENV", "dev"),
    )


def setup_logger(log_level: str) -> logging.Logger:
    """
    初始化日志系统,同时输出到终端和文件。
    """
    log_dir = Path("logs")
    log_dir.mkdir(exist_ok=True)

    level = getattr(logging, log_level.upper(), logging.INFO)

    logging.basicConfig(
        level=level,
        format="%(asctime)s | %(levelname)s | %(message)s",
        handlers=[
            logging.FileHandler(log_dir / "agent.log", encoding="utf-8"),
            logging.StreamHandler(),
        ],
    )

    return logging.getLogger("minimal-agent")


def echo_tool(args: dict) -> str:
    """
    最小示例工具:原样返回输入文本。
    """
    return args["text"]


def route_tool_call(tool_name: str, arguments: dict) -> dict:
    """
    最小 Tool Router。

    负责根据工具名分发调用工具,并统一包装返回值。
    """
    if tool_name == "echo":
        try:
            result = echo_tool(arguments)
            return {
                "ok": True,
                "tool_name": tool_name,
                "data": result,
            }
        except Exception as e:
            return {
                "ok": False,
                "tool_name": tool_name,
                "error": str(e),
            }

    return {
        "ok": False,
        "tool_name": tool_name,
        "error": f"unknown tool: {tool_name}",
    }


async def think_next_step(state: AgentState, config: AppConfig) -> dict:
    """
    模拟 Agent 的“下一步决策”。

    真实系统里,这里通常会:
    - 调模型
    - 根据状态决定下一步
    - 输出 action
    """
    await asyncio.sleep(0.3)

    if state.step == 0:
        return {
            "action": "call_tool",
            "tool_name": "echo",
            "arguments": {
                "text": f"任务回显: {state.task}"
            }
        }

    return {
        "action": "finish"
    }


async def run_agent(task: str, config: AppConfig, logger: logging.Logger) -> AgentState:
    """
    最小 Agent 主循环。

    执行流程:
    1. 创建状态
    2. 进入循环
    3. 每一步先思考
    4. 决定是否调用工具
    5. 更新状态
    6. 判断是否结束
    """
    state = AgentState(
        task_id=str(uuid.uuid4()),
        task=task,
        max_steps=config.max_steps,
    )

    logger.info("Task started | task_id=%s | task=%s", state.task_id, state.task)

    while state.step < state.max_steps and not state.finished:
        logger.info(
            "Loop step start | task_id=%s | step=%s/%s",
            state.task_id,
            state.step + 1,
            state.max_steps,
        )

        decision = await think_next_step(state, config)
        logger.info(
            "Decision made | task_id=%s | decision=%s",
            state.task_id,
            decision,
        )

        if decision["action"] == "call_tool":
            result = route_tool_call(
                decision["tool_name"],
                decision["arguments"],
            )

            logger.info(
                "Tool result | task_id=%s | tool=%s | result=%s",
                state.task_id,
                decision["tool_name"],
                result,
            )

            state.history.append(str(result))

        elif decision["action"] == "finish":
            logger.info("Task finished | task_id=%s", state.task_id)
            state.finished = True
            break

        state.step += 1

    if not state.finished:
        logger.warning(
            "Task ended by max_steps | task_id=%s | step=%s",
            state.task_id,
            state.step,
        )

    return state


async def main():
    """
    程序入口。
    """
    config = load_config()
    logger = setup_logger(config.log_level)

    logger.info(
        "App started | env=%s | model=%s | max_steps=%s",
        config.app_env,
        config.model_name,
        config.max_steps,
    )

    final_state = await run_agent(
        task="请回显这条任务,并在下一步结束",
        config=config,
        logger=logger,
    )

    print("任务完成")
    print("task_id:", final_state.task_id)
    print("history:", final_state.history)
    print("finished:", final_state.finished)


if __name__ == "__main__":
    asyncio.run(main())

26. 这段代码里最值得看懂的点

26.1 load_config()

它把环境变量变成一个结构化配置对象,而不是让配置散落在代码各处。

26.2 setup_logger()

它负责:

  • 创建日志目录
  • 设置日志级别
  • 同时输出到终端和文件

26.3 AgentState

它是最小状态容器,后面你可以继续扩展:

  • messages
  • tool_calls
  • memory
  • artifacts

26.4 think_next_step(...)

这是最小“决策层”。
现在是假的,后面可以替换为真实模型调用。

26.5 run_agent(...)

这就是最关键的 Agent 主循环。

它体现了一个真实 Agent 的雏形:

  • 有状态
  • 有步骤
  • 有决策
  • 有工具调用
  • 有日志
  • 有结束条件

27. 这节课你应该怎么练

第一次练习

先原样跑通,观察:

  • 终端日志
  • logs/agent.log
  • 最终打印的 history

第二次练习

think_next_step(...) 改成:

  • 第 0 步调用 echo
  • 第 1 步再调用一次 echo
  • 第 2 步再结束

第三次练习

route_tool_call(...) 增加一个 add 工具。

第四次练习

history 改成更结构化的形式,例如:

history: list[dict]

每步存:

  • step
  • action
  • tool_name
  • result

第五次练习

task_id 写进最终输出文件,例如:

  • outputs/{task_id}.json

28. 这节课你真正应该记住的东西

  • 环境变量适合放 API Key 和环境配置。
  • .env 适合本地开发阶段。
  • 配置系统的目标是把“可变参数”集中管理。
  • Agent 从第一天就应该有日志。
  • INFOWARNINGERROR 至少要会用。
  • task_id 和日志结合后,排查问题会轻松很多。
  • 真正的 Agent 不是“一次模型调用”,而是“循环式决策系统”。
  • 最小 Agent 主循环通常包含:
    • 状态
    • 决策
    • 工具调用
    • 日志
    • 结束条件

Logo

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目,面向数字基础设施四大核心场景(服务器、云计算、边缘计算、嵌入式),全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐