提示：文章写完后，目录可以自动生成，如何生成可参考右边的帮助文档

---

Agent的高级用法1：设置Agent名称

Agent的高级用法2：系统提示词

基础提示词设置

举例1：

from langchain_tavily import TavilySearch
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os
load_dotenv(override=True)


# 1.导入模型
model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL")
)

# 2.导入工具
web_search = TavilySearch(max_results=2)

# 3.创建Agent
agent = create_agent(
    model=model,
    tools=[web_search],
    system_prompt="你是一名多才多艺的智能助手，可以调用工具帮助用户解决问题。"
)

# 4.运行Agent获得结果
result = agent.invoke(
    {"messages": [
        {"role": "user", "content": "请帮我查询2026年足球世界杯是哪个国家举办的？"}
    ]}
)

rprint(result)

举例2：

from langchain.agents import create_agent
from langchain_core.messages import SystemMessage
from langchain_core.tools import tool
from rich import print as rprint

# 工具：实现两数相加
@tool
def add_numbers(a: int, b: int) -> str:
    """计算并返回两个数的和。"""
    return f"和为：{a + b}"


# 创建客服助手Agent
agent = create_agent(
    model=model,
    tools=[add_numbers],  # 工具列表
    # system_prompt="你是一个数学助手，解决日常的算术问题"
    system_prompt=SystemMessage(content="你是一个数学助手，解决日常的算术问题")
)

response = agent.invoke(
    {"messages": [
        {"role": "user", "content": "10加上20再加上30是多少？"}
    ]},
)

rprint(response)
# print(response["messages"][-1].content)

举例3：

from langchain.agents import create_agent
from langchain.tools import tool
from langchain.messages import SystemMessage, HumanMessage

flag = 0

@tool
def get_weather(city: str):
    """
    天气查询工具

    Args:
        city: 城市名称
    """
    global flag
    flag += 1

    if flag < 3:
        # raise Exception("暂时无法访问")
        return "TEMP_UNAVAILABLE: 天气服务暂时不可用，请稍后重试"

    return f"{city}今天天气挺好"


messages = [
    HumanMessage("你好，杭州今天的天气如何？")
]

agent = create_agent(
    model=model,
    tools=[get_weather],
    system_prompt=SystemMessage(
        "你是一个天气助手。"
        "当工具返回以 'TEMP_UNAVAILABLE:' 开头的结果时，"
        "说明是临时故障，不要立即放弃；"
        "你应再次调用同一个工具，最多重试 3 次。"
        "如果 3 次后仍失败，再向用户说明服务暂时不可用。"
    )
)
response = agent.invoke({"messages": messages})
# print(response)
for msg in response["messages"]:
    msg.pretty_print()

动态提示词设置

• 对于需要根据运行时上下文或Agent状态调整提示词的高级场景，可以使用中间件（Middleware）实现动态提示词。例如，通过 @dynamic_prompt装饰器创建中间件，根据用户角色生成不同的系统提示。

案例：构建客户问题处理的Agent，对于普通用户和VIP用户采用不同的提示词。

from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt, ModelRequest
from langchain_core.tools import BaseTool, tool
from typing import TypedDict
import json

from init_llm import deepseek_llm


# 工具1：模拟查询订单信息
@tool
def query_order_info(order_id: str) -> str:
    """根据订单ID查询订单的详细信息，包括状态、商品列表和创建时间。"""
    # 模拟数据库查询结果
    order_database = {
        "ORD123456": {"status": "已发货", "items": ["手机X1"], "create_time": "2025-01-15"},
        "ORD654321": {"status": "待付款", "items": ["耳机Y1"], "create_time": "2025-01-18"}
    }
    order_data = order_database.get(order_id)
    if order_data:
        return json.dumps(order_data, ensure_ascii=False)
    else:
        return f"错误：未找到订单 {order_id}。"

# 工具2：模拟检索常见问题解答
@tool
def search_faq(keyword: str) -> str:
    """根据关键词从知识库中检索相关的政策条款或解决方案。"""
    # 模拟FAQ知识库
    faq_knowledge_base = {
        "退货": "支持7天无理由退货，商品需完好且包装齐全。",
        "保修": "电子产品享受1年免费保修服务。",
        "发货": "下单后48小时内发货，偏远地区可能延迟。"
    }
    # 简单关键词匹配
    for topic, answer in faq_knowledge_base.items():
        if topic in keyword:
            return answer
    return f"未找到与'{keyword}'直接相关的政策，请尝试其他关键词或联系人工客服。"

