第8章: Agent 智能体入门：工具调用让AI会办事

# 一个典型的 MCP 工作流（概念级）

1) Agent 连接 MCP Server
2) 列举可用 Tools / Resources
3) 模型输出下一步动作：调用 tool(name, args)
4) Agent 执行 tool 调用 -> 得到结果
5) 将结果作为 Observation 追加到上下文 -> 继续 Think/Act/Observe

import dashscope
from dashscope import Generation
import json
import os

# 设置API Key
dashscope.api_key = os.getenv('DASHSCOPE_API_KEY')

class SimpleAgent:
    def __init__(self):
        """初始化Agent"""
        self.messages = []
        self.tools = {
            "calculator": self.calculator,
            "get_weather": self.get_weather
        }
    
    def calculator(self, expression):
        """计算器工具"""
        try:
            result = eval(expression)
            return f"计算结果：{result}"
        except Exception as e:
            return f"计算错误：{str(e)}"
    
    def get_weather(self, city):
        """获取天气工具（模拟）"""
        # 实际应用中应该调用真实的天气API
        weather_data = {
            "北京": "晴天，15-25℃",
            "上海": "多云，18-26℃",
            "深圳": "小雨，22-28℃"
        }
        return weather_data.get(city, f"{city}的天气信息暂时无法获取")

    def get_tool_definitions(self):
        """定义可用工具的描述"""
        return [
            {
                "type": "function",
                "function": {
                    "name": "calculator",
                    "description": "执行数学计算，支持加减乘除等基本运算",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "expression": {
                                "type": "string",
                                "description": "要计算的数学表达式，如：'3.14 * 2 + 5'"
                            }
                        },
                        "required": ["expression"]
                    }
                }
            },
            {
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "description": "获取指定城市的天气信息",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "city": {
                                "type": "string",
                                "description": "城市名称，如：北京、上海"
                            }
                        },
                        "required": ["city"]
                    }
                }
            }
        ]

    def run(self, user_input):
        """运行Agent - ReAct循环"""
        print(f"🙋 用户：{user_input}\n")
        
        # 添加用户消息
        self.messages.append({
            "role": "user",
            "content": user_input
        })
        
        # ReAct循环：最多执行5轮
        for i in range(5):
            print(f"--- 第{i+1}轮思考 ---")
            
            # 调用LLM进行推理
            response = Generation.call(
                model='qwen-plus',
                messages=self.messages,
                tools=self.get_tool_definitions(),
                result_format='message'
            )
            
            # 基础错误检查
            if response is None:
                print("❌ 错误：API响应为空")
                return "抱歉，无法获取响应"

            if response.status_code != 200:
                print(f"❌ 错误：API请求失败，状态码：{response.status_code}")
                if hasattr(response, 'message') and response.message:
                    print(f"错误信息：{response.message}")
                return "抱歉，API请求失败"

            if response.output is None:
                print("❌ 错误：API响应output为空")
                return "抱歉，无法获取响应内容"
                
            assistant_message = response.output.choices[0].message

            # 如果assistant_message是对象，转换为字典
            if not isinstance(assistant_message, dict):
                assistant_message = assistant_message.__dict__
            
            # 检查是否需要调用工具
            tool_calls = assistant_message.get('tool_calls')
            if tool_calls:
                # 有工具调用
                tool_call = tool_calls[0]
                # tool_call 可能是 dict，也可能是对象
                if isinstance(tool_call, dict):
                    function_name = tool_call['function']['name']
                    arguments = json.loads(tool_call['function']['arguments'])
                else:
                    function_name = tool_call.function.name
                    arguments = json.loads(tool_call.function.arguments)
                
                print(f"💭 思考：需要使用工具 {function_name}")
                print(f"🎬 行动：调用 {function_name}，参数：{arguments}")
                
                # 执行工具
                if function_name in self.tools:
                    result = self.tools[function_name](**arguments)
                    print(f"👀 观察：{result}\n")
                    
                    # 将工具调用和结果添加到消息历史
                    self.messages.append({
                        "role": "assistant",
                        "content": "",
                        "tool_calls": [tool_call]
                    })
                    self.messages.append({
                        "role": "tool",
                        "content": result,
                        "name": function_name
                    })
                else:
                    print(f"❌ 错误：工具 {function_name} 不存在\n")
                    break
            else:
                # 没有工具调用，返回最终答案
                final_answer = assistant_message.get('content', '无法获取回答')
                print(f"✅ 最终答案：{final_answer}\n")
                return final_answer
        
        return "抱歉，无法完成任务"

# 创建Agent
agent = SimpleAgent()

# 测试1：需要使用计算器
print("=" * 50)
print("测试1：计算任务")
print("=" * 50)
agent.run("请帮我计算 123.456 * 789 + 456 的结果")

# 测试2：需要查询天气
print("\n" + "=" * 50)
print("测试2：天气查询")
print("=" * 50)
agent = SimpleAgent()  # 重新创建，清空历史
agent.run("北京今天天气怎么样？")

# 测试3：不需要工具
print("\n" + "=" * 50)
print("测试3：普通对话")
print("=" * 50)
agent = SimpleAgent()
agent.run("请介绍一下Python编程语言")

import json
import os
from typing import Any, Dict, List

import dashscope
from dashscope import Generation


dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")


def get_weather(city: str) -> str:
    # 示例：真实项目请调用天气 API
    return "晴 15-25℃"


