Agent 基础与工具调用

答疑机器人能回答员工手册里的问题了，但知识库之外的事情它无能为力——不能搜互联网、也不能查数据库。这不是 RAG 的问题，而是大模型的本质限制：它只能基于已有文本对话，无法主动与外部环境交互。这节课我主要搞清楚了 Agent 工具调用从零到工业标准的完整链路。

这篇讲什么

• Agent 为什么能"动手"：工具定义 → 工具选择 → 结构化输出 → 工具执行四步链路
• Function Calling 的完整机制与 JSON Schema 设计（核心重点）
• ReAct 框架：思考-行动-观察循环
• MCP 协议：为什么要解耦工具定义和使用

环境初始化略，见课程仓库。我用阿里云 PAI-DSW 跑的代码，CPU 实例就够，新用户有 3 个月免费额度。

1 第一个工具函数：最简单的实现

让机器人能联网搜索，最直接的思路是：写一个 web_search 函数，每次都调用它，把结果和问题一起发给大模型。

# 1. 用户的原始请求
user_request = "你好，请帮我搜集一些关于 Transformer 模型的最新资料。"

# 2. "硬编码"执行工具函数
def web_search(query: str):
    """模拟执行网络搜索并返回JSON格式的结果"""
    print(f"--- [工具执行中] 正在搜索: {query} ---")
    return '''{
        "results": [
            {"title": "Attention Is All You Need (Transformer 论文原文)", "url": "https://arxiv.org/abs/1706.03762", "snippet": "The dominant sequence transduction models..."},
            {"title": "The Illustrated Transformer", "url": "https://jalammar.github.io/illustrated-transformer/", "snippet": "A visual and intuitive explanation."}
        ]
    }'''

tool_result = web_search(query=user_request)

# 3. 将用户请求和工具结果拼接，发送给大模型
completion = client.chat.completions.create(
    model="qwen-plus", 
    messages=[
        {'role': 'system', 'content': '你是一位课程研究助理，根据工具结果生成友好的回复。'},
        {'role': 'user', 'content': f'用户原始请求: "{user_request}"\n工具执行结果: {tool_result}'}
    ]
)
print(completion.choices[0].message.content)

局限性：只适合"有且只有一个工具、每次必须调用"的场景。有多个工具时，谁来决定调哪个？

2 意图识别：让 Agent 决定用什么工具

2.1 关键词匹配：脆弱的方案

# 伪代码：脆弱的关键词路由器
def route_to_tool(user_input):
    if "论文" in user_input or "arxiv" in user_input:
        return "search_arxiv_paper"
    elif "搜索" in user_input or "资料" in user_input:
        return "web_search"
    else:
        return "no_tool_needed"

用户说"我想看看那篇 Attention is All You Need 讲了什么"，这个路由器就识别不出来了——没有预设关键词。

2.2 用大模型来做意图识别

在提示词里列出所有工具，让大模型决定调哪个。这更灵活，但问题是模型返回的结果格式不固定，无法程序化解析。

3 结构化输出：让工具调用可靠

核心问题：大模型倾向于生成多样化文本，我需要的是确定性的 JSON 结构。

解决方案是"引导-校验-重试"闭环：

import json
from pydantic import BaseModel, Field, ValidationError, TypeAdapter
from typing import Union, Literal

class WebSearchParams(BaseModel):
    query: str = Field(description="用于网络搜索的关键词。")

class SearchArxivParams(BaseModel):
    query: str = Field(description="用于在 Arxiv.org 上搜索的论文标题或关键词。")

class WebSearchCall(BaseModel):
    tool_name: Literal["web_search"]
    parameters: WebSearchParams

class SearchArxivCall(BaseModel):
    tool_name: Literal["search_arxiv_paper"]
    parameters: SearchArxivParams

ToolCall = Union[WebSearchCall, SearchArxivCall]

