第6章: Agent 实战入门：让模型学会“用工具做事”

import os

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_community.chat_models import ChatTongyi
from requests.exceptions import ConnectionError


@tool
def add(a: int, b: int) -> int:
    """计算两个整数的和，用于精确的数学加法运算"""
    return a + b


api_key = os.getenv("DASHSCOPE_API_KEY")
llm = ChatTongyi(model_name="qwen-turbo", dashscope_api_key=api_key, temperature=0)

agent = create_agent(
    model=llm,
    tools=[add],
    system_prompt="需要精确计算时必须调用 add 工具，直接给最终答案。",
    debug=True,
)

try:
    state = agent.invoke({"messages": [("human", "计算 12 + 30") ]})
    print("Agent 响应:", state)
    print("最终答案:", state["messages"][-1].content)
except ConnectionError as e:
    print("❌ 网络连接错误:", str(e))
    print("建议：检查网络连接或稍后重试")
except Exception as e:
    print("❌ 其他错误:", str(e))

import os
import logging

# 配置日志以便观察 Agent 执行过程
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

from langchain_experimental.agents.agent_toolkits import create_python_agent
from langchain_experimental.tools import PythonREPLTool
from langchain_community.llms import Tongyi


api_key = os.getenv("DASHSCOPE_API_KEY")
llm = Tongyi(model_name="qwen-turbo", dashscope_api_key=api_key, temperature=0)

# ============ 方式一：create_python_agent（推荐用于纯 Python 场景）============
# 这是一个"专用 Agent 工厂"，专门用于创建能执行 Python 代码的 Agent
# 内部已经封装了：
#   1. 适合代码生成的 system_prompt
#   2. 错误处理和重试逻辑
#   3. 结果格式化输出
python_agent = create_python_agent(
    # 核心参数
    llm=llm,                          # LLM 模型，负责生成 Python 代码
    tool=PythonREPLTool(),             # Python 执行工具，在沙箱中运行代码
    
    # 执行控制参数
    verbose=True,                      # 是否打印详细执行日志（开发时建议开启）
    
    # AgentExecutor 的额外配置（通过 agent_executor_kwargs 传递）
    agent_executor_kwargs={
        "handle_parsing_errors": True,  # 自动处理代码解析错误，失败时重试
        "max_iterations": 15,           # 最大迭代次数，防止无限循环
        # early_stopping_method: 达到最大迭代次数时的处理方式
        # - "generate" (默认): 调用 LLM 生成最终答案，基于已收集的信息给出一个总结性回答
        # - "force": 直接返回最后一次工具调用的结果，不做额外处理（可能不完整）
        # 建议：生产环境用 "generate"，确保用户得到有意义的回复
        "early_stopping_method": "generate",
    },
)

# 调用方式：使用 "input" 字段传递用户问题
result = python_agent.invoke({"input": "用 Python 算一下第 10 个斐波那契数"})
print(f"结果类型: {type(result)}")  # dict
print(f"输出内容: {result['output']}")  # 最终答案字符串

# ============ 方式二：create_agent + PythonREPLTool（通用方式）============
# 对比：使用通用 create_agent 达到类似效果
# 区别：需要手动配置更多参数，但更灵活

from langchain.agents import create_agent

# 通用方式创建 Agent
# 注意：这里需要手动配置 system_prompt 来指导模型生成 Python 代码
general_agent = create_agent(
    model=llm,
    tools=[PythonREPLTool()],  # 同样可以传入 PythonREPLTool
    system_prompt=(
        "你是一个 Python 代码执行助手。"
        "当需要计算或数据处理时，请编写 Python 代码并执行。"
        "直接给出代码执行结果，不需要额外解释。"
    ),
    debug=True,
)

# 调用方式不同：使用 "messages" 字段
general_result = general_agent.invoke({
    "messages": [("human", "计算第 10 个斐波那契数")]
})
print(f"通用方式结果: {general_result['messages'][-1].content}")

import os

from langchain_experimental.agents.agent_toolkits import create_csv_agent
from langchain_community.chat_models import ChatTongyi


api_key = os.getenv("DASHSCOPE_API_KEY")
llm = ChatTongyi(model_name="qwen-turbo", dashscope_api_key=api_key, temperature=0)

