用最短代码写出第一个智能体:从纯大模型到能调用一个工具

生成式大模型可以生成文本、图片、声音以及视频,普通用户利用Web页面、APP或者桌面应用使用大模型提供的这种能力。而从开发者的角度来说,则可以调用大模型服务平台所提供的API使用大模型的生成能力。

通常,开发者在调用大模型服务平台提供的服务能力时会对其提供的API以函数的形式进行封装或者称为适配,以便在需要切换大模型服务供应商时仅修改适配函数即可。例如以下代码是chat()函数对DeepSeek所提供的会话模型(文本生成服务)适配(DeepSeek采用兼容OpenAI的API的策略,并可直接使用OpenAI的SDK)。

def chat(messages: list[dict], 
         *, 
         json_mode: bool = False, 
         model: str | None = None, 
         **kwargs) -> str:
    """调用 DeepSeek Chat Completions。返回 assistant 的 content 字符串。

    参数:
      - messages: OpenAI 兼容的消息列表
      - json_mode: True 时在请求中设置 response_format={"type":"json_object"}
      - model: 指定模型,默认取环境变量 DEEPSEEK_MODEL 或 deepseek-chat
    环境变量:
      - DEEPSEEK_API_KEY
      - DEEPSEEK_BASE_URL (默认 https://api.deepseek.com)
      - DEEPSEEK_MODEL (默认 deepseek-chat)
    """

    api_key = os.getenv("DEEPSEEK_API_KEY")
    base_url = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
    model = model or os.getenv("DEEPSEEK_MODEL", "deepseek-chat")
    if not api_key:
        raise RuntimeError("缺少 DEEPSEEK_API_KEY。请在 .env 中配置或导出到环境变量。")

    # OpenAI 官方 SDK,指向 DeepSeek 的 base_url(OpenAI 兼容)
    from openai import OpenAI
    client = OpenAI(api_key=api_key, base_url=base_url)

    params = dict(
        model=model,
        messages=messages,
        temperature=kwargs.get("temperature", 0.2),
    )
    if json_mode:
        params["response_format"] = {"type": "json_object"}

    resp = client.chat.completions.create(**params)
    return resp.choices[0].message.content

应用程序调用函数chat()即可使得大模型根据传入的messages参数中消息列表的内容生成对应的响应文本。

例如可以使用如下测试代码尝试调用chat()函数。

def test_chat():
    """测试 chat 函数的基本功能"""
    
    # 获取用户输入
    user_input = input("请输入您的测试消息: ")
    messages = [
        {"role": "user", "content": user_input}
    ]
    
    print("测试 chat 函数:")
    print("-" * 40)
    
    # 调用 chat 函数
    try:
        response = chat(messages)
        print("输入消息:")
        for msg in messages:
            print(f"  {msg['role']}: {msg['content']}")
        print("\n返回内容:")
        print(response)
        print("-" * 40)
        
    except Exception as e:
        print(f"调用 chat 函数时出错: {e}")

运行测试代码,如图 1所示,可以看到大模型正确地对输入的提示词进行了响应。

图 1 大模型生成文本响应

但是,生成式大模型的能力如果是只能写诗或者讲笑话,那么它将只是一个玩具。事实上,生成式大模型已经具备推理分析能力,正因为如此,基于生成式大模型能够在具体场景中解决实际问题的智能体(Agent)才会成为各类智能应用的主要形态。

智能体是“一个能感知外部世界,并根据所感知的信息采取相应行动的智能系统(
An agent is anything that can be viewed as perceiving its environment through sensors and acting upon that environment through actuators.
)”。在生成式人工智能领域通常认为一个智能体主要由规划、记忆、工具与行动构成,如图 2所示。

图 2 智能体(Agent)构成概览

智能体“感知外部世界”并“采取相应行动”通常需要使用上图中的“工具(Tools)”完成,即智能体要具备调用外部工具的能力。

