🎯 Agent 基础

步骤 1 - 推理：流行度是时效性的，需要搜索最新数据
步骤 2 - 行动：调用 search_products("wireless headphones")
步骤 3 - 观察：获得前 5 个结果，第一名是 WH-1000XM5

步骤 4 - 推理：需要确认该产品的库存情况
步骤 5 - 行动：调用 check_inventory("WH-1000XM5")
步骤 6 - 观察：库存 10 件

步骤 7 - 最终答案：
"WH-1000XM5 是目前最受欢迎的无线耳机，库存 10 件。"

"""
create_agent() 基础示例
功能：创建一个带搜索和天气工具的简单 Agent
"""
from langchain.agents import create_agent
from langchain.tools import tool

# 定义工具
@tool
def search(query: str) -> str:
    """搜索信息"""
    return f"搜索结果：{query}"

@tool
def get_weather(location: str) -> str:
    """获取天气信息"""
    return f"{location} 的天气：晴朗，22°C"

# 创建 Agent
agent = create_agent(
    model="gpt-4",              # 模型标识符（字符串）
    tools=[search, get_weather], # 工具列表
    system_prompt="你是一个有用的助手。请简洁准确地回答问题。"
)

# 调用 Agent
result = agent.invoke({
    "messages": [{"role": "user", "content": "旧金山今天天气怎么样？"}]
})

print(result["messages"][-1].content)

"""
create_agent() 高级配置示例
功能：使用 ChatModel 实例进行精细控制
"""
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.tools import tool

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)  # 仅用于演示，生产环境需安全处理
        return f"计算结果：{result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

# 使用 ChatModel 实例进行精细配置
model = ChatOpenAI(
    model="gpt-4",
    temperature=0.1,     # 降低随机性，提高准确性
    max_tokens=1000,     # 限制输出长度
    timeout=30,          # 设置超时时间（秒）
    max_retries=2        # 失败重试次数
)

# 创建 Agent
agent = create_agent(
    model=model,
    tools=[calculator],
    system_prompt="""你是一个精确的数学助手。

职责：
1. 使用 calculator 工具计算数学表达式
2. 解释计算步骤
3. 确保结果准确无误

注意：
- 只处理数学相关的问题
- 对于非数学问题，礼貌地拒绝"""
)

# 测试
result = agent.invoke({
    "messages": [{"role": "user", "content": "计算 (123 + 456) * 2 等于多少？"}]
})

print(result["messages"][-1].content)

"""
优秀的 system_prompt 模板
"""
SYSTEM_PROMPT = """# 角色定义
你是 [角色名称]，专门负责 [主要职责]。

# 核心职责
1. [职责1]
2. [职责2]
3. [职责3]

# 工作流程
当用户提出请求时：
- 步骤1：[分析需求]
- 步骤2：[调用工具]
- 步骤3：[整合结果]
- 步骤4：[生成回复]

# 工具使用
你有以下工具可用：
- tool1：[用途和使用时机]
- tool2：[用途和使用时机]

# 回复风格
- 简洁：直接回答问题
- 准确：基于工具返回的实际数据
- 友好：保持礼貌和专业

# 限制和注意事项
- 不要：[禁止行为]
- 必须：[强制要求]
- 如果遇到 [情况]，则 [处理方式]
"""

# 使用示例
from langchain.agents import create_agent

agent = create_agent(
    model="gpt-4",
    tools=[...],
    system_prompt=SYSTEM_PROMPT
)

"""
动态 system_prompt 示例
功能：根据用户角色生成不同的提示词
"""
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from langchain.agents import create_agent

@dynamic_prompt
def user_role_prompt(request: ModelRequest) -> str:
    """根据用户角色生成系统提示词"""
    # 从运行时上下文获取用户角色
    user_role = request.runtime.context.get("user_role", "user")

    base = "你是一个有用的技术助手。"

    if user_role == "expert":
        return f"""{base}

用户是技术专家，你应该：
- 提供深入的技术细节
- 使用专业术语
- 假设用户了解基础概念
- 直接给出高级解决方案"""

    elif user_role == "beginner":
        return f"""{base}

用户是初学者，你应该：
- 用简单的语言解释概念
- 避免复杂的术语
- 提供步骤化的指导
- 多给示例和类比"""

    else:
        return f"""{base}

提供平衡的技术解释，适合一般开发者。"""

