1
import os
2
from dataclasses import dataclass, fields as dataclass_fields
3
from typing import Annotated, Literal, Optional, TypedDict, Union
4

5
from dotenv import load_dotenv
6
from langchain.agents import create_agent
7
from langchain.agents.structured_output import ToolStrategy
8
from langchain.chat_models import init_chat_model
9
from langchain.messages import HumanMessage
10
from langchain.tools import tool
11
from langchain_core.messages import SystemMessage
12
from pydantic import BaseModel, Field
13

14
load_dotenv(override=True)
15

16
model = init_chat_model(
17
    model="deepseek-v4-flash",
18
    model_provider="openai",
19
    api_key=os.getenv("DEEPSEEK_API_KEY"),
20
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
21
)
22

23
# ---------- 2-1 最简 ToolStrategy ----------
24
class ContactInfo(BaseModel):
25
    """个人联系信息"""
26
    name: str = Field(description="姓名")
27
    email: str = Field(description="电子邮箱")
28
    phone: str = Field(description="手机号")
29

30
agent = create_agent(model=model, response_format=ToolStrategy(ContactInfo))
31
question = HumanMessage("从这段话中抽取结构化信息：小明的邮箱地址为：songhk@atguigu.com，手机号：12345678912")
32
response = agent.invoke({"messages": [question]})
33
if "structured_response" in response:
34
    print(response["structured_response"])          # name='小明' email='songhk@atguigu.com' phone='12345678912'
35
    print(type(response["structured_response"]))    # <class '__main__.ContactInfo'>
36

37
# ---------- 2-2 伪 ToolMessage + 自定义内容 ----------
38
for msg in response["messages"]:
39
    print(type(msg).__name__, "| name =", getattr(msg, "name", None), "| content =", str(msg.content)[:60])
40
# HumanMessage | name = None | content = 从这段话中抽取结构化信息：…
41
# AIMessage    | name = None | content =                     ← 发起工具调用，工具名就是 Schema 类名
42
# ToolMessage  | name = 'ContactInfo' | content = Returning structured response: name='小明' …
43

44
agent2 = create_agent(model=model, response_format=ToolStrategy(ContactInfo, tool_message_content="提取完成！"))
45
response2 = agent2.invoke({"messages": [question]})
46
print(response2["messages"][-1].content)      # 提取完成！
47
print(response2["structured_response"])       # 结构化结果照样正确
48

49
# ---------- 2-3 客户分析报告（工具 + 结构化） ----------
50
@tool(parse_docstring=True)
51
def search_customer_database(query: str) -> str:
52
    """在客户数据库中搜索信息
53

54
    Args:
55
        query (str): 客户查询字符串，例如 "张三" 或 "李四"
56

57
    Returns:
58
        str: 客户记录字符串，包含客户姓名、等级、最近购买日期和累计消费
59
    """
60
    if "张三" in query.lower():
61
        return "客户记录：张三，VIP客户，最近购买日期：2026-01-15，累计消费：$15,000"
62
    elif "李四" in query.lower():
63
        return "客户记录：李四，普通客户，最近购买日期：2025-12-20，累计消费：$3,200"
64
    return f"关于客户{query}，无记录"
65

66
@tool(parse_docstring=True)
67
def send_email(customer: str) -> str:
68
    """发送感谢邮件
69

70
    Args:
71
        customer (str): 客户名称，例如 "张三" 或 "李四"
72

73
    Returns:
74
        str: 确认消息，包含已发送的客户名称
75
    """
76
    return f"已向 {customer} 发送感谢邮件"
77

78
class CustomerAnalysis(BaseModel):
79
    """客户分析报告"""
80
    customer_name: str = Field(None, description="客户姓名")
81
    customer_tier: Literal["潜在客户", "普通客户", "VIP客户", "流失风险"] = Field(
82
        "潜在客户", description="客户等级,只能是潜在客户、普通客户、VIP客户或流失风险"
83
    )
84
    recent_activity: str = Field(None, description="最近活动")
85
    spending_level: Literal["低", "中", "高"] = Field(None, description="消费水平")
86
    send_email: bool = Field(False, description="是否已发送感谢邮件")
87

88
agent3 = create_agent(
89
    model=model,
90
    system_prompt=SystemMessage(content=""
91
        "请分析指定客户的情况："
92
        "1. 先搜索客户数据库了解最新情况 "
93
        "2. 如果是VIP客户，则发送感谢邮件 "
94
        "3. 基于搜索结果生成结构化分析报告 "
95
        "4. 如果用户提问与客户记录无关或找不到客户信息，则返回空对象，不发送感谢邮件"),
96
    tools=[search_customer_database, send_email],
97
    response_format=ToolStrategy(CustomerAnalysis),
98
)
99
result = agent3.invoke({"messages": [{"role": "user", "content": "请分析客户张三"}]})
100
print("消息类型链:", " → ".join(type(m).__name__ for m in result["messages"]))
101
# HumanMessage → AIMessage → ToolMessage → AIMessage → ToolMessage → AIMessage → ToolMessage
102
print([m.name for m in result["messages"] if type(m).__name__ == "ToolMessage"][-1])   # CustomerAnalysis
103
if "structured_response" in result:
104
    print(result["structured_response"])