智能体调用外部工具的能力是利用生成式大模型的推理分析能力完成的。具体来说,是在向大模型提交提示词时同步将可用工具以特定格式进行描述,并详细说明其功能作用与所需参数及其含义,而大模型则结合当前要解决的问题进行推理分析,判断是否需要调用外部工具以及应当调用哪个工具。

例如,当前有一个计算两个数之和(加法)的工具,期望大模型在需要进行加法计算的场景中调用该工具执行加法计算(当前生成式大模型以推理方式在绝大部分情况下也能够获得加法的正确结果,但这里希望大模型通过工具而不是自行计算加法)。通常而言,大模型识别出在响应用户的需求中有加法计算的需要时(思考),它便会要求应用程序先调用这个加法工具获得加法运算结果(行动),然后应用程序将结果返回给大模型(观察)生成最终回复(结论)。事实上这个过程就形成了智能体的最小闭环,如图 3所示。

图 3 思考→行动→观察→结论

在上述闭环里,令大模型理解可用工具(包括工具的名称、作用、参数等信息)并在需要调用工具时按特定结构化文本返回调用工具要求是整个闭环顺畅流转的关键所在。

首先实现加法工具。例如,以下是加法工具可能的一种代码实现。

class SumArgs(BaseModel):
    a: float
    b: float

def sum_tool(**kwargs) -> float:
    args = SumArgs(**kwargs)
    return float(args.a) + float(args.b)

然后需要将工具信息提交给大模型并在需要时获得其工具调用指令,而将工具信息提供给大模型及大模型返回工具调用指令有以下两种实现方法。

  1. 基于特别设计提示词的工具调用

在提示词中描述工具并约定工具调用指令格式是智能体实现时经常采用的方法。例如,本案例编写的提示词如下。

AGENT_SYSTEM = (
    "你是一个会用工具的 Agent。严格遵守以下规则:\n"
    "1) 你只能输出 **JSON 对象**(json),且仅限以下两种结构之一:\n"
    "   - {\"tool\": \"<name>\", \"args\": { ... }}\n"
    "   - {\"final\": \"<你的回答>\"}\n"
    "2) 如果需要用工具,请选择合适的工具并给出参数;若不需要,请直接给出 final。\n"
    "3) 严禁输出除 JSON 之外的任何文字、注释或 Markdown 代码块。\n"
    "可用工具:\n"
    "sum: 对两个数值进行加法运算\n"
    "  - 功能:计算两个浮点数的和\n"
    "  - 参数:a (number) 第一个加数, b (number) 第二个加数\n"
    "  - 返回:两个数的和\n"
    "  - 示例:{\"tool\": \"sum\", \"args\": {\"a\": 2, \"b\": 3}} 会返回 5\n"
)

在上述提示词中首先约定了大模型需要遵守的3条规则,分别对返回信息格式、行为规则及禁止事项进行了约定,最后则对加法工具进行了描述。

上述提示词中工具的名称是“sum”,而在前文提及的加法实现工具的函数名称是“
sum_tool”,两者并不相同。这是编程上的一个小技巧。因为通常智能体需要调用的工具不止一个,所以在实现工具调用时通常会封装一个“路由”或“分发”函数具体实现对工具的调用。而在这个环节中,可以添加一个“工具字典”建立提交给大模型的工具名称与实际工具函数名称之间的映射关系,这样可以在提交工具信息给大模型时使用更能体现工具作用的工具名称。

实现工具映射及工具路由的代码如下。

TOOL_REGISTRY = {
    "sum": sum_tool,
}

def dispatch_tool(name: str, args: Dict[str, Any]) -> Any:
    if name not in TOOL_REGISTRY:
        raise KeyError(f"未知工具: {name}")
    fn = TOOL_REGISTRY[name]
    try:
        return fn(**args)
    except ValidationError as ve:
        # 透出更友好的错误信息给上层
        raise ValueError(f"参数校验失败: {ve}") from ve

