1
from langchain_core.messages import SystemMessage
2
from pydantic import BaseModel, Field
3
from typing import Literal
4
from langchain.agents import create_agent
5
from langchain.agents.structured_output import ToolStrategy
6
from langchain.tools import tool
7

8
# ---------- 1. 两个"真"工具 ----------
9
@tool(parse_docstring=True)
10
def search_customer_database(query: str) -> str:
11
    """在客户数据库中搜索信息
12

13
    Args:
14
        query (str): 客户查询字符串，例如 "张三" 或 "李四"
15

16
    Returns:
17
        str: 客户记录字符串，包含客户姓名、等级、最近购买日期和累计消费
18
    """
19
    # 模拟数据库查询结果
20
    if "张三" in query.lower():
21
        return "客户记录：张三，VIP客户，最近购买日期：2026-01-15，累计消费：$15,000"
22
    elif "李四" in query.lower():
23
        return "客户记录：李四，普通客户，最近购买日期：2025-12-20，累计消费：$3,200"
24
    else:
25
        return f"关于客户{query}，无记录"
26

27

28
@tool(parse_docstring=True)
29
def send_email(customer: str) -> str:
30
    """发送感谢邮件
31

32
    Args:
33
        customer (str): 客户名称，例如 "张三" 或 "李四"
34

35
    Returns:
36
        str: 确认消息，包含已发送的客户名称
37
    """
38
    return f"已向 {customer} 发送感谢邮件"
39

40

41
# ---------- 2. 定义 Pydantic Schema ----------
42
class CustomerAnalysis(BaseModel):
43
    """客户分析报告"""
44
    customer_name: str = Field(None, description="客户姓名")
45
    customer_tier: Literal["潜在客户", "普通客户", "VIP客户", "流失风险"] = Field(
46
        "潜在客户", description="客户等级,只能是潜在客户、普通客户、VIP客户或流失风险"
47
    )
48
    recent_activity: str = Field(None, description="最近活动")
49
    spending_level: Literal["低", "中", "高"] = Field(None, description="消费水平")
50
    send_email: bool = Field(False, description="是否已发送感谢邮件")
51

52

53
# ---------- 3. 创建智能体 ----------
54
agent = create_agent(
55
    model=model,
56
    system_prompt=SystemMessage(content=""
57
        "请分析指定客户的情况："
58
        "1. 先搜索客户数据库了解最新情况 "
59
        "2. 如果是VIP客户，则发送感谢邮件 "
60
        "3. 基于搜索结果生成结构化分析报告 "
61
        "4. 如果用户提问与客户记录无关或找不到客户信息，则返回空对象，不发送感谢邮件"
62
    ),
63
    tools=[search_customer_database, send_email],     # 工具 + 结构化输出同时用
64
    response_format=ToolStrategy(CustomerAnalysis),
65
)
66

67
# ---------- 4. 执行分析 ----------
68
result = agent.invoke({
69
    "messages": [{"role": "user", "content": "请分析客户张三"}]
70
    # "messages": [{"role": "user", "content": "请分析客户李四"}]
71
    # "messages": [{"role": "user", "content": "请分析客户王五"}]
72
    # "messages": [{"role": "user", "content": "今天天气如何"}]
73
})
74

75
# ---------- 5. 处理结果 ----------
76
if "structured_response" in result:
77
    analysis = result["structured_response"]
78
    print(analysis)

1
# 把所有字段定义成一个标准 JSON Schema 字典（替代上面的 Pydantic 模型）
2
customer_analysis_schema = {
3
    "title": "CustomerAnalysis",           # 结构名（会当"虚拟工具"的名字用）
4
    "type": "object",                      # 表示"这是一个对象"
5
    "description": "客户分析报告",
6
    "properties": {                        # 每个字段的定义
7
        "customer_name": {
8
            "type": "string",
9
            "default": "",
10
            "description": "客户姓名",
11
        },
12
        "customer_tier": {
13
            "type": "string",
14
            "enum": ["潜在客户", "普通客户", "VIP客户", "流失风险"],   # ← 枚举写在这里
15
            "default": "潜在客户",
16
            "description": "客户等级",
17
        },
18
        "recent_activity": {
19
            "type": "string",
20
            "default": "",
21
            "description": "最近活动",
22
        },
23
        "spending_level": {
24
            "type": "string",
25
            "enum": ["低", "中", "高"],
26
            "default": "低",
27
            "description": "消费水平",
28
        },
29
        "send_email": {
30
            "type": "boolean",             # ← 布尔类型（注意 JSON 里写 false）
31
            "default": False,
32
            "description": "是否已发送感谢邮件",
33
        },
34
    },
35
    # 所有字段都是必须输出的
36
    "required": ["customer_name", "customer_tier", "recent_activity", "spending_level"],
37
}
38

39
agent = create_agent(
40
    model=model,
41
    system_prompt=SystemMessage(content="...同写法1的四条指令..."),
42
    tools=[search_customer_database, send_email],
43
    response_format=ToolStrategy(customer_analysis_schema),   # ← 直接把字典交给 ToolStrategy
44
)

1
from dataclasses import dataclass
2
from langchain.agents import create_agent
3
from langchain.agents.structured_output import ToolStrategy
4
from langchain.messages import HumanMessage
5
from pydantic import Field              # ← 课程举例1 的片段里没写这一行，得自己补
6

7
@dataclass
8
class ContactInfo:
9
    """用户的联系方式"""
10
    name: str = Field(description="用户姓名")
11
    email: str = Field(description="用户邮箱地址")
12
    phone: str = Field(description="用户手机号")
13

14
agent = create_agent(
15
    model=model,
16
    response_format=ToolStrategy(ContactInfo),
17
)
18

19
response = agent.invoke({
20
    "messages": [
21
        HumanMessage("从这段话中抽取结构化信息：小明的邮箱地址为：songhk@atguigu.com，手机号：12345678912")
22
    ]
23
})
24

25
for msg in response["messages"]:
26
    msg.pretty_print()