# 定义运行时上下文的数据结构
class AgentContext(TypedDict):
    query_type: str  # 用于动态判断问题类型，例如 'normal'（普通用户） 或 'vip'（vip用户）

# 动态提示词中间件：根据问题类型调整Agent的“角色”和回答策略
@dynamic_prompt
def dynamic_support_prompt(request: ModelRequest) -> str:
    """
    根据 query_type 生成不同的系统提示词。
    """
    # print("request:", request)
    query_type = request.runtime.context.get("query_type", "normal")
    base_instruction = "你是一名专业的电商客服助手。请根据工具查询结果，准确、清晰地回答用户问题。"

    if query_type == "vip":
        # 针对复杂或需要升级处理的问题
        return f"""{base_instruction}
                当前角色：高级支持专员
                工作要求：
                1.深度分析：仔细分析用户描述，识别潜在的根本问题。
                2.精准分类：将问题明确归类（如“物流问题”、“产品质量”、“售后申请”）。
                3.方案规划：若工具能解决，提供具体步骤；若需人工，明确告知后续流程。
                请使用更专业、严谨的语言。
                """
    else:
        # 针对常规客服问题
        return f"""{base_instruction}
                当前角色：一线客服助手
                工作要求：
                1.快速响应：优先使用工具获取订单/政策信息。
                2.直接解答：对于明确问题（如退货、物流），直接给出基于知识的答案。
                3.简洁友好：回复要简单明了，避免复杂术语。
                保持友好和高效的沟通风格。
                """

if __name__ == "__main__":
    """创建客服助手Agent"""
    # 创建带有动态提示词中间件的Agent
    agent = create_agent(
        model=deepseek_llm,
        tools=[query_order_info, search_faq], # 工具列表
        middleware=[dynamic_support_prompt],  # 挂载动态提示词中间件
        context_schema=AgentContext  # 关联上下文schema
    )

    """演示动态提示词效果"""
    # user_query = "我的订单ORD123456还没收到，包装破损了怎么办？"
    user_query = "我的订单ORD654321已签收，但是物品坏了怎么办？"
    print(f"用户问题：{user_query}\n")

    print("场景一：普通客服模式 (query_type: 'normal')")
    result_normal = agent.invoke(
        {"messages": [{"role": "user", "content": user_query}]},
        context={"query_type": "normal"}  # 标准模式
    )

    # print("result_normal:", result_normal)
    # 提取Agent的最后一条回复
    final_response_normal = result_normal["messages"][-1].content
    print(final_response_normal)

    print("\n" + "="*50)

    print("场景二：VIP模式 (query_type: 'vip')")
    result_vip = agent.invoke(
        {"messages": [{"role": "user", "content": user_query}]},
        context={"query_type": "vip"}
    )
    # print("result_vip:", result_vip)
    # 提取Agent的最后一条回复
    final_response_vip = result_vip["messages"][-1].content
    print(final_response_vip)

以上代码运行结果如下：

用户问题：我的订单ORD654321已签收，但是物品坏了怎么办？

场景一：普通客服模式 (query_type: 'normal')
根据查询到的信息，我为您提供以下建议：

**关于您的情况：**
1. 订单状态显示"待付款"，但您已签收商品且发现损坏
2. 这可能是系统状态更新延迟，或者存在其他特殊情况

**建议解决方案：**

1. **立即拍照取证**：请拍摄清晰的商品损坏照片，包括外包装、商品本身和损坏部位

2. **联系人工客服**：由于您的订单状态异常（显示待付款但已签收），建议您：
   - 通过在线客服或客服热线联系人工客服
   - 提供订单号ORD654321和损坏照片
   - 说明具体情况：已签收但商品损坏

3. **售后处理**：根据我们的退货政策，商品需完好才能享受7天无理由退货。但**商品损坏属于质量问题**，通常会有专门的售后处理流程。