回到提示词上,为了使得模型能够更好地理解应如何调用工具并应用返回的结果,可以在提示词中添加一些样例,加法工具的示例如下。

FORMAT_EXAMPLE = (
    "示例:\n"
    "用户:2+3是多少?\n"
    "助手:{\"tool\":\"sum\",\"args\":{\"a\":2,\"b\":3}}\n"
    "(本地执行 sum 返回 5)\n"
    "助手:{\"final\":\"结果是 5\"}"
)

然后将上述提示词以及用户当前的输入组装到消息列表中并通过chat()函数提交给大模型进行推理分析。在获得大模型返回的信息后,需要先对其格式进行验证,判断返回信息是否按要求以JSON格式返回。随后从返回的JSON中抽取判断大模型当前是要求调用工具还是已经生成了最终回复。如果是调用工具的指令,则继续从中抽取出调用工具所需要的参数字典;如果是最终回复,则继续从中获取最终回复的文本内容。

实现上述逻辑的代码如下。

def run_agent(user_input: str, max_steps: int = 3) -> str:
    """最小单工具 Agent 主循环。返回 final 文本。"""
    messages = [
        {"role": "system", "content": AGENT_SYSTEM},
        {"role": "user", "content": FORMAT_EXAMPLE},
        {"role": "user", "content": user_input},
    ]

    for _ in range(max_steps):
        out = chat(messages, json_mode=True)
        try:
            data = _parse_json(out)
        except json.JSONDecodeError:
            # 强制提醒:只输出 JSON
            messages.append({"role": "assistant", "content": "输出格式错误,请只输出 JSON 对象。"})
            continue

        if "tool" in data:
            name = data["tool"]
            args = data.get("args", {}) or {}
            try:
                print(f"[Debug: 执行工具:{name},参数:{args}]")
                result = dispatch_tool(name, args)
            except Exception as e:
                # 把错误作为“观察”反馈给模型,提示修正
                messages.append({"role": "assistant", "content": json.dumps(data, ensure_ascii=False)})
                messages.append({"role": "user", "content": f"OBS: 工具执行失败,原因:{e}. 请修正参数或改用 final。"})
                continue

            # 把工具结果作为观察提供,提示收敛为 final
            messages.append({"role": "assistant", "content": json.dumps(data, ensure_ascii=False)})
            messages.append({"role": "user", "content": f'OBS: 工具 {name} 返回 {result}. 若可给出答案,请输出 {{"final": "..."}}'})
            continue

        if "final" in data:
            return str(data["final"])

        # 若既无 tool 又无法终止,要求重试
        messages.append({"role": "assistant", "content": "缺少 tool 或 final 字段,请按约定输出。"})
    raise RuntimeError("达到最大步数仍未获得 final。")

在上述代码中,还针对每一步可能产生的异常进行了处理。从具体处理手法可以看到,通常是将错误信息提交给大模型,让大模型重新给出正确的回复。这是智能体常用的异常处理方式,它提升了程序的健壮性和可用性。

最后,可通过以下代码完成接收用户的输入并启动智能体响应用户的输入。

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print('用法: python agent.py "2+3是多少?"')
        sys.exit(1)
    query = " ".join(sys.argv[1:]).strip()
    try:
        ans = run_agent(query)
        print(ans)
    except Exception as e:
        print(f"[Error] {e}")
        sys.exit(2)

运行程序,并将希望智能体解决的问题作为参数一并给出,可以看到加法工具被成功调用,智能体也给出了用户问题的结果。某次运行效果如图 4所示。

图 4 工具调用及结果

本示例完整源代码可参见:https://gitcode.com/gtyan/AgentHandBook/tree/main/01-mini-agent

2. 基于函数调用(Function Calling)机制的工具调用