agent = create_csv_agent(
    llm=llm,
    path="data/house_price.csv",        # CSV 文件路径
    verbose=True,                       # 打印详细执行日志
    agent_executor_kwargs={"handle_parsing_errors": True},
    allow_dangerous_code=True,          # 允许执行代码（⚠️ 安全风险）
)

result = agent.invoke({"input": "这份数据有多少行？"})
print(result["output"])

import os

from langchain_community.chat_models import ChatTongyi
from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.tools.retriever import create_retriever_tool
from langchain.agents import create_agent


api_key = os.getenv("DASHSCOPE_API_KEY")
llm = ChatTongyi(model_name="qwen-turbo", dashscope_api_key=api_key, temperature=0)
embeddings = DashScopeEmbeddings(model="text-embedding-v1", dashscope_api_key=api_key)


# === 1️⃣ 单库检索 Agent ===
docs = ["LangChain 是构建 LLM 应用的框架。", "Agent 是能调用工具的智能体。", "Tools 是 Agent 可以调用的外部函数。"]
vs = Chroma.from_texts(docs, embeddings)

# create_retriever_tool：把 retriever 包成标准 Tool
# 参数说明：
#   - retriever: BaseRetriever 实例，负责执行检索
#   - name: str，工具名称，Agent 通过此名称识别和调用工具
#   - description: str，工具描述，帮助 Agent 理解何时使用该工具
search_tool = create_retriever_tool(
    retriever=vs.as_retriever(search_kwargs={"k": 2}),  # k=2 表示返回最相关的2条结果
    name="search_langchain_docs",
    description="在 LangChain 知识库中搜索相关文档，回答关于 LangChain 的问题。",
)

agent1 = create_agent(
    model=llm,
    tools=[search_tool],
    system_prompt="你是一个知识库助手。需要回答问题时，先用 search_langchain_docs 检索相关文档，再基于检索结果回答。",
)
r1 = agent1.invoke({"messages": [("human", "LangChain 是什么？")]})
print("=== 单库检索 ===")
print(r1["messages"][-1].content)


# === 2️⃣ 多库路由 Agent ===
product_vs = Chroma.from_texts(["产品支持 Python 和 Java SDK。", "定价按 API 调用次数计费。"], embeddings)
faq_vs = Chroma.from_texts(["如何获取 API Key？请在控制台申请。", "支持 Python、Java、Node.js。"], embeddings)

# 多个检索工具，Agent 会根据问题内容自动选择
product_tool = create_retriever_tool(
    retriever=product_vs.as_retriever(search_kwargs={"k": 2}),
    name="search_product_docs",
    description="在产品文档中搜索，回答关于产品功能、SDK、定价的问题。",
)
faq_tool = create_retriever_tool(
    retriever=faq_vs.as_retriever(search_kwargs={"k": 2}),
    name="search_faq",
    description="在技术 FAQ 中搜索，回答关于 API Key、编程语言支持等技术问题。",
)

agent2 = create_agent(
    model=llm,
    tools=[product_tool, faq_tool],
    system_prompt="你是一个智能助手。根据问题类型选择合适的工具：产品/定价问题用 search_product_docs，技术/API 问题用 search_faq。",
)
r2 = agent2.invoke({"messages": [("human", "怎么获取 API Key？")]})
print("\n=== 多库路由 ===")
print(r2["messages"][-1].content)


# === 3️⃣ 对话式检索（多轮） ===
agent3 = create_agent(
    model=llm,
    tools=[search_tool],
    system_prompt="你是一个对话式知识库助手。能理解上下文，先检索再回答，支持多轮对话。",
)
messages = [("human", "Agent 能做什么？")]
r3 = agent3.invoke({"messages": messages})
print("\n=== 对话式检索（第1轮）===")
print(r3["messages"][-1].content)

# 第2轮：携带历史继续追问
messages.append(("ai", r3["messages"][-1].content))
messages.append(("human", "那 Tools 呢？"))
r4 = agent3.invoke({"messages": messages})
print("\n=== 对话式检索（第2轮）===")
print(r4["messages"][-1].content)

import os
import logging
from pprint import pprint

# 配置日志观察 ReAct 的推理过程
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
from langchain_community.chat_models import ChatTongyi


api_key = os.getenv("DASHSCOPE_API_KEY")
# LangGraph 当前版本要求使用 model 参数名
model = ChatTongyi(model_name="qwen-turbo", dashscope_api_key=api_key, temperature=0)


