环境
- uv 0.8.10 (Homebrew 2025-08-13)
- python 3.13
项目目录
wanghaima@wanghaimadeMacBook-Pro langchain01 % tree
.
├── 1.my_llm.py
├── 2.init_llm.py
├── README.md
├── __pycache__
│ ├── 2.init_llm.cpython-313.pyc
│ ├── env_utils.cpython-313.pyc
│ └── test_2_init_llm.cpython-313-pytest-9.0.3.pyc
├── .env
├── env_utils.py
├── main.py
├── pyproject.toml
├── test_2_init_llm.py
└── uv.lockpyproject.toml
[project]
name = "langchain01"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"langchain-deepseek>=1.0.1",
"pytest>=9.0.3",
"python-dotenv>=1.2.2",
]
.env
DEEPSEEK_API_KEY="sk-88a7bddc09724dc5a59abb8013dffa23hm"
DEEPSEEK_BASE_URL="https://api.deepseek.com"env_utils.py
import os
from dotenv import load_dotenv
# 加载 .env 文件
load_dotenv('.env', override=True)
# 获取当前环境，默认为开发环境
DEEPSEEK_API_KEY = os.getenv('DEEPSEEK_API_KEY', '')
DEEPSEEK_BASE_URL = os.getenv('DEEPSEEK_BASE_URL', '')1.2.init_llm.py
import asyncio
from typing import Optional
from pydantic import BaseModel, Field
from langchain_deepseek.chat_models import ChatDeepSeek
from langchain_core.tools import tool
from env_utils import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL
# 两种方式创建大模型对象，
# 1.通用使用模型类（推荐）
# https://docs.langchain.com/oss/python/integrations/chat/deepseek
deepseek_llm = ChatDeepSeek(
model="deepseek-chat",
temperature=0,
max_tokens=None,
timeout=None,
max_retries=2,
api_key=DEEPSEEK_API_KEY,
base_url=DEEPSEEK_BASE_URL,
# other params...
)
# demo1: 简单字符串调用
def demo1_simple_invoke():
"""简单字符串调用示例"""
print("=" * 50)
print("Demo1: 简单字符串调用")
print("=" * 50)
resp = deepseek_llm.invoke("你好")
print(f"类型: {type(resp)}")
print(f"响应: {resp}")
# output
# 类型: <class 'langchain_core.messages.ai.AIMessage'>
# 响应: content='你好！很高兴见到你。有什么我可以帮你的吗？无论是聊天、解答问题，还是需要一些建议，我都在这里。😊' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 29, 'prompt_tokens': 5, 'total_tokens': 34, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 5}, 'model_provider': 'deepseek', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_058df29938_prod0820_fp8_kvcache_20260402', 'id': 'f2029618-bc4c-4349-bb2d-2f51114499b9', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--019e05af-52e1-74f1-877d-d474e0c6dcc5-0' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 5, 'output_tokens': 29, 'total_tokens': 34, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}
# demo2: 消息列表调用
def demo2_messages_invoke():
"""消息列表调用示例"""
print("\n" + "=" * 50)
print("Demo2: 消息列表调用")
print("=" * 50)
messages = [
("system", "You are a helpful translator. Translate the user sentence to French."),
("human", "I love programming."),
]
resp = deepseek_llm.invoke(messages)
print(f"类型: {type(resp)}")
print(f"响应: {resp}")
# output
# 类型: <class 'langchain_core.messages.ai.AIMessage'>
# 响应: content="J'adore la programmation." additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 7, 'prompt_tokens': 21, 'total_tokens': 28, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 21}, 'model_provider': 'deepseek', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_058df29938_prod0820_fp8_kvcache_20260402', 'id': '3db947c0-1a90-4343-b2a2-e286b1425ac9', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--019e05b4-b36d-7732-90d2-c1b7ddd6df04-0' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 21, 'output_tokens': 7, 'total_tokens': 28, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}
# demo3: 流式输出
def demo3_stream():
"""流式输出示例"""
print("\n" + "=" * 50)
print("Demo3: 流式输出")
print("=" * 50)
for chunk in deepseek_llm.stream("你好，你是谁？"):
print(chunk.text, end="")
# ouput
# 你好！我是DeepSeek，由深度求索公司创造的AI助手。很高兴认识你！😊
# 我是一个纯文本模型，可以帮你解答问题、处理文档、进行对话交流等。我支持阅读链接、上传文件（图片、PDF、Word、Excel等），还能一次性处理超长文本（比如整本书的内容）。最重要的是，我完全免费使用！
# 有什么我可以帮你的吗？无论是学习、工作还是日常问题，尽管问我！✨%
# demo4: 流式输出并累积完整响应
def demo4_stream_accumulate():
"""流式输出并累积完整响应示例"""
print("\n" + "=" * 50)
print("Demo4: 流式输出并累积完整响应")
print("=" * 50)
stream = deepseek_llm.stream("你好,你是谁？")
full = next(stream)
for chunk in stream:
full += chunk
print(chunk.text, end="")
full
print(f"\n完整响应: {full}")
# output
# 完整响应: content='你好！我是DeepSeek，由深度求索公司创造的AI助手。很高兴认识你！😊\n\n我是一个纯文本模型，可以帮你解答问题、处理信息、进行对话等等。虽然我不支持多模态识别（比如直接识别图片内容），但我可以读取你上传的文件（图片、PDF、Word、Excel等）中的文字信息来帮助你。\n\n我的特点包括：\n- **免费使用**，没有收费计划\n- **超长上下文**（1M tokens），可以一次性处理像《三体》三部曲那么大体量的内容\n- **支持联网搜索**（需要手动开启）\n- **支持语音输入**（App端）\n\n有什么我可以帮你的吗？无论是学习、工作还是日常问题，尽管问我！💪' additional_kwargs={} response_metadata={'model_provider': 'deepseek', 'finish_reason': 'stop', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_058df29938_prod0820_fp8_kvcache_20260402'} id='lc_run--019e05b7-0c8e-7900-93ef-499b244332cd' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 8, 'output_tokens': 155, 'total_tokens': 163, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}} tool_call_chunks=[] chunk_position='last'
# demo5: 异步调用
async def demo5_async():
"""异步调用示例"""
print("\n" + "=" * 50)
print("Demo5: 异步调用")
print("=" * 50)
messages = [
("system", "You are a helpful translator. Translate the user sentence to French."),
("human", "I love programming."),
]
# 异步调用
resp = await deepseek_llm.ainvoke(messages)
print(f"类型: {type(resp)}")
print(f"resp:{resp}")
print(f"异步调用结果: {resp.content}")
# 异步流式调用
print("\n异步流式输出:")
async for chunk in deepseek_llm.astream(messages):
print(chunk.text, end="")
# demo6: Tool calling (工具调用)
@tool
def get_current_weather(location: str, unit: str = "celsius"):
"""
获取指定地点的当前天气
Args:
location: 城市名称，如 "北京"、"上海"
unit: 温度单位，"celsius" 或 "fahrenheit"
Returns:
天气信息字符串
"""
# 这里是模拟数据，实际应用中应该调用真实的天气 API
weather_data = {
"北京": {"temperature": "22", "condition": "晴天"},
"上海": {"temperature": "25", "condition": "多云"},
"广州": {"temperature": "28", "condition": "晴天"},
}
if location in weather_data:
data = weather_data[location]
return f"{location}当前天气: {data['condition']}, 温度 {data['temperature']}°{unit[0].upper()}"
else:
return f"未找到{location}的天气信息"
def demo6_tool_calling():
"""工具调用示例"""
print("\n" + "=" * 50)
print("Demo6: Tool Calling (工具调用)")
print("=" * 50)
# 绑定工具到模型
tools = [get_current_weather]
llm_with_tools = deepseek_llm.bind_tools(tools)
# 调用模型
query = "北京和上海的天气怎么样？"
print(f"\n用户提问: {query}")
response = llm_with_tools.invoke(query)
print(f"\n模型回答response: {response}")
# output
# 模型回答response: content='我来查一下北京和上海的天气情况。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 120, 'prompt_tokens': 335, 'total_tokens': 455, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 256}, 'prompt_cache_hit_tokens': 256, 'prompt_cache_miss_tokens': 79}, 'model_provider': 'deepseek', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_058df29938_prod0820_fp8_kvcache_20260402', 'id': '3238ec04-417c-47a7-b7c1-7d41d35a68c4', 'finish_reason': 'tool_calls', 'logprobs': None} id='lc_run--019dfb4b-f6d6-7112-9541-691488637f9d-0' tool_calls=[{'name': 'get_current_weather', 'args': {'location': '北京', 'unit': 'celsius'}, 'id': 'call_00_QuoniczsWct0Qa6Mti8F8874', 'type': 'tool_call'}, {'name': 'get_current_weather', 'args': {'location': '上海', 'unit': 'celsius'}, 'id': 'call_01_FZtArqKfntl9xF6g7eXY6113', 'type': 'tool_call'}] invalid_tool_calls=[] usage_metadata={'input_tokens': 335, 'output_tokens': 120, 'total_tokens': 455, 'input_token_details': {'cache_read': 256}, 'output_token_details': {}}
# ***这里需要自己写代码判断是否需要调用工具，模型不会主动调用。不过可以使用create_agent()方式调用工具，模型就可以自动循环调用工具了。***
# 检查是否有工具调用
if response.tool_calls:
print(f"\n模型决定调用工具:")
for tool_call in response.tool_calls:
print(f" 工具名称: {tool_call['name']}")
print(f" 工具参数: {tool_call['args']}")
# 执行工具调用
if tool_call['name'] == 'get_current_weather':
result = get_current_weather.invoke(tool_call['args'])
print(f" 工具返回: {result}")
else:
print(f"模型直接回答: {response.content}")
# demo7: Structured output (结构化输出)
class PersonInfo(BaseModel):
"""人物信息结构"""
name: str = Field(description="人物姓名")
age: int = Field(description="人物年龄")
occupation: str = Field(description="职业")
hobbies: list = Field(description="兴趣爱好列表")
email: Optional[str] = Field(default=None, description="电子邮箱")
def demo7_structured_output():
"""结构化输出示例"""
print("\n" + "=" * 50)
print("Demo7: Structured Output (结构化输出)")
print("=" * 50)
# 使用 with_structured_output 方法
structured_llm = deepseek_llm.with_structured_output(PersonInfo)
# 调用模型获取结构化数据
query = "请介绍一个虚构的人物，包括姓名、年龄、职业和兴趣爱好"
print(f"用户提问: {query}\n")
result = structured_llm.invoke(query)
# result 是一个 PersonInfo 对象
print(f"类型: {type(result)}")
print(f"resp: {result}")
# output
# 类型: <class '__main__.PersonInfo'>
# resp: name='林逸风' age=28 occupation='独立游戏开发者' hobbies=['摄影', '攀岩', '弹吉他', '阅读科幻小说'] email=None
print(f"姓名: {result.name}")
print(f"年龄: {result.age}")
print(f"职业: {result.occupation}")
print(f"爱好: {result.hobbies}")
if result.email:
print(f"邮箱: {result.email}")
# output
# 姓名: 林逸风
# 年龄: 28
# 职业: 独立游戏开发者
# 爱好: ['摄影', '攀岩', '弹吉他', '阅读科幻小说']
# demo8: Token usage (Token 使用统计)
def demo8_token_usage():
"""Token 使用统计示例"""
print("\n" + "=" * 50)
print("Demo8: Token Usage (Token 使用统计)")
print("=" * 50)
# 调用模型
response = deepseek_llm.invoke("请用50个字介绍一下Python编程语言")
# 获取 token 使用信息
print(f"响应内容response: {response}\n")
# 响应内容response: content='Python是一种易学、开源的高级编程语言，语法简洁，支持多种编程范式。拥有丰富的标准库和第三方库，广泛应用于Web开发、数据分析、人工智能等领域。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 35, 'prompt_tokens': 12, 'total_tokens': 47, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 12}, 'model_provider': 'deepseek', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_058df29938_prod0820_fp8_kvcache_20260402', 'id': 'b342f9e2-e9a6-4f0f-8f21-0e910c83cb2a', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--019dfb5b-798a-7282-83c7-6ef95ae8d4ef-0' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 12, 'output_tokens': 35, 'total_tokens': 47, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}
# print(f"响应内容: {response.content}\n")
# 如果 response 对象有 response_metadata 属性，才执行这里的代码
# response 对象包含 token 使用统计
if hasattr(response, 'response_metadata'):
metadata = response.response_metadata
if 'token_usage' in metadata:
token_usage = metadata['token_usage']
print("Token 使用统计:")
print(f" 提示词 Token: {token_usage.get('prompt_tokens', 'N/A')}")
print(f" 完成 Token: {token_usage.get('completion_tokens', 'N/A')}")
print(f" 总计 Token: {token_usage.get('total_tokens', 'N/A')}")
else:
print("响应元数据中没有 token_usage 信息")
print(f"可用的元数据键: {list(metadata.keys())}")
# output:
# Token 使用统计:
# 提示词 Token: 12
# 完成 Token: 35
# 总计 Token: 47
# demo9: Response metadata (响应元数据)
def demo9_response_metadata():
"""响应元数据示例"""
print("\n" + "=" * 50)
print("Demo9: Response Metadata (响应元数据)")
print("=" * 50)
# 调用模型
response = deepseek_llm.invoke("你好")
# 获取完整的响应对象信息
print(f"响应类型: {type(response)}")
print(f"响应内容: {response}\n")
# output
# 响应类型: <class 'langchain_core.messages.ai.AIMessage'>
# 响应内容: content='你好！很高兴见到你。有什么我可以帮你的吗？无论是聊天、解答问题，还是需要一些建议，我都在这里。😊' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 29, 'prompt_tokens': 5, 'total_tokens': 34, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 5}, 'model_provider': 'deepseek', 'model_name': 'deepseek-v4-flash', 'system_fingerprint': 'fp_058df29938_prod0820_fp8_kvcache_20260402', 'id': '50f8065c-6c32-4cf3-bccd-e3fdda44934b', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--019dfb60-6607-7cd3-9e77-5b43693fbe42-0' tool_calls=[] invalid_tool_calls=[] usage_metadata={'input_tokens': 5, 'output_tokens': 29, 'total_tokens': 34, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}
# print(f"响应内容: {response.content}\n")
# 查看响应元数据
print("响应元数据:")
print("-" * 40)
if hasattr(response, 'response_metadata'):
for key, value in response.response_metadata.items():
print(f"{key}: {value}")
# output:
# 响应元数据:
# ----------------------------------------------
# token_usage: {'completion_tokens': 29, 'prompt_tokens': 5, 'total_tokens': 34, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 5}
# model_provider: deepseek
# model_name: deepseek-v4-flash
# system_fingerprint: fp_058df29938_prod0820_fp8_kvcache_20260402
# id: 50f8065c-6c32-4cf3-bccd-e3fdda44934b
# finish_reason: stop
# logprobs: None
# 查看响应 ID
if hasattr(response, 'id'):
print(f"\n响应 ID: {response.id}")
# output:
# 响应 ID: 50f8065c-6c32-4cf3-bccd-e3fdda44934b
# 查看模型信息
if hasattr(response, 'response_metadata'):
metadata = response.response_metadata
if 'model' in metadata:
print(f"使用模型: {metadata['model']}")
if 'finish_reason' in metadata:
print(f"结束原因: {metadata['finish_reason']}")
# output:
# 结束原因: stop
def main():
"""主函数，分别调用各个demo"""
# print("DEEPSEEK_API_KEY", DEEPSEEK_API_KEY)
# print("DEEPSEEK_BASE_URL", DEEPSEEK_BASE_URL)
# demo1_simple_invoke()
# demo2_messages_invoke()
# demo3_stream()
# demo4_stream_accumulate()
# asyncio.run(demo5_async())
# demo6_tool_calling()
demo7_structured_output()
# demo8_token_usage()
# demo9_response_metadata()
if __name__ == "__main__":
main()
作者：海马 创建时间：2026-05-06 11:53
最后编辑：海马 更新时间：2026-05-13 07:11
最后编辑：海马 更新时间：2026-05-13 07:11