105
    # customer_name='张三' customer_tier='VIP客户' recent_activity='最近购买日期：2026-01-15' spending_level='高' send_email=True
106
    print(result["structured_response"].customer_tier)                                  # 对象属性取值：VIP客户
107

108
# ---------- 2-4 同一份 Schema 的三种写法 ----------
109
# ① 手写 JSON Schema 字典
110
contact_schema = {
111
    "title": "ContactInfo",              # 结构名 = 虚拟工具名
112
    "type": "object",
113
    "description": "个人联系信息",
114
    "properties": {
115
        "name": {"type": "string", "default": "", "description": "姓名"},
116
        "email": {"type": "string", "default": "", "description": "电子邮箱"},
117
        "phone": {"type": "string", "default": "", "description": "手机号（可省略）"},
118
    },
119
    "required": ["name", "email"],       # 只要求前两个
120
}
121
agent4 = create_agent(model=model, response_format=ToolStrategy(contact_schema))
122
r4 = agent4.invoke({"messages": [question]})
123
print(type(r4["structured_response"]).__name__)        # dict ← 普通字典，没有属性访问
124

125
# ② TypedDict（Annotated[类型, 描述]；可选字段用 Optional）
126
class ContactInfoTD(TypedDict):
127
    """个人联系信息"""
128
    name: Annotated[str, "姓名"]
129
    email: Annotated[str, "电子邮箱"]
130
    phone: Annotated[Optional[str], "手机号"]
131

132
agent5 = create_agent(model=model, response_format=ToolStrategy(ContactInfoTD))
133
r5 = agent5.invoke({"messages": [question]})
134
print(type(r5["structured_response"]).__name__)        # dict ← 同样不做运行时校验
135

136
# ③ @dataclass（注意 Field 来自 pydantic）
137
@dataclass
138
class ContactInfoDC:
139
    """用户的联系方式"""
140
    name: str = Field(description="用户姓名")
141
    email: str = Field(description="用户邮箱地址")
142
    phone: str = Field(description="用户手机号")
143

144
agent6 = create_agent(model=model, response_format=ToolStrategy(ContactInfoDC))
145
r6 = agent6.invoke({"messages": [question]})
146
print(type(r6["structured_response"]))                 # <class '__main__.ContactInfoDC'> ← 是对象
147
print([f.default for f in dataclass_fields(ContactInfoDC)][0])
148
# FieldInfo(annotation=NoneType, required=True, description='用户姓名') ← 坑：default 被 Field 占了
149

150
# ---------- 3-1 Union 多类型 + handle_errors ----------
151
class EventDetails(BaseModel):
152
    """活动详情"""
153
    event_name: str = Field(description="活动名称")
154
    date: str = Field(description="活动日期")
155

156
agent7 = create_agent(
157
    model=model,
158
    response_format=ToolStrategy(Union[ContactInfo, EventDetails], tool_message_content="提取完成！"),
159
)
160
for text in [
161
    "从这段话中抽取结构化信息：小明的邮箱地址为：songhk@atguigu.com，手机号：12345678912",
162
    "从这段话中抽取结构化信息：2026年高考报名人数突破1200万",
163
]:
164
    r = agent7.invoke({"messages": [HumanMessage(text)]})
165
    if "structured_response" in r:
166
        print(type(r["structured_response"]).__name__, r["structured_response"])
167
# ContactInfo name='小明' email='songhk@atguigu.com' phone='12345678912'
168
# EventDetails event_name='2026年高考报名人数突破1200万' date='…'
169

170
# 把 handle_errors 换成 False / "请检查输入数据" 再跑一次，对比失败时的表现：
171
agent8 = create_agent(
172
    model=model,
173
    response_format=ToolStrategy(
174
        Union[ContactInfo, EventDetails],
175
        tool_message_content="提取完成！",
176
        handle_errors=True,        # 改 False：直接抛 MultipleStructuredOutputsError；改字符串：当作错误提示回传
177
    ),
178
)
179
r8 = agent8.invoke({"messages": [HumanMessage("从这段话中抽取结构化信息：小明的邮箱地址为：songhk@atguigu.com，手机号：12345678912")]})
180
for m in r8["messages"]:
181
    if type(m).__name__ == "ToolMessage":
182
        print(m.name, "->", str(m.content)[:110])
183
# 触发"多结构化输出"错误时（模型一次返回两个结构化结果）：
184
#   ContactInfo  -> Error: Model incorrectly returned multiple structured responses (ContactInfo, EventDetails) …
185
#   EventDetails -> Error: Model incorrectly returned multiple structured responses (ContactInfo, EventDetails) …
186
#   ContactInfo  -> 提取完成！        ← 重试成功，用的还是自定义的伪消息内容