MYAIGC

主题

工具调用与规划 ​

Agent的能力来源于工具，规划能力决定Agent的智能程度。本节深入讲解工具调用和任务规划。

工具调用机制 ​

工具调用的本质 ​

1. 模型分析用户请求
2. 决定是否需要调用工具
3. 构造工具调用参数
4. 执行工具
5. 将结果返回给模型
6. 模型基于结果继续处理

OpenAI Function Calling ​

python
from openai import OpenAI

client = OpenAI()

# 定义工具
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取城市天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名称"}
                },
                "required": ["city"]
            }
        }
    }
]

# 发送请求
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "北京天气怎么样？"}],
    tools=tools
)

# 检查是否需要调用工具
if response.choices[0].message.tool_calls:
    tool_call = response.choices[0].message.tool_calls[0]
    print(f"工具: {tool_call.function.name}")
    print(f"参数: {tool_call.function.arguments}")

使用LangChain工具 ​

python
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

@tool
def get_weather(city: str) -> str:
    """获取城市天气信息"""
    weather_data = {"北京": "晴天25°C", "上海": "多云28°C"}
    return weather_data.get(city, "未知城市")

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except:
        return "计算错误"

# 绑定工具
model = ChatOpenAI(model="gpt-4o")
model_with_tools = model.bind_tools([get_weather, calculate])

# 调用
response = model_with_tools.invoke("北京天气加上5度是多少？")

# 处理工具调用
if response.tool_calls:
    for tool_call in response.tool_calls:
        if tool_call["name"] == "get_weather":
            result = get_weather.invoke(tool_call["args"])
            print(f"天气: {result}")

复杂工具设计 ​

工具参数验证 ​

python
from langchain_core.tools import tool
from pydantic import BaseModel, Field, field_validator

class EmailInput(BaseModel):
    """发送邮件的输入参数"""
    to: str = Field(description="收件人邮箱")
    subject: str = Field(description="邮件主题")
    body: str = Field(description="邮件正文")
    
    @field_validator('to')
    @classmethod
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('无效的邮箱地址')
        return v

@tool(args_schema=EmailInput)
def send_email(to: str, subject: str, body: str) -> str:
    """发送邮件"""
    # 实际发送逻辑
    return f"邮件已发送至 {to}"

嵌套工具调用 ​

python
# Agent可能先调用一个工具获取信息，再用这个信息调用另一个工具
# 例如：先搜索用户信息，再根据ID查询订单

@tool
def get_user_id(username: str) -> int:
    """根据用户名获取用户ID"""
    return {"张三": 1, "李四": 2}.get(username, -1)

@tool
def get_orders(user_id: int) -> list:
    """根据用户ID获取订单"""
    orders = {
        1: [{"id": 101, "product": "手机"}, {"id": 102, "product": "电脑"}],
        2: [{"id": 103, "product": "耳机"}]
    }
    return orders.get(user_id, [])

# Agent会自动串行调用这两个工具

工具组合 ​

python
from langchain_core.tools import StructuredTool

class MathToolkit:
    """数学工具集"""
    
    @staticmethod
    def add(a: float, b: float) -> float:
        return a + b
    
    @staticmethod
    def multiply(a: float, b: float) -> float:
        return a * b
    
    def get_tools(self):
        return [
            StructuredTool.from_function(self.add, name="add"),
            StructuredTool.from_function(self.multiply, name="multiply")
        ]

任务规划 ​

ReAct规划 ​

python
from langchain import hub
from langchain.agents import create_react_agent

# 使用标准ReAct提示词
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(model, tools, prompt)

Plan-and-Solve ​

python
from langchain_core.prompts import ChatPromptTemplate

plan_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个任务规划专家。

将用户任务分解为清晰的步骤。

格式：
Step 1: [步骤描述]
Step 2: [步骤描述]
..."""),
    ("user", "{task}")
])

def plan_task(task: str):
    """规划任务"""
    return model.invoke(plan_prompt.format(task=task))

def execute_plan(plan: str, tools):
    """执行计划"""
    steps = parse_steps(plan)
    results = []
    for step in steps:
        result = agent.invoke({"input": step})
        results.append(result)
    return results

思维链规划 ​

python
def chain_of_thought_planning(task: str):
    """使用思维链进行规划"""
    prompt = f"""任务：{task}

让我们一步步思考：
1. 首先，分析任务需要什么信息
2. 然后，确定需要使用哪些工具
3. 接着，规划执行顺序
4. 最后，考虑可能的异常处理

思考过程："""
    
    return model.invoke(prompt)

高级规划模式 ​

自我反思 ​

python
def agent_with_reflection(task: str, max_iterations: int = 5):
    """带自我反思的Agent"""
    for i in range(max_iterations):
        # 执行
        result = agent.invoke({"input": task})
        
        # 反思
        reflection = model.invoke(f"""
        任务：{task}
        执行结果：{result}
        
        请评估这个结果：
        1. 是否完整解决了任务？
        2. 是否有错误？
        3. 是否需要改进？
        
        如果结果满意，回复"COMPLETE"。否则给出改进建议。
        """)
        
        if "COMPLETE" in reflection.content:
            return result
        
        # 根据反思调整
        task = reflection.content
    
    return result

动态工具选择 ​

python
def select_tools_for_task(task: str, all_tools: list) -> list:
    """根据任务动态选择工具"""
    tool_descriptions = "\n".join([
        f"- {t.name}: {t.description}" 
        for t in all_tools
    ])
    
    prompt = f"""任务：{task}

可用工具：
{tool_descriptions}

请选择完成此任务需要的工具，只输出工具名称列表。"""

    selected = model.invoke(prompt)
    selected_names = parse_tool_names(selected.content)
    
    return [t for t in all_tools if t.name in selected_names]

完整示例：智能助手 ​

python
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.memory import ConversationBufferMemory

# 定义工具
@tool
def search_web(query: str) -> str:
    """搜索互联网获取信息"""
    # 实际搜索逻辑
    return f"关于'{query}'的搜索结果..."

@tool
def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算错误: {e}"

@tool
def send_notification(message: str) -> str:
    """发送通知"""
    # 实际发送逻辑
    return f"通知已发送: {message}"

# 创建Agent
tools = [search_web, get_current_time, calculate, send_notification]
model = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个智能助手，可以使用工具帮助用户完成任务。

规则：
1. 优先使用工具获取准确信息
2. 需要计算时使用calculate工具
3. 需要搜索信息时使用search_web工具
4. 需要时间信息时使用get_current_time工具
5. 需要通知用户时使用send_notification工具
6. 回答要准确、简洁"""),
    ("placeholder", "{chat_history}"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_tool_calling_agent(model, tools, prompt)

# 添加记忆
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True,
    max_iterations=10
)

# 使用
def chat(user_input: str):
    result = agent_executor.invoke({"input": user_input})
    return result["output"]

# 测试
print(chat("现在几点了？"))
print(chat("帮我搜索一下LangChain是什么"))
print(chat("100 * 50 等于多少？"))

小结 ​

概念说明
工具调用模型决定调用哪个工具及参数
参数验证确保工具输入合法
任务规划分解复杂任务为步骤
自我反思Agent评估和改进自己的执行

下一步 ​

继续学习 多Agent协作，了解多个Agent如何协同工作。