def calculator(expression: str) -> float:
    # 注意：这里只是教学示例，不要在生产环境 eval
    return eval(expression)


def export_markdown(content: str) -> str:
    # 教学版：真实项目可以写文件、上传对象存储、生成下载链接等
    return "已导出 markdown（示例）"


class SimpleReActAgent:
    def __init__(self):
        # memory：用于保存“可检验的中间状态/工具执行轨迹”。
        # - 这不是为了让模型“凭空更聪明”，而是为了让多轮任务更稳定、可复盘。
        # - 建议结构化存储（step/tool/args/observation），方便调试与复现。
        # - 若要做“长期记忆”，应把摘要写入数据库/向量库，而不是一直塞在内存列表里。
        self.memory: List[Dict[str, Any]] = []
        self.messages: List[Dict[str, Any]] = []
        self.tools = {
            "get_weather": get_weather,
            "calculator": calculator,
            "export_markdown": export_markdown,
        }

    def get_tool_definitions(self):
        return [
            {
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "description": "获取指定城市的天气信息",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "city": {"type": "string", "description": "城市名称，如：北京、上海"}
                        },
                        "required": ["city"],
                    },
                },
            },
            {
                "type": "function",
                "function": {
                    "name": "calculator",
                    "description": "执行数学计算，支持加减乘除等基本运算",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "expression": {"type": "string", "description": "要计算的数学表达式"}
                        },
                        "required": ["expression"],
                    },
                },
            },
            {
                "type": "function",
                "function": {
                    "name": "export_markdown",
                    "description": "导出 markdown 内容（教学示例）",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "content": {"type": "string", "description": "要导出的 markdown 正文"}
                        },
                        "required": ["content"],
                    },
                },
            },
        ]

    def run(self, user_goal: str, max_steps: int = 8):
        if not dashscope.api_key:
            raise RuntimeError("请先设置环境变量 DASHSCOPE_API_KEY")

        # messages：短期记忆（Conversation Memory）
        # - 本次对话的上下文在这里滚动累积，并随每轮请求发给模型。
        # - 真正工程化时会做：滑动窗口、总结摘要、敏感信息过滤、工具结果裁剪。
        self.messages = [
            {
                "role": "system",
                "content": """你是一个严谨的规划型 Agent。请按照以下规则工作：
1. 如果用户没有提供必要信息（如城市、预算等），请先询问这些信息
2. 只有获得必要信息后，才开始调用工具
3. 每轮你可以选择调用一个工具来推进任务
4. 最后导出完整的出行计划清单

必要信息：
- 出行城市（必须）
- 预算范围（可选，如果没有会按中等预算估算）
- 出行人数（可选，默认1人）""",
            },
            {"role": "user", "content": user_goal},
        ]

        for step in range(max_steps):
            resp = Generation.call(
                model="qwen-plus",
                messages=self.messages,
                tools=self.get_tool_definitions(),
                result_format="message",
            )
            assistant_message = resp.output.choices[0].message
            if not isinstance(assistant_message, dict):
                assistant_message = assistant_message.__dict__

            tool_calls = assistant_message.get("tool_calls")
            if not tool_calls:
                # 如果没有工具调用，检查是否是询问信息的回复
                content = assistant_message.get("content", "")
                self.messages.append({
                    "role": "assistant", 
                    "content": content
                })
                
                # 如果是询问信息，我们需要模拟用户提供信息来继续
                if "城市" in content or "预算" in content or "人数" in content:
                    # 模拟用户提供完整信息
                    simulated_response = "北京，预算2000元，2人"
                    self.messages.append({
                        "role": "user", 
                        "content": simulated_response
                    })
                    print(f"👤 模拟用户回复：{simulated_response}")
                    continue  # 继续下一轮
                else:
                    return content

            tool_call = tool_calls[0]
            if isinstance(tool_call, dict):
                tool_name = tool_call["function"]["name"]
                tool_args = json.loads(tool_call["function"]["arguments"])
            else:
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)

            if tool_name not in self.tools:
                return f"工具不允许：{tool_name}"

            # 过滤参数，只传递函数实际需要的参数
            import inspect
            tool_func = self.tools[tool_name]
            func_params = inspect.signature(tool_func).parameters
            filtered_args = {k: v for k, v in tool_args.items() if k in func_params}
            
            observation = tool_func(**filtered_args)
            
            print(f"🔧 调用工具：{tool_name}({filtered_args}) → {observation}")

            # 把"本轮动作与结果"写入 memory（State/Scratchpad）
            # - 这里存的是工具调用轨迹，方便后续：复盘、调试、或做摘要写入长期记忆。
            # - 注意：不要把敏感原文（Token/手机号等）直接写入长期存储；必要时先脱敏。
            self.memory.append({
                "step": step + 1,
                "tool": tool_name,
                "args": filtered_args,
                "args": tool_args,
                "observation": observation,
            })

            self.messages.append({
                "role": "assistant",
                "content": "",
                "tool_calls": [tool_call],
            })
            self.messages.append({
                "role": "tool",
                "content": str(observation),
                "name": tool_name,
            })

            if tool_name == "export_markdown":
                return observation

        return "未在步数内完成（示例）"


agent = SimpleReActAgent()
result = agent.run("帮我做一个周末出行计划：查天气、估算预算、输出清单")
print(result)
print("memory=", agent.memory)