def get_structured_output(user_request: str, max_retries: int = 2):
    messages = [{'role': 'user', 'content': build_prompt(user_request)}]
    adapter = TypeAdapter(ToolCall)
    
    for attempt in range(max_retries):
        response = client.chat.completions.create(
            model="qwen-plus", messages=messages, temperature=0
        )
        raw_output = response.choices[0].message.content
        
        try:
            data = json.loads(raw_output.strip('```json').strip('```'))
            validated_data = adapter.validate_python(data)
            return validated_data.model_dump()
        except (json.JSONDecodeError, ValidationError) as e:
            # 验证失败，把错误信息带回给模型让它修正
            messages.extend([
                {'role': 'assistant', 'content': raw_output},
                {'role': 'user', 'content': f"格式错误: {e}，请严格按照JSON格式重新输出"}
            ])
    return None

4 Function Calling：行业标准

手动实现那套"意图识别 → 结构化输出 → 验证重试"比较繁琐。主流大模型服务商（阿里云、OpenAI 等）已在 API 里内置了这个能力，就是 Function Calling。

import json

tools = [
    {
        "type": "function",
        "function": {
            "name": "search_arxiv_paper",
            "description": "在 Arxiv.org 上搜索学术论文",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "论文的标题或关键词"},
                },
                "required": ["query"],
            },
        }
    }
]

messages = [{"role": "user", "content": "帮我找一下那篇经典的 Transformer 论文 'Attention Is All You Need'"}]
response = client.chat.completions.create(
    model="qwen-plus", messages=messages, tools=tools, tool_choice="auto"
)
response_message = response.choices[0].message

if response_message.tool_calls:
    tool_call = response_message.tool_calls[0]
    function_name = tool_call.function.name
    function_args = json.loads(tool_call.function.arguments)
    
    # 执行工具（模拟）
    tool_result = json.dumps({"paper_id": "1706.03762", "url": "https://arxiv.org/abs/1706.03762"})
    
    # 把工具结果传回给模型，让它生成最终回复
    messages.append(response_message)
    messages.append({"tool_call_id": tool_call.id, "role": "tool", "name": function_name, "content": tool_result})
    
    final_response = client.chat.completions.create(model="qwen-plus", messages=messages)
    print(final_response.choices[0].message.content)

ReAct = 思考-行动-观察循环：工具调用的结果通过再次 API 调用传给大模型，模型"观察"结果后"思考"是否完成，然后决定"行动"（继续调工具）或返回最终答案。

用 AgentScope 框架可以把这套逻辑大幅简化：

import asyncio
from agentscope.agent import ReActAgent
from agentscope.tool import Toolkit, ToolResponse
from agentscope.model import DashScopeChatModel
from agentscope.message import Msg, TextBlock
from agentscope.formatter import DashScopeChatFormatter

def search_arxiv_paper(query: str) -> ToolResponse:
    """在 Arxiv.org 上搜索学术论文。
    
    Args:
        query (str): 搜索关键词
    """
    print(f"--- [工具执行中] 正在 Arxiv 搜索: {query} ---")
    paper_url = "https://arxiv.org/abs/1706.03762"
    return ToolResponse(content=[TextBlock(type="text", text=f"找到论文 '{query}'：{paper_url}")])

async def run_agentscope_example():
    toolkit = Toolkit()
    toolkit.register_tool_function(search_arxiv_paper)
    
    agent = ReActAgent(
        name="Course Research Agent",
        sys_prompt="你是一个课程研究助理，擅长帮人搜集和整理学习资料。",
        model=DashScopeChatModel(model_name="qwen-plus", api_key=os.environ.get("DASHSCOPE_API_KEY")),
        toolkit=toolkit,
        formatter=DashScopeChatFormatter()
    )
    
    msg = Msg(name="user", content="帮我找一下那篇经典的 Transformer 论文 'Attention Is All You Need'", role="user")
    await agent(msg)

await run_agentscope_example()

AgentScope 相比手动实现的优势：工具只需写带 docstring 的普通函数（自动解析 Schema）、自动处理 tool_calls 解析和多轮调用。

为什么还要学手动实现？ 因为框架出问题时需要看底层，生产环境常需加权限验证/缓存/日志等定制逻辑。

5 MCP 协议：工具的规模化管理

