「工程实践」LangChain 入门与实战

# 示例：提示词 → 模型 → 输出解析器
chain = prompt | llm | output_parser

OPENAI_API_KEY=your-dashscope-api-key
OPENAI_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1

[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "langchain-demo"
version = "0.1.0"
description = "LangChain Core Knowledge and Practice"
requires-python = ">=3.10"
dependencies = [
    "openai>=0.1.0",
    "langchain>=0.1.0",
    "langchain-openai>=0.1.0",
    "python-dotenv>=1.0.0",
    "pydantic-settings>=2.0",
]

[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]

[tool.ruff]
line-length = 120

# 使用 uv 安装依赖（推荐，更快更可靠）
uv sync

# 或使用 pip（需在项目根目录运行）
pip install -e .

import os
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()  # 从 .env 文件加载 API Key

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=[{"role": "user", "content": "你是谁"}]
)

print(completion.model_dump_json())

{"id":"chatcmpl-d8b7bf2c-f581-923d-89c7-104b7c1316e4","choices":[{"finish_reason":"stop","index":0,"logprobs":null,"message":{"content":"你好！我是 DeepSeek，由深度求索公司创造的 AI 助手。😊\n\n我是一个纯文本模型，可以帮你解答问题、处理信息、进行对话。虽然我不支持多模态识别（比如直接识别图片内容），但我可以通过文件上传功能读取图像、txt、pdf、ppt、word、excel 等文件中的文字信息来帮助你。\n\n我的一些特点：\n- **免费使用**，无需付费\n- **上下文长度 1M**，可以一次性处理像《三体》三部曲那么大体量的内容\n- 支持**联网搜索**（需要手动开启）\n- **知识截止日期**到 2025 年 5 月\n- App 端支持**语音输入** 功能\n\n有什么我可以帮你的吗？无论是学习、工作还是日常问题，我都很乐意帮忙！🌟","refusal":null,"role":"assistant","annotations":null,"audio":null,"function_call":null,"tool_calls":null}}],"created":1780278674,"model":"deepseek-v4-pro","object":"chat.completion","service_tier":null,"system_fingerprint":null,"usage":{"completion_tokens":173,"prompt_tokens":5,"total_tokens":178,"completion_tokens_details":null,"prompt_tokens_details":{"audio_tokens":null,"cached_tokens":0}}}

import os
import sys
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

sys.stdout.reconfigure(encoding='utf-8')

load_dotenv()

llm = ChatOpenAI(
    model_name="deepseek-v4-pro"
)

response = llm.invoke("你是谁")
print(response.content)

你好！我是 DeepSeek，由深度求索公司创造的 AI 助手。😊

我是一个纯文本模型，可以帮你解答各种问题、提供信息、进行对话交流等等。
我的特点包括：完全免费使用、上下文长度 1M、支持联网搜索、支持语音输入（App 端）。
知识截止到 2025 年 5 月。

有什么我可以帮你的吗？无论是学习、工作还是日常问题，尽管问我！🌟

llm = ChatOpenAI(
    model_name="deepseek-v4-pro",
    streaming=True,
)
for chunk in llm.stream("你是谁"):
    print(chunk.content, end="", flush=True)

import os
from dotenv import load_dotenv
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

load_dotenv()

# 1. 定义工具函数
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city} 25℃ 晴"

# 2. 绑定工具函数到 LLM
llm = ChatOpenAI(model_name="deepseek-v4-pro")
llm_with_tools = llm.bind_tools([get_weather])

# 3. 调用并处理工具调用
response = llm_with_tools.invoke("深圳今天天气怎么样？")
if response.tool_calls:
    for tool_call in response.tool_calls:
        print(f"需要调用工具：{tool_call['name']}")
        print(f"参数：{tool_call['args']}")
        if tool_call['name'] == "get_weather":
            result = get_weather.invoke(tool_call['args'])
            print(f"工具返回结果：{result}")
            final_response = llm.invoke(f"总结：{result}")
            print(f"最终回复：{final_response.content}")

需要调用工具：get_weather
参数：{'city': '深圳'}
工具返回结果：深圳 25℃ 晴
最终回复：根据您提供的信息，深圳当前天气晴朗，气温25℃。

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

load_dotenv()

llm = ChatOpenAI(model_name="deepseek-v4-pro")
history = InMemoryChatMessageHistory()

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

chain = prompt | llm

def chat(input_text: str) -> str:
    response = chain.invoke({
        "history": history.messages,
        "input": input_text
    })
    history.add_user_message(input_text)
    history.add_ai_message(response.content)
    return response.content

# 多轮对话
chat("你好，我叫小明")
chat("我叫什么名字？")  # AI 会记住"小明"

# 打印交互过程
print("完整对话历史:")
for i, message in enumerate(history.messages, 1):
    role = "用户" if message.type == "human" else "AI"
    print(f"{i}. [{role}]: {message.content}")

完整对话历史:
1. [用户]: 你好，我叫小明
2. [AI]: 你好呀，小明！很高兴认识你～有什么我可以帮你的吗？😊
3. [用户]: 我叫什么名字？
4. [AI]: 你刚才告诉我啦，你叫**小明**～ 😄 放心，我记着呢！还有什么想聊的吗？

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate

load_dotenv()

template = PromptTemplate.from_template("给我讲个关于{subject}的一句话笑话")
print("===Template===")
print(template)

llm = ChatOpenAI(model_name="deepseek-v4-pro")
ret = llm.invoke(template.format(subject='兔子'))
print("===打印输出===")
print(ret.content)

===Template===
input_variables=['subject'] input_types={} partial_variables={} template='给我讲个关于{subject}的一句话笑话'
===打印输出===
“兔子红着眼睛对胡萝卜说：我可不是为你哭的，是昨晚熬夜打游戏熬的！”

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

load_dotenv()

prompt_template = ChatPromptTemplate([
    ("system", "translate the following into {language}."),
    ("user", "{text}")
])

llm = ChatOpenAI(model_name="deepseek-v4-pro")
parser = StrOutputParser()

chain = prompt_template | llm | parser

for token in chain.stream({
    "language": "English",
    "text": "你好，现在是几点？"
}):
    print(token, end="", flush=True)

Hello, what time is it now?

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import List
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser

load_dotenv()

llm = ChatOpenAI(model_name="deepseek-v4-pro", temperature=0)

# 定义数据模型
class ProductInfo(BaseModel):
    """商品信息"""
    name: str = Field(description="商品名称")
    price: float = Field(description="价格")
    category: str = Field(description="分类")
    features: List[str] = Field(description="特点列表")

# 创建解析器
parser = PydanticOutputParser(pydantic_object=ProductInfo)

# 构建 Prompt
prompt = ChatPromptTemplate.from_template("""
    {format_instructions}
    请分析以下产品：{product}
""")

chain = prompt | llm | parser

# 执行
result = chain.invoke({
    "product": "iPhone 17",
    "format_instructions": parser.get_format_instructions()
})

print(result)
print(result.price)  # 7999.0

name='iPhone 17' price=7999.0 category='智能手机' features=['A19芯片', '6.7英寸 OLED 屏幕', ...]
7999.0

pip install langsmith

# .env 文件
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=lsv2_pt_your_key_here
LANGCHAIN_PROJECT=my-agent-project