基于智能体调用工具的需求,部分大模型提供了函数调用(Function Calling)机制,即在与大模型交互的API中提供了专门的工具说明参数。对于penAI/DeepSeek而言,它由一个名为tools的工具说明字典列表和一个名为tool_choice的工具选择策略参数构成。

面对这类大模型且需要使用其函数调用能力时,可修改适配大模型的函数为其增加使用工具的相关参数。以下代码所实现的chat_with_tools()函数即是可能的一种可能适配。

def chat_with_tools(messages: list[dict], 
                    tools: list[dict], 
                    *, 
                    model: str | None = None, 
                    tool_choice: str = "auto", 
                    **kwargs) -> dict:
    """调用 DeepSeek(OpenAI 兼容)Chat Completions 的 Function Calling。

    返回简化过的 assistant message:
    { "role": "assistant", "content": str|None, "tool_calls": [ {id, type, function:{name,arguments}} ] }
    """

    api_key = os.getenv("DEEPSEEK_API_KEY")
    base_url = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
    model = model or os.getenv("DEEPSEEK_MODEL", "deepseek-chat")
    if not api_key:
        raise RuntimeError("缺少 DEEPSEEK_API_KEY。请在 .env 中配置或导出到环境变量。")

    from openai import OpenAI
    client = OpenAI(api_key=api_key, base_url=base_url)
    params = dict(
        model=model,
        messages=messages,
        tools=tools,
        tool_choice=tool_choice,   # "auto" | {"type":"function","function":{"name":"sum"}}
        temperature=kwargs.get("temperature", 0.2),
    )
    resp = client.chat.completions.create(**params)
    msg = resp.choices[0].message

    # 适配:把 OpenAI SDK Message 对象转为 dict(避免上层依赖 SDK 类型)
    out = {
        "role": msg.role,
        "content": msg.content,
        "tool_calls": []
    }
    if getattr(msg, "tool_calls", None):
        for tc in msg.tool_calls:
            out["tool_calls"].append({
                "id": tc.id,
                "type": tc.type,
                "function": {
                    "name": tc.function.name,
                    "arguments": tc.function.arguments,
                }
            })
    return out

上述chat_with_tools()函数通过tools和tool_choice参数向大模型提交了可用工具的信息,而对于大模型的返回信息chat_with_tools()函数也对它进行格式转换处理,以便在切换大模型时可以隔离修改。

智能体调用chat_with_tools()函数的逻辑参见如下代码。代码中需要重点关注两点,一是调用时特别为tools形参提供了TOOLS_SPEC实参,为tool_choice形参提供了“auto”参数值;二是在收到chat_with_tools()函数经大模型推理的包含函数调用指令的返回值后,先将模型回复内容添加对话消息列表,然后解析出其中所包含的函数调用指令及对应参数并调用_exec_tool()函数执行函数调用,再将函数调用结果继续添加到对话消息列表。

上述过程也被置于一个循环之中,直至大模型给出最终回复或者超出设定的最大尝试次数。

def run_agent_fc(user_input: str, max_steps: int = 3) -> str:
    messages = [{"role": "system", "content": SYSTEM_FC},
                {"role": "user", "content": user_input}]

    for step in range(max_steps):
        reply = chat_with_tools(messages, tools=TOOLS_SPEC, tool_choice="auto")
        # 如果模型直接给出文本答案,且没有工具调用,直接返回
        if (reply.get("content") and not reply.get("tool_calls")):
            return str(reply["content"]).strip()

        # 否则执行 tool_calls(可能有多个)
        if reply.get("tool_calls"):
            # optional:把 assistant 的原始回复也追加进去(即使 content 为 None)
            messages.append({"role": "assistant", "content": reply.get("content") or "", "tool_calls": reply["tool_calls"]})

            tool_msgs = []
            for call in reply["tool_calls"]:
                name = call["function"]["name"]
                args_str = call["function"]["arguments"]
                call_id = call["id"]
                try:
                    result = _exec_tool(name, args_str)
                    tool_msgs.append({"role": "tool", "tool_call_id": call_id, "name": name, "content": json.dumps({"ok": True, "result": result}, ensure_ascii=False)})
                except Exception as e:
                    tool_msgs.append({"role": "tool", "tool_call_id": call_id, "name": name, "content": json.dumps({"ok": False, "error": str(e)}, ensure_ascii=False)})
            messages.extend(tool_msgs)
            # 进入下一轮让模型整合工具结果→最终答复
            continue

        # 若既无 content 也无 tool_calls,则提示重试
        messages.append({"role": "user", "content": "请根据需要调用工具或直接给出答案。"})
    raise RuntimeError("达到最大步数仍未获得答案。")