5.1 问题：Schema 硬编码

Function Calling 模式下，我在 Agent 侧硬编码了 web_search 的 JSON Schema。工具一多，维护成本爆炸。工具提供方也没法主动"接入" AI 生态。

5.2 MCP 的解耦思想

MCP（Model Context Protocol）的核心：谁提供工具，谁定义工具。

• MCP Server（工具提供方）：声明工具的名称、描述、参数
• MCP Client（Agent 侧）：连接 Server，动态拉取工具清单，无需硬编码 Schema

类比：Function Calling 是主板内部总线，MCP 是 USB 标准接口——后者允许任何第三方设备即插即用。

5.3 用 MCP 重构 web_search

Step 1：本地调试（stdio 模式）

# run_mcp_server_example.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("MockWebSearch")

@mcp.tool()
def web_search(query: str, max_results: int = 3) -> str:
    """模拟联网搜索，根据关键词返回搜索结果。
    Args:
        query: 搜索关键词
        max_results: 最大返回结果数量，默认为 3
    """
    # 模拟返回数据...
    return f"搜索「{query}」的模拟结果"

if __name__ == "__main__":
    mcp.run(transport="stdio")

import asyncio, os, sys
from agentscope.agent import ReActAgent
from agentscope.mcp import StdIOStatefulClient
from agentscope.tool import Toolkit
from agentscope.model import DashScopeChatModel
from agentscope.message import Msg
from agentscope.formatter import DashScopeChatFormatter

async def run_local_mcp_example():
    # 1. 启动本地 MCP Server
    web_search_client = StdIOStatefulClient(
        name="web_search_service",
        command=sys.executable,
        args=["run_mcp_server_example.py"],
        cwd=os.getcwd(),
    )
    await web_search_client.connect()
    
    # 2. 工具发现：Client 自动从 Server 拉取工具清单
    toolkit = Toolkit()
    await toolkit.register_mcp_client(web_search_client)
    
    # 3. 创建 Agent
    agent = ReActAgent(
        name="Research Assistant Agent",
        sys_prompt="你是一个课程研究助理。",
        model=DashScopeChatModel(model_name="qwen-plus", api_key=os.environ.get("DASHSCOPE_API_KEY")),
        toolkit=toolkit,
        formatter=DashScopeChatFormatter()
    )
    
    msg = Msg(name="user", content="搜索一下最近关于'大型语言模型'的最新进展。", role="user")
    await agent(msg)
    await web_search_client.close()

await run_local_mcp_example()

Step 2：接入线上服务（Streamable HTTP 模式）

只需把 Client 从 StdIOStatefulClient 换成 HttpStatelessClient，其他代码不变：

from agentscope.mcp import HttpStatelessClient

client = HttpStatelessClient(
    name="web_search_service",
    transport="streamable_http",
    url="https://dashscope.aliyuncs.com/api/v1/mcps/WebSearch/mcp",
    headers={"Authorization": "Bearer " + os.environ.get("DASHSCOPE_API_KEY")}
)

• stdio：Server 和 Client 在同一台机器，适合本地调试（Claude Desktop、Cursor 等 IDE 也用这种方式接入本地 MCP）
• Streamable HTTP：Server 部署在远端，适合多 Agent/多用户共用同一 MCP Server 的生产环境

个人总结

• 工具调用的演进路径：硬编码 → 关键词匹配 → 大模型意图识别 → 结构化输出 → Function Calling → MCP，每一步都是被上一步的局限性逼出来的
• Function Calling 本质上是大模型服务商把"引导-校验-重试"内置了，省去了大量手动代码，但理解底层原理是调试的前提
• ReAct 循环（思考-行动-观察）是 Agent 能"持续干活"的关键机制，单次 Function Calling 只是这个循环的一次迭代
• MCP 最大的价值是"谁提供谁定义"，把工具 Schema 的维护责任从 Agent 开发者转移给工具提供方，大幅降低长期维护成本
• AgentScope 这类框架值得用，但至少要知道它封装了什么，出了问题才知道去哪里找