Mini Agent

项目不使用 LangChain、AutoGen 等 Agent 框架，只使用 OpenAI Python SDK 展示 Agent 最核心的工作方式：

用户问题
  -> 调用 OpenAI
  -> 模型一次返回一个或多个工具调用
  -> Python 并行执行同一批工具
  -> 按 tool_call_id 回填全部结果
  -> 再次调用 OpenAI
  -> 模型返回最终答案

• 只支持 OpenAI Chat Completions 风格接口。
• 一次运行只回答一个问题，不提供 REPL 交互模式。
• 核心代码集中在 agent.py，便于从上到下完整阅读。
• 记录每一步请求、OpenAI 原始响应、工具调用和工具结果。
• 使用有上限的线程池并行执行同一轮中的多个独立工具调用。
• 保留最大执行步数，防止模型无限调用工具。
• 工具实现刻意保持简单，让注意力集中在 Agent 主循环。

mini-agent/
├── agent.py          # Agent 循环、OpenAI 调用和工具实现
├── requirements.txt # Python 依赖
├── .env.example     # 环境变量示例，仅作参考
├── .gitignore       # 排除本地缓存和密钥文件

└── README.md

• Python 3.10 或更高版本
• OpenAI API Key

python -m pip install -r requirements.txt

运行前需要通过当前操作系统或 Shell 的环境变量设置方式配置：

使用三种相互独立的工具更容易观察同一批并行执行：

python agent.py --max-parallel-tools 3 "必须在第一个 assistant 响应中一次性发出以下 3 个相互独立的工具调用，不要等待任何结果：调用 get_current_time 获取时间；调用 calculate 计算 123+456；调用 web_search 搜索 Python 3.14。收到全部结果后再统一回答。"

--max-parallel-tools 只控制 Python 收到一批 tool_calls 后的执行并发数，parallel_tool_calls=True 也只是允许模型批量返回，并不能强制模型生成多个调用。是否出现 calls=3 workers=3 仍取决于所用模型和兼容接口；如果模型只返回一个调用，日志就会显示 calls=1 workers=1，这不是线程池没有并行，而是本轮没有更多任务可并行。

默认情况下，每条日志都会带上源码文件名、行号和函数名，例如：

2026-07-27 15:00:00,000 | INFO | agent.py:164 | run_agent() | OpenAI raw response:

程序会输出以下可观测信息：

1. 当前 Agent 步数。

2. 本次发送给 OpenAI 的完整 messages。

3. OpenAI SDK 返回的原始响应。

4. 并行工具批次的调用数量、工作线程数和总耗时。

5. 每个工具的 tool_call_id、参数、结果和各自耗时。

6. Agent 是否已经生成最终答案。

示意流程：

Agent step 1
OpenAI request messages: user question
OpenAI raw response: assistant requests calculate(...)
Tool call: calculate
Tool result: 1694

Agent step 2
OpenAI request messages: question + tool call + tool result
OpenAI raw response: assistant final answer
Answer: 结果是 1694

日志不会输出 OPENAI_API_KEY，但会包含用户问题、模型回答和工具结果。处理敏感数据时应使用 --quiet，或者进一步调整日志策略。

TOOL_SCHEMAS 使用 OpenAI function calling 格式向模型描述工具。模型只能看到工具名称、说明和参数结构，看不到 Python 函数源码。

本项目提供三个工具：

• get_current_time()：返回当前本地时间。
• calculate(expression)：使用简洁的 eval() 计算表达式，仅供教学。
• web_search(query)：通过 ddgs 执行真实网页搜索，返回最多 5 条结果。

TOOL_FUNCTIONS 建立工具名到 Python 函数的映射：

TOOL_FUNCTIONS = {
    "get_current_time": get_current_time,
    "calculate": calculate,
    "web_search": web_search,
}

模型返回 calculate 时，程序通过这个字典找到并执行真正的 Python 函数。

run_agent() 是项目核心。它维护一个 messages 列表，并在每一步调用：

client.chat.completions.create(
    model=model,
    messages=messages,
    tools=TOOL_SCHEMAS,
    tool_choice="auto",
    parallel_tool_calls=True,
)

如果响应中没有 tool_calls，模型的文本就是最终答案。如果同一轮存在多个 tool_calls，execute_tool_calls() 会使用 ThreadPoolExecutor 同时执行它们：

with ThreadPoolExecutor(max_workers=worker_count) as executor:
    results = list(executor.map(execute_tool_call, tool_calls))

executor.map() 并发执行任务，但按输入顺序返回结果。每条结果仍通过自己的 tool_call_id 与请求对应：

{
    "role": "tool",
    "tool_call_id": tool_call["id"],
    "content": result,
}

所有结果回填后，下一次请求会携带完整历史，因此模型可以汇总回答。

一次响应可能包含多个工具调用。tool_call_id 用于准确关联模型发出的工具请求和 Python 返回的工具结果。

--max-parallel-tools 控制线程池大小，默认最多同时执行 4 个工具。即使模型一次返回很多调用，也不会无限创建线程。工具内部异常会被转换成普通工具结果，不影响同批次中的其他工具。

模型可能持续或重复调用工具。max_steps 给循环设置明确上限，避免程序无限运行。默认最多执行 8 轮。

以新增天气工具为例：

1. 在 agent.py 中实现 Python 函数：

def get_weather(city: str) -> str:
    return "这里调用真实天气 API"

2. 注册函数：

TOOL_FUNCTIONS["get_weather"] = get_weather

3. 在 TOOL_SCHEMAS 中增加对应 JSON Schema。

Schema 名称、注册表名称和函数参数必须保持一致。

本项目刻意保持教学性质，没有实现：

1. 流式输出
2. 多轮对话历史
3. 上下文压缩
4. 工具依赖分析、读写冲突检测和独占工具调度

这些能力很重要，但不属于理解 Agent 最小循环所必需的部分。

• 不要将 API Key 写入 agent.py 或提交到 Git。

• calculate() 为了教学简洁性使用了受限命名空间的 eval()，仍不应处理不可信输入或用于生产环境。

• 接入文件、Shell、数据库或网络工具时，需要额外增加权限控制、参数验证、超时和隔离机制。

• OpenAI API 调用可能产生费用，请关注所选模型和账户用量。

本项目基于 MIT License 开源。你可以自由使用、复制、修改、合并、发布和分发本项目，但需要保留原始版权声明和许可证文本。