为tools参数所指定的TOOLS_SPEC实参是一个按函数调用所要求的函数描述规范构建的列表,当前这个列表中仅包含描述加法工具的一条说明,其具体描述形式及内容如下,描述中各项信息的具体含义与内容根据名称很容易理解,不再赘述。

TOOLS_SPEC = [
    {
        "type": "function",
        "function": {
            "name": "sum",
            "description": "计算两数之和,返回 a+b。",
            "parameters": {
                "type": "object",
                "properties": {
                    "a": {"type": "number", "description": "加数1"},
                    "b": {"type": "number", "description": "加数2"},
                },
                "required": ["a", "b"],
                "additionalProperties": False
            }
        }
    }
]

用于执行工具的_exec_tool()函数通过传入的参数查找对应工具并使用传入的工具实参执行它,其代码如下。

def _exec_tool(name: str, args_str: str) -> Any:
    if name not in TOOL_FUNCS:
        raise KeyError(f"未知工具:{name}")
    args = _safe_json_loads(args_str) if isinstance(args_str, str) else (args_str or {})
    print(f"[Debug: 执行工具:{name},参数:{args}]")
    return TOOL_FUNCS[name](**args)

_exec_tool()函数的这种设计可以方便地扩展更多工具而不仅局限于当前案例中的加法工具。代码中用到的TOOL_FUNCS与前述“基于特别设计提示词的工具调用”案例中的TOOL_REGISTRY一样是一个保存可用工具的字典。

在向大模型提交提示词(消息列表)时,提交了系统提示词SYSTEM_FC和用户输入的提示词。特别地,相比“基于特别设计提示词的工具调用”案例的提示词,SYSTEM_FC并没有对工具进行具体说明,如前所述在函数调用机制下这些信息经由
tools参数传递而不必在提示词中说明。

SYSTEM_FC = (
    "你是一个会合理使用工具的助手。\n"
    "当需要进行加法或用户请求显然涉及加总时,请调用对应工具;\n"
    "否则,直接给出最终回答。\n"
    "请保持回答简洁。"
)

使用函数调用机制还需要特别注意一点,在大模型返回的工具调用指令列表(可能同时需要调用多个工具)中的每一条工具调用指令中都会包含一个id,智能体执行对应工具获得工具执行结果后需要回传该结果,此时必须将该id与对应工具执行结果作为一条tool消息追加到消息列表。其具体逻辑流程如图 5所示,其目的是让大模型能够理解每一个工具所对应的执行结果分别是什么,这也是在回传工具执行结果之前必须将大模型返回的工具调用指令消息原样追加到消息列表中的原因。

图 5 工具ID传递表明工具执行结果与工具的对应关系

至此,基于函数调用机制的智能体代码解析完毕。参照前述案例测试代码,调用run_agent_fc()即可获得与前述案例相同的效果,其运行效果截图不再贴出。

本示例完整源代码可参见:https://gitcode.com/gtyan/AgentHandBook/tree/main/01-mini-agent-fc

3. 两种调用工具方式的对比与总结

总结上述两种工具调用方式(“基于特别设计提示词”方式以下称为A;“基于模型的函数调用”方式以下称为B),有以下对比与应用建议。