**温馨提示：**
- 请尽快联系客服，避免超过售后处理时限
- 保留好所有相关证据（照片、物流签收记录等）
- 不要自行处理损坏商品，等待客服指导

由于订单状态异常，建议您优先联系人工客服处理，他们会为您核实具体情况并提供专业的解决方案。

==================================================
场景二：VIP模式 (query_type: 'vip')
基于目前的信息，我为您提供以下专业分析和建议：

## 问题分类：产品质量问题

### 当前情况分析：
1. **订单状态异常**：系统显示"待付款"状态与您描述的"已签收"不符
2. **商品损坏**：您明确表示收到的物品已损坏

### 建议处理方案：

**第一步：核实订单信息**
请确认订单号ORD654321是否正确。如果可能，请提供：
- 准确的订单号
- 签收时间
- 损坏物品的具体情况描述和照片

**第二步：标准售后流程**
根据我们的退货政策：
- 支持7天无理由退货，商品需完好且包装齐全
- 对于质量问题，我们有专门的售后处理流程

**第三步：立即行动建议**
由于订单状态显示异常，我建议您：
1. **联系人工客服**：通过官方客服渠道说明情况
2. **提供证据**：准备商品损坏的清晰照片或视频
3. **保留包装**：保持商品原包装完整

### 后续处理路径：
1. **如果订单确实已签收**：我们将为您启动质量问题售后流程
2. **如果订单状态有误**：需要技术部门核实系统状态
3. **紧急处理**：建议立即通过官方客服渠道联系，获取优先处理

**重要提醒**：请尽快联系客服，以便在签收后的有效时间内启动售后流程。质量问题通常有特殊的处理标准和时效要求。

您是否需要我协助您查找官方客服联系方式，或者有其他相关信息需要补充？

Agent的高级用法3：结构化输出

7.1 模型 vs Agent的结构化输出对比

7.2 结构化输出的4种策略

① ProviderStrategy

② ToolStrategy

③ type / AutoStrategy

④ None

7.3 ToolStrategy使用详解

7.3.1 结构化输出：schema参数

7.3.2 自定义工具消息：tool_message_content参数

7.3.3 错误处理：handle_errors参数

Agent的高级用法4：流式输出及模式

from langchain.chat_models import init_chat_model
from dotenv import load_dotenv
import os

# 从.env文件中加载环境变量
load_dotenv(override=True)

CLOSEAI_API_KEY = os.getenv("CLOSEAI_API_KEY")
CLOSEAI_BASE_URL = os.getenv("CLOSEAI_BASE_URL")

model = init_chat_model(
    model="gpt-5.4-mini",
    model_provider="openai",
    api_key=CLOSEAI_API_KEY,
    base_url=CLOSEAI_BASE_URL
)

rom langchain.agents import create_agent
from langchain.tools import tool
from typing import Dict, Any
from rich import print as rprint

@tool
def query_customer_data(customer_id: str) -> Dict[str, Any]:
    """
    查询客户基本信息

    Args:
        customer_id: 客户ID，用于唯一标识客户

    Returns:
        包含客户基本信息的字典，如姓名、等级、加入日期等
    """
    # 模拟数据库查询
    return {"name": "张三","level": "VIP","join_date": "2023-01-15"}


@tool
def check_order_history(customer_id: str) -> Dict[str, Any]:
    """
    查询客户订单历史

    Args:
        customer_id: 客户ID，用于唯一标识客户

    Returns:
        包含客户订单历史的字典，如总订单数、总花费等
    """
    return {"total_orders": 15,"total_spent": 25800.00}


@tool
def get_current_promotions() -> Dict[str, Any]:
    """
    获取当前可用促销活动

    Returns:
        包含当前可用促销活动的字典，如活动名称、有效日期等
    """
    return {
        "promotions": ["老用户优惠", "会员专属折扣"],
        "valid_until": "2027-01-31"
    }


# 创建客户服务Agent
customer_service_agent = create_agent(
    model=model,
    tools=[query_customer_data, check_order_history, get_current_promotions]
)

for chunk in customer_service_agent.stream(
    {
        "messages" : [
            {"role":"user","content":"查询客户id为cust1234的完整的信息、历史订单和可用优惠"}
        ]
    },
    stream_mode="values"
):
    rprint(chunk)
    print("-" * 50)