# ============ 工具定义 ============
@tool
def calculator(expression: str) -> str:
    """执行数学计算，支持加减乘除和括号。"""
    try:
        # 安全评估：只允许数字和运算符
        allowed = set("0123456789+-*/.() ")
        if not all(c in allowed for c in expression):
            return "错误：表达式包含非法字符"
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"计算错误：{e}"


@tool
def search_weather(city: str) -> str:
    """查询指定城市的天气。"""
    weather_db = {
        "北京": "晴朗，25°C",
        "上海": "多云，22°C",
        "广州": "小雨，28°C",
        "深圳": "阴天，26°C"
    }
    return weather_db.get(city, f"未找到 {city} 的天气信息")


@tool
def get_current_date() -> str:
    """获取今天的日期。"""
    from datetime import datetime
    return datetime.now().strftime("%Y年%m月%d日")


tools = [calculator, search_weather, get_current_date]


# ============ 创建 ReAct Agent ============
# create_react_agent 关键参数说明（以当前 LangGraph 版本为准）：
#   - model: 模型实例（注意：参数名是 model，而不是 llm）
#   - tools: 工具列表
#
# create_react_agent 返回的是一个可调用的 graph，本身就可以直接 invoke

react_agent = create_react_agent(
    model=model,
    tools=tools,
)


# ============ 执行示例 ============
questions = [
    "今天日期是多少？帮我计算 15 * 8 + 12 等于多少",
    "北京和上海的天气分别是怎样的？",
    "如果我有 100 元，买了 3 个每个 15 元的商品，还剩多少钱？"
]

for question in questions:
    print(f"\n{'='*60}")
    print(f"问题：{question}")
    print('='*60)

    # 直接调用由 create_react_agent 返回的 graph
    result = react_agent.invoke({
        "messages": [("human", question)]
    })

    print("\nAgent 原始输出 (原始结构):")
    pprint(result)

    # 尝试从返回结果中解析 ReAct 过程（Thought / Action / Observation）
    if isinstance(result, dict):
        messages = result.get("messages") or result.get("steps")
    else:
        messages = None

    if messages:
        print("\n--- ReAct 过程回放（按时间顺序）---")
        for idx, msg in enumerate(messages, start=1):
            role = getattr(msg, "type", getattr(msg, "role", ""))
            content = getattr(msg, "content", msg)
            print(f"\n[{idx}] role={role}")
            print(content)
    else:
        print("\n(未能自动解析出 messages 列表，请根据原始输出结构手动查看 Thought/Action/Observation)")

import os
import re
import logging

# 配置日志：开启 DEBUG 级别可看到 Agent 内部执行细节
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger("classic_agent_demo")

from langchain_core.tools import tool
from langchain_core.prompts import (
    ChatPromptTemplate,
    MessagesPlaceholder,
    HumanMessagePromptTemplate,
    SystemMessagePromptTemplate,
)
from langchain_community.chat_models import ChatTongyi
from langchain_classic.agents import (
    create_tool_calling_agent,
    create_structured_chat_agent,
    AgentExecutor,  # 注意：从 langchain_classic 导入，不是 langchain
)


api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key:
    raise ValueError("请先在环境变量中设置 DASHSCOPE_API_KEY")

llm = ChatTongyi(model_name="qwen-turbo", dashscope_api_key=api_key, temperature=0)
logger.info(f"✅ LLM 初始化完成：model={llm.model_name}, temperature=0")


# ============ 工具定义 ============
@tool
def calculator(expression: str) -> str:
    """安全计算器，只允许数字与加减乘除括号。"""
    logger.info(f"🔧 [工具调用] calculator: expression={expression}")
    
    if not re.fullmatch(r"[0-9\s\+\-\*\/\(\)\.]+", expression):
        logger.warning(f"⚠️ [工具拒绝] 非法表达式: {expression}")
        return "非法表达式"
    
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        logger.info(f"✅ [工具结果] calculator: result={result}")
        return str(result)
    except Exception as e:
        logger.error(f"❌ [工具错误] calculator: {e}")
        return f"计算错误：{e}"