1)适用范围

A:适配所有能聊天的模型,可跨厂商、跨版本;可完全自定义协议。

B:依赖厂商实现与语义(OpenAI/DeepSeek 等),更标准化,但存在供应商能力差异与锁定度。

2)稳定性与脆弱性

A:对提示词敏感,最常见问题是“非 JSON 输出”“字段缺失”“格式漂移”。需要较多“格式强约束 + 重试”。

B:返回结构化的tool_calls,自带函数名与参数JSON,更稳定;但仍可能出现参数幻觉或多工具顺序不当,需要校验与回退。

3)开发与维护成本

A:前期快,协议自定义。但要自己实现JSON 解析、参数校验、路由、错误回传与格式对齐。协议一旦演进,需要自行管理版本。

B:前期要补齐tools规范、tool_choice策略与消息回放(含tool_call_id对齐);一旦调通,后续扩展与维护更省心。

4)多工具/并发调用

A:可实现,但需要自行设计多调用语法及调用结果与调用指令对齐逻辑,复杂度上升。

B:天然支持多个tool_calls,并通过tool_call_id明确“哪个调用对应哪个结果”,易观测、易回放。

5)参数与模式约束

A:靠提示词+样例引导,约束力弱;需在服务端再做强校验(例如使用Pydantic/JSON Schema)以保证安全、正确性。

B:直接约束参数形态;服务端仍应复核。

6)安全与治理

A:可完全私有化,不出网;但要自建可观测性(思考/行动/观察回放)与审计字段。

B:结构化事件天然利于埋点、审计与回放;但要评估外呼合规、厂商日志留存策略与数据边界。

7)可移植性与演进

A:高度可移植;更适合“先有协议、后换模型”的场景。

B:易用但绑定接口语义;如需迁移到不支持函数调用的模型,需要降级适配层。

8)性能与成本

A:工具说明放在提示词里,token成本与响应波动取决于提示词长度与模型发挥。

B:tools规范通常单独传递,调用路径更确定;但多轮tool_calls也会拉长时延与
token。

具体设计时,在技术选型上则可根据不同场景做出相应选择。

1)优先选择“基于特别设计提示词”的场景

  • 需要离线/私有化部署大模型或目标模型不支持函数调用。
  • 要做跨供应商/跨模型的通用Demo或教学,协议完全由开发者控制。
  • 想要在非常规调用模式(如复合输出、流式中途换协议)上做实验。

2)优先选择“基于模型的函数调用”的场景

  • 使用的主要模型原生支持函数调用,且要做生产级多工具编排。
  • 重视参数强约束、可观测性、可回放与多工具结果对齐(tool_call_id)。
  • 需要更稳定的解析路径与更低的提示词脆弱性。

最后,以下最佳实践可供实现工具调用时参考。

1)抽象一层“工具调用适配器”:把“基于特别设计提示词”的JSON输出和“基于模型的函数调用”的tool_calls都统一成内部的AgentAction(name, args, id)结构;业务只面向这层编程,后续可随时切换/混用。

2)永远要有服务端强校验:例如Pydantic/JSON Schema校验结合友好错误回传(把错误作为“观察”反馈给模型,促使其自我修正)。

3)可观测性与回放:记录“思考→行动→观察→结论”每一步;“基于特别设计提示词”要保存模型原始JSON与路由日志,“基于模型的函数调用”要保留tool_call_id映射。

4)降级与回退:函数调用异常时自动降级到提示词协议;提示词协议连错两次时直接收敛final并给出保底答案/人工转派。

5)安全边界:高风险工具(Shell/SQL/写文件)一律沙箱化并考虑使用白名单参数;无论哪种方式,都不能只信任模型生成的参数。

完整源代码:

https://gitcode.com/gtyan/AgentHandBook/tree/main/01-mini-agent

https://gitcode.com/gtyan/AgentHandBook/tree/main/01-mini-agent-fc