AI Agent:从工具调用到自主任务执行(实战篇)

pip install langchain langchain-openai mcp uvicorn fastmcp

export OPENAI_API_KEY="sk-..."

# travel_mcp_server.pyfrom mcp.server.fastmcp import FastMCPimport random# 初始化 MCP 服务器mcp = FastMCP("TravelAssistantServer")@mcp.tool()def get_weather(city: str) -> str:    """    查询指定城市的天气情况。    Args:        city: 城市名称，例如 '北京', '上海'    Returns:        天气描述字符串    """    # 模拟真实 API 调用    weathers = ["晴朗", "多云", "小雨", "大雨", "雪"]    temp = random.randint(10, 30)    status = random.choice(weathers)    return f"{city}今天天气{status}，气温{temp}摄氏度。"@mcp.tool()def search_attractions(city: str, preference: str = "自然风光") -> list:    """    搜索指定城市的旅游景点。    Args:        city: 城市名称        preference: 偏好类型，如 '自然风光', '历史人文', '室内娱乐'    Returns:        景点列表    """    # 模拟数据库查询逻辑    db = {        "北京": {            "自然风光": ["颐和园", "香山公园"],            "历史人文": ["故宫", "长城", "天坛"],            "室内娱乐": ["国家博物馆", "环球影城"]        },        "上海": {            "自然风光": ["外滩", "世纪公园"],            "历史人文": ["豫园", "中共一大会址"],            "室内娱乐": ["上海迪士尼", "海洋水族馆"]        }    }    # 简单的容错处理    if city not in db:        return ["未知城市，无法推荐具体景点，建议查询通用攻略。"]    # 如果偏好不匹配，默认返回第一项    attractions = db[city].get(preference, list(db[city].values())[0])    return attractionsif __name__ == "__main__":    # 启动 MCP 服务器 (使用 stdio 模式，适合本地调试)    # 生产环境通常使用 SSE (http) 模式    mcp.run()

# agent_runner.pyimport asynciofrom langchain_openai import ChatOpenAIfrom langchain.agents import create_tool_calling_agent, AgentExecutorfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_mcp import MCPToolkit  # 假设 langchain-mcp 集成包已可用，或使用标准加载方式from mcp.client.stdio import stdio_clientfrom langchain.tools import Tool# --- 1. 连接 MCP Server 并加载工具 ---async def load_mcp_tools():    """    连接到本地运行的 MCP Server 并转换为 LangChain Tools    """    tools = []    # 这里模拟通过 stdio 连接到我们刚才写的 server    # 在实际生产中，你可能需要配置具体的 command 和 args    from mcp.client.stdio import StdioServerParameters    server_params = StdioServerParameters(        command="python",        args=["travel_mcp_server.py"]    )    async with stdio_client(server_params) as (read, write):        from mcp.client.session import ClientSession        async with ClientSession(read, write) as session:            await session.initialize()            # 获取工具列表            response = await session.list_tools()            langchain_tools = []            for tool_def in response.tools:                # 将 MCP 工具定义转换为 LangChain 可调用的对象                # 这里为了演示简化了转换逻辑，实际需封装调用 session.call_tool                async def make_func(t_name, t_desc):                    async def func(**kwargs):                        async with stdio_client(server_params) as (r, w):                            async with ClientSession(r, w) as s:                                await s.initialize()                                result = await s.call_tool(t_name, kwargs)                                return str(result.content)                    return func                tool = Tool(                    name=tool_def.name,                    description=tool_def.description,                    coroutine=make_func(tool_def.name, tool_def.description)                )                langchain_tools.append(tool)            return langchain_tools# --- 2. 定义 Agent 核心逻辑 ---async def run_agent():    # 加载工具    print("🔌 正在连接 MCP 服务器加载工具...")    tools = await load_mcp_tools()    print(f"✅ 成功加载 {len(tools)} 个工具: {[t.name for t in tools]}")    # 初始化 LLM (大脑)    llm = ChatOpenAI(model="gpt-4o", temperature=0)    # 定义 Prompt (赋予角色)    prompt = ChatPromptTemplate.from_messages([        ("system", "你是一个专业的旅行规划助手。你需要根据用户的需求，自主调用工具查询天气和景点，并给出合理的建议。如果天气不好，请优先推荐室内活动。"),        ("human", "{input}"),        ("placeholder", "{agent_scratchpad}") # 用于存放思考过程    ])    # 构建 Agent (ReAct 模式)    # LangChain 1.0 推荐使用 create_tool_calling_agent    agent = create_tool_calling_agent(llm, tools, prompt)    # 创建执行器 (Executor)    agent_executor = AgentExecutor(        agent=agent,        tools=tools,        verbose=True, # 开启详细日志，观察思考过程        handle_parsing_errors=True    )    # --- 3. 执行任务 ---    user_query = "我想去北京玩，帮我查下天气，如果下雨就推荐室内景点，否则推荐自然风光，并简单估算一下门票花费（假设每个景点 50 元）。"    print(f"\n👤 用户指令：{user_query}\n")    print("🤖 Agent 开始思考与执行...\n")    response = await agent_executor.ainvoke({"input": user_query})    print("\n" + "="*50)    print("📝 最终回复:")    print(response["output"])    print("="*50)if __name__ == "__main__":    asyncio.run(run_agent())

python agent_runner.py

🔌 正在连接 MCP 服务器加载工具...✅ 成功加载 2 个工具: ['get_weather', 'search_attractions']👤 用户指令：我想去北京玩，帮我查下天气，如果下雨就推荐室内景点，否则推荐自然风光，并简单估算一下门票花费（假设每个景点 50 元）。🤖 Agent 开始思考与执行...--- [Thought Step 1] ---Agent 思考: 用户想去北京。我需要先查天气，根据天气决定推荐什么类型的景点。Action: get_weatherAction Input: {"city": "北京"}--- [Observation] ---Tool Output: 北京今天天气小雨，气温 18 摄氏度。--- [Thought Step 2] ---Agent 思考: 天气是"小雨"，根据指令，我应该推荐"室内娱乐"类的景点。Action: search_attractionsAction Input: {"city": "北京", "preference": "室内娱乐"}--- [Observation] ---Tool Output: ["国家博物馆", "环球影城"]--- [Thought Step 3] ---Agent 思考: 找到了两个室内景点。用户要求估算花费，假设每个 50 元。总共 2 个景点，花费 = 2 * 50 = 100 元。现在可以生成最终回答了。Final Answer: 您好！为您查询到北京今天有**小雨**，气温 18℃。考虑到天气因素，我为您推荐以下**室内景点**：1. **国家博物馆**2. **环球影城**💰 **预算估算**：按每个景点门票约 50 元计算，两个景点总计约 **100 元**。祝您在北京旅途愉快，记得带伞哦！☔️==================================================📝 最终回复:您好！为您查询到北京今天有**小雨**，气温 18℃... (同上)==================================================