@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息。"""
    logger.info(f"🔧 [工具调用] get_weather: city={city}")
    
    data = {"北京": "晴天 25°C", "上海": "多云 22°C", "广州": "雨天 28°C"}
    result = data.get(city, f"未找到 {city} 的天气")
    
    logger.info(f"✅ [工具结果] get_weather: {result}")
    return result


tools = [calculator, get_weather]
logger.info(f"📦 已注册 {len(tools)} 个工具: {[t.name for t in tools]}")


# ============ 1️⃣ Tool Calling Agent ============
# 说明：这是现代 LLM（GPT-4/Qwen 等）推荐的范式
# - 模型直接输出 tool_call 对象，而非文本 JSON
# - agent_scratchpad 用 MessagesPlaceholder（消息列表形式）
logger.info("\n" + "="*50)
logger.info("🔧 创建 Tool Calling Agent...")
logger.info("="*50)

tc_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能助手，可以使用工具来回答问题。"),
    ("human", "{input}"),
    # MessagesPlaceholder：Tool Calling 范式用消息列表存储思考过程
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])

tc_agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=tc_prompt)
logger.info(f"✅ Tool Calling Agent 创建成功: type={type(tc_agent).__name__}")

# AgentExecutor：执行 Agent 的运行时环境
# - verbose=True：打印详细执行日志（可以看到每一步的思考过程）
# - handle_parsing_errors=True：自动处理解析错误
tc_executor = AgentExecutor(
    agent=tc_agent, 
    tools=tools, 
    verbose=True,  # 开启详细日志，你会看到 "Entering Agent Executor" 等输出
    handle_parsing_errors=True,
)
logger.info("✅ Tool Calling Executor 创建成功")

logger.info("\n🚀 执行 Tool Calling Agent: 计算 15 * 8 + 12")
print("\n" + "="*50)
print("🔧 Tool Calling Agent 执行中...")
print("="*50)

r1 = tc_executor.invoke({"input": "计算 15 * 8 + 12"})

print("\n" + "="*50)
print(f"✅ Tool Calling Agent 结果: {r1['output']}")
print("="*50)


# ============ 2️⃣ Structured Chat Agent ============
# 说明：这是早期 LLM 兼容的范式，通过文本输出结构化 JSON
# - 模型输出 {"action": "工具名", "action_input": "参数"} 格式的 JSON
# - agent_scratchpad 是字符串变量（不是 MessagesPlaceholder）
# - {tools} 和 {tool_names} 必须保留为模板变量，AgentExecutor 会自动注入
logger.info("\n" + "="*50)
logger.info("💬 创建 Structured Chat Agent...")
logger.info("="*50)

sc_system = """你是一个智能助手，可以使用工具来回答问题。

可用工具：
{tools}

工具名称：{tool_names}

使用 JSON 格式调用工具：
```json
{{
    "action": "工具名称",
    "action_input": "工具参数"
}}
```

最终答案格式：
```json
{{
    "action": "Final Answer",
    "action_input": "你的最终回答"
}}
```"""

sc_prompt = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template(sc_system),
    # 注意：Structured Chat 的 agent_scratchpad 是字符串，用 {agent_scratchpad} 嵌入
    HumanMessagePromptTemplate.from_template("{input}\n\n{agent_scratchpad}"),
])

sc_agent = create_structured_chat_agent(llm=llm, tools=tools, prompt=sc_prompt)
logger.info(f"✅ Structured Chat Agent 创建成功: type={type(sc_agent).__name__}")

sc_executor = AgentExecutor(
    agent=sc_agent, 
    tools=tools, 
    verbose=True,  # 同样开启详细日志
    handle_parsing_errors=True,
)
logger.info("✅ Structured Chat Executor 创建成功")

logger.info("\n🚀 执行 Structured Chat Agent: 查询北京天气")
print("\n" + "="*50)
print("💬 Structured Chat Agent 执行中...")
print("="*50)

r2 = sc_executor.invoke({"input": "北京今天天气怎么样？"})

print("\n" + "="*50)
print(f"✅ Structured Chat Agent 结果: {r2['output']}")
print("="*50)


# ============ 执行总结 ============
logger.info("\n" + "="*50)
logger.info("📊 执行总结")
logger.info("="*50)
logger.info(f"Tool Calling Agent 输出: {r1['output']}")
logger.info(f"Structured Chat Agent 输出: {r2['output']}")
logger.info("\n💡 关键区别：")
logger.info("  - Tool Calling: 模型直接输出 tool_call 对象，更现代、更可靠")
logger.info("  - Structured Chat: 模型输出 JSON 文本，需要解析，兼容性更好")