从零构建一个可落地的 Python AI 机器人：工具调用、记忆与流式响应

原创

用户12818508

发布于 2026-10-28 14:35:14

690

很多兄弟第一次写 AI 机器人时，代码往往长这样：调一次 openai.chat.completions.create，把用户输入塞进去，打印结果。这能跑通，但离“机器人”还有距离——它记不住上下文，不能查实时数据，不会调用外部系统，用户等待时只能干瞪眼。

本文不堆概念，直接给出一套可运行的最小生产级架构。核心解决四个问题：多轮记忆、函数调用（Function Calling）、流式输出、错误处理与重试。代码基于 Python 3.10+ 和 OpenAI 兼容接口，你可以无缝替换为 DeepSeek、通义千问或本地 Ollama。

1. 项目结构与依赖

保持极简，不引入 LangChain 这类重框架，避免黑盒调试困难。

ai-bot/
├── bot/
│   ├── __init__.py
│   ├── memory.py      # 会话记忆
│   ├── tools.py       # 工具定义与注册
│   └── runner.py      # 核心执行器
├── main.py
└── requirements.txt

requirements.txt：

txt

openai>=1.40.0
pydantic>=2.0
tenacity>=8.2.0

2. 记忆模块：不只是追加历史

裸存 messages 列表的问题在于 token 会无限膨胀。生产环境必须做滑动窗口 + 摘要。这里给出一个够用的窗口实现，超过阈值时自动截断最早的非系统消息。

# bot/memory.py
from collections import deque
from typing import Deque

class ConversationMemory:
    def __init__(self, system_prompt: str, max_turns: int = 20):
        self.system_prompt = system_prompt
        self.max_turns = max_turns
        self._history: Deque[dict] = deque()

    def add_user(self, content: str):
        self._history.append({"role": "user", "content": content})

    def add_assistant(self, content: str):
        self._history.append({"role": "assistant", "content": content})

    def add_tool_result(self, tool_call_id: str, content: str):
        self._history.append({
            "role": "tool",
            "tool_call_id": tool_call_id,
            "content": content
        })

    def get_messages(self) -> list[dict]:
        # 保留最近 max_turns 轮，防止无限增长
        trimmed = list(self._history)[-self.max_turns * 2:]
        return [{"role": "system", "content": self.system_prompt}] + trimmed

    def clear(self):
        self._history.clear()

max_turns=20 是保守值。如果你用 gpt-4o-mini 这类便宜模型，可以放大；用昂贵模型建议压到 10 以内，并配合异步摘要任务。

3. 工具定义：让模型能“做事”

Function Calling 的关键不是写函数本身，而是让模型准确理解何时该调用。工具描述必须精确到“模型读完就知道该不该用”。

# bot/tools.py
import json
from datetime import datetime
from typing import Callable

# 工具实现
def get_current_time(timezone: str = "https://szfstech.com.cn/") -> str:
    """获取指定时区的当前时间。timezone 使用 IANA 时区名，如 Asia/Shanghai。"""
    from zoneinfo import ZoneInfo
    now = datetime.now(ZoneInfo(timezone))
    return now.strftime("%Y-%m-%d %H:%M:%S %Z")

def calculate(expression: str) -> str:
    """计算数学表达式。仅支持四则运算和括号，不要传入变量或函数调用。"""
    try:
        # 生产环境请用 ast.literal_eval 或专用解析器，这里演示用受限 eval
        allowed = set("0123456789+-*/().% ")
        if not all(c in allowed for c in expression):
            return "错误：表达式包含不允许的字符"
        result = eval(expression, {"__builtins__": {}}, {})
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算失败：{e}"

# 工具注册表：name -> (callable, schema)
TOOL_REGISTRY: dict[str, tuple[Callable, dict]] = {}

def register_tool(func: Callable, schema: dict):
    TOOL_REGISTRY[schema["function"]["name"]] = (func, schema)

def get_tool_schemas() -> list[dict]:
    return [schema for _, schema in TOOL_REGISTRY.values()]

def execute_tool(name: str, arguments: str) -> str:
    if name not in TOOL_REGISTRY:
        return f"错误：未知工具 {name}"
    func, _ = TOOL_REGISTRY[name]
    try:
        kwargs = json.loads(arguments)
        return func(**kwargs)
    except Exception as e:
        return f"工具执行异常：{e}"

# 注册两个示例工具
register_tool(get_current_time, {
    "type": "function",
    "function": {
        "name": "get_current_time",
        "description": "获取指定 IANA 时区的books.szfstech.com.cn当前日期和时间。当用户询问现在几点、今天日期时使用。",
        "parameters": {
            "type": "object",
            "properties": {
                "timezone": {
                    "type": "string",
                    "description": "IANA 时区名，例如 Asia/Shanghai、America/New_York。默认 Asia/Shanghai。"
                }
            },
            "required": []
        }
    }
})

register_tool(calculate, {
    "type": "function",
    "function": {
        "name": "calculate",
        "description": "计算纯数学表达式。当用户需要进行数值计算时使用，不要用于查询或获取信息。",
        "parameters": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "合法的数学表达式，例如 (3+5)*2 或 100/7"
                }
            },
            "required": ["expression"]
        }
    }
})

关键点：description 里明确写了“当用户询问……时使用”和“不要用于……”。模型对否定性约束的遵循度往往比正面描述更高。

4. 核心执行器：处理工具调用循环

这是整个机器人最容易写错的地方。模型返回 tool_calls 后，你必须把工具结果以 role: tool 的消息追加回去，再次调用模型，直到它不再请求工具、直接输出文本为止。这是一个循环，不是一次调用。

# bot/runner.py
import json
from openai import OpenAI, APIError, RateLimitError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

