例子tesla_agent完全解析
更新时间: 2026-03-26 14:38:32
"""
Tesla Agent - LangChain 1.2.13 + LangGraph 架构
使用 create_agent 函数创建特斯拉产品助手
"""
import textwrap
import time
# ========== Tool 相关 ==========
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field
# ========== LLM 相关 ==========
from langchain_community.chat_models.tongyi import ChatTongyi
# ========== Agent 创建 ==========
from langchain.agents import create_agent
# ========== Prompt Templates ==========
CONTEXT_QA_TMPL = """根据以下提供的信息，回答用户的问题
信息：{context}
问题：{query}"""
def output_response(messages: list) -> None:
"""动态显示响应结果，打字机效果"""
if not messages:
return
# 获取最后一条消息的内容
last_message = messages[-1]
content = last_message.content if hasattr(last_message, 'content') else str(last_message)
if not content:
return
print("\n" + "="*60)
print("回答：")
print("="*60)
for line in textwrap.wrap(content, width=60):
for word in line.split():
for char in word:
print(char, end="", flush=True)
time.sleep(0.05)
print(" ", end="", flush=True)
print()
print("-"*60 + "\n")
class ProductInput(BaseModel):
"""产品查询输入参数"""
product_name: str = Field(description="产品名称，例如：Model 3")
class CompanyInfoInput(BaseModel):
"""公司信息查询输入参数"""
query: str = Field(description="具体的问题")
class TeslaDataSource:
"""特斯拉数据源，提供产品查询和公司信息查询功能"""
def __init__(self, llm):
self.llm = llm
def find_product_description(self, product_name: str) -> str:
"""通过产品名称查询产品描述"""
product_info = {
"Model 3": "具有简洁、动感的外观设计，流线型车身和现代化前脸。定价23.19-33.19万",
"Model Y": "在外观上与Model 3相似，但采用了更高的车身和更大的后备箱空间。定价26.39-36.39万",
"Model X": "拥有独特的翅子门设计和更加大胆的外观风格。定价89.89-105.89万",
"Model S": "特斯拉的旗舰轿车，拥有卓越的性能和超长续航里程。定价69.89-89.89万",
}
return product_info.get(
product_name,
f"没有找到产品'{product_name}'，可用产品：{', '.join(product_info.keys())}"
)
def find_company_info(self, query: str) -> str:
"""查询公司相关信息"""
context = """
特斯拉最知名的产品是电动汽车，其中包括Model S、Model 3、Model X和Model Y等多款车型。
特斯拉以其技术创新、高性能和领先的自动驾驶技术而闻名。
公司不断推动自动驾驶技术的研发，并在车辆中引入各种驾驶辅助功能，
如自动紧急制动、自适应巡航控制和车道保持辅助等。
特斯拉还在能源存储领域有所布局，推出了Powerwall家用电池和太阳能屋顶等产品。
公司由埃隆·马斯克创立，致力于加速世界向可持续能源的转变。
"""
prompt = CONTEXT_QA_TMPL.format(query=query, context=context)
return self.llm.invoke(prompt).content
def create_tesla_agent(api_key: str):
"""创建特斯拉Agent - 使用 LangGraph 架构"""
# ========== 初始化 Chat 模型（必须是 BaseChatModel 类型）==========
model = ChatTongyi(
model="qwen-turbo",
dashscope_api_key=api_key
)
# 创建数据源
tesla_data_source = TeslaDataSource(model)
# ========== 定义工具 ==========
tools = [
StructuredTool.from_function(
func=tesla_data_source.find_product_description,
name="查询产品名称",
description="通过产品名称查找产品详细信息，包括价格、外观等。输入参数是产品名称，例如：'Model 3'",
args_schema=ProductInput
),
StructuredTool.from_function(
func=tesla_data_source.find_company_info,
name="公司相关信息",
description="查询特斯拉公司的相关信息，包括技术创新、产品线、自动驾驶技术等。输入参数是具体的问题",
args_schema=CompanyInfoInput
),
]
# ========== System Prompt ==========
system_prompt = """你是一个特斯拉产品助手，可以回答关于特斯拉产品和公司的问题。
你有以下工具可以使用：
- 查询产品名称: 通过产品名称查找产品详细信息，包括价格、外观等
- 公司相关信息: 查询特斯拉公司的相关信息，包括技术创新、产品线、自动驾驶技术等
请根据用户的问题，选择合适的工具来获取信息，然后给出准确、简洁的回答。
注意事项：
1. 当用户询问特定产品（如 Model 3, Model Y 等）时，使用"查询产品名称"工具
2. 当用户询问公司相关信息、技术创新、自动驾驶等时，使用"公司相关信息"工具
3. 回答要简洁明了，重点突出
"""
# ========== 使用 create_agent 创建 Agent ==========
# 注意：参数名是 model 而不是 llm
agent_graph = create_agent(
model=model, # 使用 model 参数（BaseChatModel 类型）
tools=tools, # 工具列表
system_prompt=system_prompt, # 系统提示词
debug=True # 启用调试模式
)
return agent_graph
def main():
"""主函数：交互式问答系统"""
# API密钥配置（建议使用环境变量）
import os
DASHSCOPE_API_KEY = os.getenv(
"DASHSCOPE_API_KEY",
"sk-cd87191136be484d92f1ac53551440f1"
)
if not DASHSCOPE_API_KEY or not DASHSCOPE_API_KEY.startswith("sk-"):
print("⚠️ 请设置有效的DASHSCOPE_API_KEY环境变量")
print("示例：export DASHSCOPE_API_KEY='your-api-key'")
return
try:
# 创建Agent
print("\n正在初始化特斯拉Agent...")
agent_graph = create_tesla_agent(DASHSCOPE_API_KEY)
print("✅ Tesla Agent 启动成功！\n")
# 显示使用说明
print("="*60)
print("🚗 特斯拉产品助手 (LangGraph 架构)")
print("="*60)
print("你可以问以下类型的问题：")
print(" • 产品查询：'Model 3的价格是多少？'")
print(" • 公司信息：'特斯拉有哪些技术创新？'")
print(" • 综合问题：'Model Y和Model 3有什么区别？'")
print("\n输入 'exit' 或按 Ctrl+C 退出")
print("="*60 + "\n")
# 交互循环
while True:
try:
user_input = input("🤔 你想问什么：").strip()
# 退出命令
if user_input.lower() in ['exit', 'quit', '退出', 'q']:
print("👋 感谢使用，再见！")
break
# 空输入跳过
if not user_input:
continue
# 执行Agent（使用 invoke 方式）
print("\n🔄 正在思考...")
# LangGraph 架构的调用方式
result = agent_graph.invoke({
"messages": [{"role": "user", "content": user_input}]
})
# 显示结果
output_response(result["messages"])
except KeyboardInterrupt:
print("\n\n👋 感谢使用，再见！")
break
except Exception as e:
print(f"\n❌ 执行出错: {e}")
import traceback
traceback.print_exc()
print("请重试或换个问题试试\n")
except Exception as e:
print(f"❌ Agent初始化失败: {e}")
import traceback
traceback.print_exc()
print("请检查：")
print(" 1. API密钥是否正确")
print(" 2. 网络连接是否正常")
print(" 3. 依赖包是否正确安装")
if __name__ == "__main__":
main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
# 整体架构概览
这段代码实现了一个特斯拉产品问答助手，它使用 LangChain + LangGraph 框架创建了一个智能 Agent，能够：
- 查询特斯拉产品信息（如 Model 3、Model Y 等）
- 回答特斯拉公司相关问题（如技术创新、自动驾驶等）
- 自动判断使用哪个工具来回答问题
用户提问
↓
[LangGraph Agent] ← 智能调度中心
↓
判断需要什么信息
↓
├─→ [查询产品名称工具] → 返回产品信息
└─→ [公司信息工具] → 调用 LLM → 返回公司信息
↓
整理答案 → 返回给用户
1
2
3
4
5
6
7
8
9
10
2
3
4
5
6
7
8
9
10
# 代码逐段详解
# 导入部分（第 1-17 行）
import textwrap
import time
from langchain_core.tools import StructuredTool
from pydantic import BaseModel, Field
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain.agents import create_agent
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
| 导入内容 | 作用 | 通俗解释 |
|---|---|---|
| textwrap | 文本格式化 | 让长文本自动换行，显示更整齐 |
| time | 时间控制 | 实现打字机效果的延迟 |
| StructuredTool | 工具定义类 | 定义 Agent 可以使用的"工具" |
| BaseModel, Field | 数据验证 | 定义工具的输入参数格式 |
| ChatTongyi | 通义千问聊天模型 | 阿里云的 AI 模型，负责思考和回答 |
| create_agent | Agent 创建函数 | 创建整个智能助手的核心函数 |
为什么需要这些？
比喻：
你要开餐厅，需要：
- textwrap = 服务员（整理订单）
- time = 厨师计时器（控制上菜速度）
- StructuredTool = 厨具（锅碗瓢盆）
- ChatTongyi = 主厨（负责做菜）
- create_agent = 餐厅管理系统（协调一切）
# 辅助函数（第 20-54 行）
# messages: list - 消息列表，包含对话历史
def output_response(messages: list) -> None:
"""动态显示响应结果，打字机效果"""
if not messages:
return
# 1. 获取最后一条消息（Agent 的最终回答）
last_message = messages[-1]
# 2. 提取消息内容
content = last_message.content if hasattr(last_message, 'content') else str(last_message)
if not content:
return
print("\n" + "="*60)
print("回答：")
print("="*60)
# 3. 使用 textwrap.wrap 自动换行（每行 60 个字符）
for line in textwrap.wrap(content, width=60):
for word in line.split():
for char in word:
print(char, end="", flush=True) # 不换行打印，立即显示
time.sleep(0.05)
print(" ", end="", flush=True) # 延迟 0.05 秒，产生打字效果
print()
print("-"*60 + "\n")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# 数据模型定义（第 57-68 行）
class ProductInput(BaseModel):
"""产品查询输入参数"""
product_name: str = Field(description="产品名称，例如：Model 3")
class CompanyInfoInput(BaseModel):
"""公司信息查询输入参数"""
query: str = Field(description="具体的问题")
1
2
3
4
5
6
7
8
2
3
4
5
6
7
8
作用：定义工具的输入参数格式
为什么需要这个？ AI 模型在调用工具时，需要知道：
- 需要什么参数？
- 参数是什么类型？
- 参数的说明是什么？
class ProductInput(BaseModel):
# 继承 BaseModel，让 Pydantic 自动验证数据
product_name: str = Field(
description="产品名称，例如：Model 3"
)
# | | |
# | | └─ 参数说明，告诉 AI 这个参数是什么
# | └─ 使用 Field 定义参数
# └─ 参数类型（字符串）
1
2
3
4
5
6
7
8
9
10
2
3
4
5
6
7
8
9
10
# 数据源类（第 71-120 行）
class TeslaDataSource:
"""特斯拉数据源，提供产品查询和公司信息查询功能"""
def __init__(self, llm):
self.llm = llm
def find_product_description(self, product_name: str) -> str:
"""通过产品名称查询产品描述"""
product_info = {
"Model 3": "具有简洁、动感的外观设计，流线型车身和现代化前脸。定价23.19-33.19万",
"Model Y": "在外观上与Model 3相似，但采用了更高的车身和更大的后备箱空间。定价26.39-36.39万",
"Model X": "拥有独特的翅子门设计和更加大胆的外观风格。定价89.89-105.89万",
"Model S": "特斯拉的旗舰轿车，拥有卓越的性能和超长续航里程。定价69.89-89.89万",
}
return product_info.get(
product_name,
f"没有找到产品'{product_name}'，可用产品：{', '.join(product_info.keys())}"
)
def find_company_info(self, query: str) -> str:
"""查询公司相关信息"""
context = """
特斯拉最知名的产品是电动汽车，其中包括Model S、Model 3、Model X和Model Y等多款车型。
特斯拉以其技术创新、高性能和领先的自动驾驶技术而闻名。
公司不断推动自动驾驶技术的研发，并在车辆中引入各种驾驶辅助功能，
如自动紧急制动、自适应巡航控制和车道保持辅助等。
特斯拉还在能源存储领域有所布局，推出了Powerwall家用电池和太阳能屋顶等产品。
公司由埃隆·马斯克创立，致力于加速世界向可持续能源的转变。
"""
prompt = CONTEXT_QA_TMPL.format(query=query, context=context)
return self.llm.invoke(prompt).content
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
作用：提供数据和查询功能
类比： 这个类就像数据库管理员，负责：
- find_product_description() - 查产品目录
- find_company_info() - 查公司资料库
# Agent 创建函数（第 123-181 行）
def create_tesla_agent(api_key: str):
"""创建特斯拉Agent - 使用 LangGraph 架构"""
# ========== 初始化 Chat 模型 ==========
model = ChatTongyi(
model="qwen-turbo",
dashscope_api_key=api_key
)
# 创建数据源
tesla_data_source = TeslaDataSource(model)
# ========== 定义工具 ==========
tools = [
StructuredTool.from_function(
func=tesla_data_source.find_product_description,
name="查询产品名称",
description="通过产品名称查找产品详细信息，包括价格、外观等。输入参数是产品名称，例如：'Model 3'",
args_schema=ProductInput
),
StructuredTool.from_function(
func=tesla_data_source.find_company_info,
name="公司相关信息",
description="查询特斯拉公司的相关信息，包括技术创新、产品线、自动驾驶技术等。输入参数是具体的问题",
args_schema=CompanyInfoInput
),
]
# ========== System Prompt ==========
system_prompt = """你是一个特斯拉产品助手..."""
# ========== 使用 create_agent 创建 Agent ==========
agent_graph = create_agent(
model=model,
tools=tools,
system_prompt=system_prompt,
debug=True
)
return agent_graph
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
这是核心函数！详细解析：
- 步骤 1：初始化模型
model = ChatTongyi(
model="qwen-turbo", # 使用通义千问 Turbo 版本
dashscope_api_key=api_key # 阿里云 API 密钥
)
1
2
3
4
2
3
4
- 步骤 2：创建数据源
tesla_data_source = TeslaDataSource(model)
# 把模型传给数据源，让它能用 LLM 智能查询
1
2
2
- 步骤 3：定义工具（重点！）
tools = [
StructuredTool.from_function(
func=tesla_data_source.find_product_description, # 工具函数
name="查询产品名称", # 工具名称
description="...", # 工具说明
args_schema=ProductInput # 参数格式
),
# 第二个工具...
]
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
StructuredTool.from_function 参数详解：
| 参数 | 作用 | 示例 |
|---|---|---|
| func | 工具的实际功能函数 | find_product_description |
| name | 工具名称（AI 用它来选择工具） | "查询产品名称" |
| description | 工具功能描述（告诉 AI 何时用这个工具） | "通过产品名称查找产品详细信息..." |
| args_schema | 参数格式定义 | ProductInput |
- 步骤 4：System Prompt
system_prompt = """你是一个特斯拉产品助手，可以回答关于特斯拉产品和公司的问题。
你有以下工具可以使用：
- 查询产品名称: 通过产品名称查找产品详细信息，包括价格、外观等
- 公司相关信息: 查询特斯拉公司的相关信息，包括技术创新、产品线、自动驾驶技术等
请根据用户的问题，选择合适的工具来获取信息，然后给出准确、简洁的回答。
注意事项：
1. 当用户询问特定产品（如 Model 3, Model Y 等）时，使用"查询产品名称"工具
2. 当用户询问公司相关信息、技术创新、自动驾驶等时，使用"公司相关信息"工具
3. 回答要简洁明了，重点突出
"""
1
2
3
4
5
6
7
8
9
10
11
12
13
2
3
4
5
6
7
8
9
10
11
12
13
作用：告诉 AI：
- 你是谁（特斯拉产品助手）
- 你有什么工具
- 如何使用这些工具
- 步骤 5：创建 Agent
agent_graph = create_agent(
model=model, # AI 模型（大脑）
tools=tools, # 工具列表（手脚）
system_prompt=system_prompt, # 角色设定
debug=True # 调试模式
)
1
2
3
4
5
6
2
3
4
5
6
类比：
组装机器人：
┌──────────────┐
│ 大脑 (模型) │ ← ChatTongyi
├──────────────┤
│ 手脚 (工具) │ ← 查询产品、查公司
├──────────────┤
│ 指令 (提示) │ ← 你是特斯拉助手...
└──────────────┘
= Agent
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
# 主函数（第 184-267 行）
def main():
"""主函数：交互式问答系统"""
# API密钥配置
import os
DASHSCOPE_API_KEY = os.getenv(
"DASHSCOPE_API_KEY",
"sk-cd87191136be484d92f1ac53551440f1"
)
try:
# 创建 Agent
print("\n正在初始化特斯拉Agent...")
agent_graph = create_tesla_agent(DASHSCOPE_API_KEY)
print("✅ Tesla Agent 启动成功！\n")
# 交互循环
while True:
try:
user_input = input("🤔 你想问什么：").strip()
# 退出命令
if user_input.lower() in ['exit', 'quit', '退出', 'q']:
print("👋 感谢使用，再见！")
break
# 空输入跳过
if not user_input:
continue
# 执行 Agent
print("\n🔄 正在思考...")
result = agent_graph.invoke({
"messages": [{"role": "user", "content": user_input}]
})
# 显示结果
output_response(result["messages"])
except KeyboardInterrupt:
print("\n\n👋 感谢使用，再见！")
break
except Exception as e:
print(f"\n❌ 执行出错: {e}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
核心逻辑：agent_graph.invoke()
result = agent_graph.invoke({
"messages": [{"role": "user", "content": user_input}]
})
1
2
3
2
3
# 完整工作流程
用户问："Model 3 的价格是多少？"
1. 用户输入
"Model 3 的价格是多少？"
2. 调用 Agent
agent_graph.invoke(messages)
3. Agent 思考（LangGraph 内部流程）
├─ 分析问题：这是产品查询
├─ 选择工具：query_product（查询产品名称）
├─ 准备参数：{"product_name": "Model 3"}
└─ 调用工具：find_product_description("Model 3")
4. 工具执行
find_product_description("Model 3")
└─ 返回："具有简洁、动感的外观设计...定价23.19-33.19万"
5. Agent 整理答案
使用 LLM 将工具返回的信息整理成自然的回答
6. 返回结果
result["messages"] 包含完整对话历史
7. 显示给用户（打字机效果）
"Model 3 的价格是 23.19-33.19 万，它具有简洁、动感的外观设计..."
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 关键技术概念
- 什么是 Agent？
定义：Agent 是一个能够自主思考和行动的 AI 系统。
特点：
- 🧠 有大脑（LLM 模型）
- 🔧 有工具（可以调用外部功能）
- 🎯 有目标（完成特定任务）
类比：
- 没有 Agent 的 AI = 只会回答问题的学生
- 有 Agent 的 AI = 会查资料、会使用工具的研究员
- 什么是 Tool？
定义：Tool 是 Agent 可以调用的外部功能。
类型：
- 数据查询工具：查询数据库、API
- 计算工具：数学计算、数据分析
- 生成工具：生成图片、文档
- 执行工具：发送邮件、调用其他系统
- 什么是 LangGraph？
定义：LangChain 的新一代框架，用于构建有状态的 Agent 应用。
特点：
- 🔄 支持循环和条件判断
- 📊 状态管理（对话历史）
- 🧩 模块化设计
架构：
StateGraph（状态图）
├─ Nodes（节点）：模型节点、工具节点
├─ Edges（边）：节点之间的连接
└─ State（状态）：共享的对话历史
1
2
3
4
2
3
4
- Pydantic 的作用
定义：Python 的数据验证库。
作用：
- ✅ 自动验证数据类型
- ✅ 生成 JSON Schema
- ✅ 提供类型提示
示例：
class ProductInput(BaseModel):
product_name: str = Field(description="产品名称")
# 自动验证
ProductInput(product_name="Model 3") ✅ 正确
ProductInput(product_name=123) ❌ 错误（类型不对）
1
2
3
4
5
6
2
3
4
5
6
# 核心要点
- Agent = 大脑 + 工具 + 指令
- 大脑：ChatTongyi 模型
- 工具：StructuredTool 定义的查询功能
- 指令：system_prompt 系统提示词
工作流程 用户提问 → Agent 思考 → 选择工具 → 执行工具 → 整理答案 → 返回
关键技术
- LangChain：工具和模型管理
- LangGraph：Agent 框架
- Pydantic：数据验证
- 通义千问：AI 模型
# 运行效果