# 创建 Agent
agent = create_agent(
    model="gpt-4",
    tools=[...],
    middleware=[user_role_prompt]
)

# 调用时提供用户角色
result = agent.invoke(
    {"messages": [{"role": "user", "content": "什么是 Agent？"}]},
    context={"user_role": "beginner"}  # 指定用户角色
)

"""
Agent 流式输出示例
功能：实时显示 Agent 的推理和工具调用过程
"""
from langchain.agents import create_agent
from langchain.tools import tool

@tool
def search(query: str) -> str:
    """搜索 AI 相关新闻"""
    return f"找到关于 '{query}' 的最新新闻..."

# 创建 Agent
agent = create_agent(
    model="gpt-4",
    tools=[search],
    system_prompt="你是新闻助手，帮助用户搜索最新资讯。"
)

# 流式调用 Agent
print("=== Agent 执行过程 ===\n")

for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "搜索最新的 AI 新闻"}]},
    stream_mode="values"  # 流式模式
):
    # 获取最新消息
    latest_message = chunk["messages"][-1]

    # 如果是文本内容
    if latest_message.content:
        print(f"💬 Agent: {latest_message.content}")

    # 如果是工具调用
    elif hasattr(latest_message, 'tool_calls') and latest_message.tool_calls:
        tool_names = [tc['name'] for tc in latest_message.tool_calls]
        print(f"🔧 调用工具: {', '.join(tool_names)}")

print("\n=== 执行完成 ===")

=== Agent 执行过程 ===

💬 Agent: 让我帮你搜索最新的 AI 新闻
🔧 调用工具: search
💬 Agent: 找到关于 'AI' 的最新新闻...
💬 Agent: 根据搜索结果，这里是最新的 AI 新闻摘要...

=== 执行完成 ===

"""
工具错误处理示例
功能：优雅地处理工具执行错误
"""
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
from langchain.agents import create_agent
from langchain.tools import tool

@tool
def divide(a: float, b: float) -> float:
    """除法运算"""
    return a / b  # 可能抛出 ZeroDivisionError

# 定义错误处理中间件
@wrap_tool_call
def handle_tool_errors(request, handler):
    """捕获并优雅地处理工具执行错误"""
    try:
        # 执行工具
        return handler(request)
    except ZeroDivisionError:
        # 除零错误
        return ToolMessage(
            content="错误：除数不能为零。请检查输入参数。",
            tool_call_id=request.tool_call["id"]
        )
    except Exception as e:
        # 其他错误
        return ToolMessage(
            content=f"工具执行失败：{str(e)}。请检查输入并重试。",
            tool_call_id=request.tool_call["id"]
        )

# 创建带错误处理的 Agent
agent = create_agent(
    model="gpt-4",
    tools=[divide],
    middleware=[handle_tool_errors],  # 应用错误处理中间件
    system_prompt="你是数学助手。如果工具执行失败，请向用户解释问题并建议解决方案。"
)

# 测试：正常情况
print("=== 正常情况 ===")
result1 = agent.invoke({
    "messages": [{"role": "user", "content": "10 除以 2 等于多少？"}]
})
print(result1["messages"][-1].content)

# 测试：除零错误
print("\n=== 除零错误 ===")
result2 = agent.invoke({
    "messages": [{"role": "user", "content": "10 除以 0 等于多少？"}]
})
print(result2["messages"][-1].content)

import logging

# 启用 LangChain 日志
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("langchain")
logger.setLevel(logging.DEBUG)

import os

# 配置 LangSmith
os.environ["LANGSMITH_API_KEY"] = "your-key"
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "agent-debugging"

# 之后的 Agent 调用会自动追踪

# 使用流式输出查看每一步
for chunk in agent.stream(input_data, stream_mode="values"):
    print(f"步骤: {chunk}")
    print(f"消息数: {len(chunk['messages'])}")
    print(f"最新消息: {chunk['messages'][-1]}")
    print("-" * 50)

from langchain.agents.middleware import before_agent

@before_agent
def limit_iterations(state, runtime):
    """限制 Agent 迭代次数"""
    max_iterations = 10
    current = len([m for m in state["messages"] if m.type == "ai"])

    if current >= max_iterations:
        raise ValueError(f"达到最大迭代次数 {max_iterations}")

    return state