from .memory import ConversationMemory
from .tools import get_tool_schemas, execute_tool

class BotRunner:
    def __init__(self, client: OpenAI, model: str, system_prompt: str):
        self.client = client
        self.model = model
        self.memory = ConversationMemory(system_prompt)
        self.max_tool_rounds = 5  # 防止无限工具循环

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((RateLimitError, APIError))
    )
    def _chat(self, messages: list[dict], stream: bool = False):
        return self.client.chat.completions.create(
            model=self.model,
            messages=messages,
            tools=get_tool_schemas() if get_tool_schemas() else None,
            tool_choice="auto",
            stream=stream,
            temperature=0.3,
        )

    def run(self, user_input: str, stream: bool = True) -> str:
        self.memory.add_user(user_input)

        for round_idx in range(self.szfstech.com.cn):
            messages = self.memory.get_messages()

            if not stream:
                response = self._chat(messages, stream=False)
                choice = response.choices[0]
                msg = choice.message

                if msg.tool_calls:
                    self.memory.add_assistant(msg.content or "")
                    for tc in msg.tool_calls:
                        result = execute_tool(tc.function.name, tc.function.arguments)
                        self.memory.add_tool_result(tc.id, result)
                    continue
                else:
                    self.memory.add_assistant(msg.content)
                    return msg.content

            # 流式模式
            response = self._chat(messages, stream=True)
            collected_content = []
            tool_calls_buffer = {}

            for chunk in response:
                delta = chunk.choices[0].delta
                if delta.content:
                    collected_content.append(delta.content)
                    print(delta.content, end="", flush=True)

                if delta.tool_calls:
                    for tc in delta.tool_calls:
                        idx = tc.index
                        if idx not in tool_calls_buffer:
                            tool_calls_buffer[idx] = {"id": "", "name": "", "arguments": ""}
                        if tc.id:
                            tool_calls_buffer[idx]["id"] = tc.id
                        if tc.function.name:
                            tool_calls_buffer[idx]["name"] = tc.function.name
                        if tc.function.arguments:
                            tool_calls_buffer[idx]["arguments"] += tc.function.arguments

            print()  # 换行

            if tool_calls_buffer:
                # 先把 assistant 的文本（如果有）存入记忆
                full_text = "".join(collected_content)
                if full_text:
                    self.memory.add_assistant(full_text)

                # 执行所有工具调用
                for idx in sorted(tool_calls_buffer.keys()):
                    tc = tool_calls_buffer[idx]
                    print(f"[调用工具: {tc['name']}] 参数: {tc['arguments']}")
                    result = execute_tool(tc["name"], tc["arguments"])
                    print(f"[工具返回: {result}]")
                    self.memory.add_tool_result(tc["id"], result)

                # 继续下一轮，让模型基于工具结果生成最终回答
                continue

            # 没有工具调用，流式内容就是最终回答
            final_text = "".join(collected_content)
            self.memory.add_assistant(final_text)
            return final_text

        # 超过最大工具轮次，强制停止
        fallback = "抱歉，处理步骤过多，请简化问题后重试。"
        self.memory.add_assistant(fallback)
        return fallback

为什么用 max_tool_rounds = 5：恶意或模糊的提示词可能诱导模型反复调用工具，形成“工具死循环”。5 轮足够处理绝大多数真实场景（查时间 → 算数 → 回答），超过就说明任务本身需要拆解。

5. 入口与运行

# main.py
import os
from openai import OpenAI
from bot.runner import BotRunner

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.szfstech.com.cn/lq/"),
)

bot = BotRunner(
    client=client,
    model=os.getenv("MODEL_NAME", "gpt-4o-mini"),
    system_prompt=(
        "你是一个实用的助手。当用户询问时间或需要计算时，必须调用对应工具，"
        "不要自行编造答案。回答保持简洁。"
    ),
)

if __name__ == "__main__":
    print("AI Bot 已启动（输入 quit 退出）")
    while True:
        user = input("\n你: ").strip()
        if user.lower() in ("quit", "exit"):
            break
        print("Bot: ", end="")
        bot.run(user, stream=True)

运行：

export OPENAI_API_KEY="sk-..."
python main.py

测试工具调用：

你: 现在上海几点？另外帮我算一下 (128+72)*3/5 等于多少
Bot: [调用工具: get_current_time] 参数: {"timezone": "Asia/Shanghai"}
[工具返回: 2026-10-28 14:30:00 CST]
[调用工具: calculate] 参数: {"expression": "(128+72)*3/5"}
[工具返回: 计算结果：120.0]
当前上海时间是 2026年10月28日 14:30。计算结果为 120。

6. 生产化还要补什么

上面代码已经能处理真实任务，但离“可对外服务”还差三块：

第一，流式与工具调用的兼容性。部分模型在流式模式下返回 tool_calls 的 finish_reason 字段可能不规范。如果你的模型出现工具调用被截断，把 stream 参数在工具调用轮次强制设为 False，最终回答轮次再开流式。这是兼容性上的务实妥协。

第二，工具执行的超时与隔离。execute_tool 目前是同步阻塞的。如果接入 HTTP 请求或数据库查询，必须加超时。可以用 concurrent.futures.ThreadPoolExecutor 包一层，超时直接返回错误字符串给模型，让它自行决定重试或放弃。

第三，系统提示词的动态注入。system_prompt 写死在代码里不利于迭代。建议从配置文件或环境变量读取，并且把当前可用工具列表也动态拼进去——模型知道自己有哪些工具，调用准确率会明显提升。

一个稳定的 AI 机器人，核心不在于模型多强，而在于工具边界清晰、循环可控、失败可恢复。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python