from langchain.chat_models import init_chat_model

model = init_chat_model(

"openai:gpt-4.1-mini",

temperature=0.2,

max_tokens=500,

timeout=30,

)

model = init_chat_model(

"openai:gpt-4.1-mini",

base_url="http://127.0.0.1:3000/v1",

temperature=0.2,

)

OPENAI_API_KEY=your_openai_api_key_here

OPENAI_BASE_URL=http://127.0.0.1:3000/v1

OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "").strip()

model = build_model(base_url=OPENAI_BASE_URL orNone)

response = model.invoke("请用 3 句话解释：为什么很多应用需要结构化输出？")

for chunk in model.stream("请把消息和模型的区别讲给一个初学者听"):

    print(chunk)

responses = model.batch([

"一句话解释 temperature 的作用",

"一句话解释 max_tokens 的作用",

"一句话解释 timeout 的作用",

])

import os

from pathlib import Path

from pprint import pprint

defload_simple_env(path: str = ".env") -> None:

"""不依赖 python-dotenv，直接把 .env 里的键值读进环境变量。"""

    env_path = Path(path)

ifnot env_path.exists():

print(f"未找到 {path}，后续需要真实调用模型时请手动配置 API Key。")

return

for raw_line in env_path.read_text(encoding="utf-8").splitlines():

        line = raw_line.strip()

ifnot line or line.startswith("#") or"="notin line:

continue

        key, value = line.split("=", 1)

        os.environ.setdefault(key.strip(), value.strip())

defget_text(message) -> str:

    content = getattr(message, "content", message)

ifisinstance(content, str):

return content

returnstr(content)

load_simple_env()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "").strip()

OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", "").strip()

HAS_OPENAI_KEY = bool(OPENAI_API_KEY) and"your_openai_api_key_here"notin OPENAI_API_KEY

print("OPENAI_API_KEY 已配置：", HAS_OPENAI_KEY)

print("OPENAI_BASE_URL：", OPENAI_BASE_URL or"未配置，默认直连官方接口")

ifnot HAS_OPENAI_KEY:

print("提示：当前 Notebook 的很多示例可以先看代码结构；如需真实调用模型，请先把 .env 里的 OPENAI_API_KEY 改成你的真实密钥。")

OPENAI_API_KEY=your_openai_api_key_here

OPENAI_BASE_URL=https://your-openai-compatible-endpoint/v1

from langchain.chat_models import init_chat_model

from langchain.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage

defbuild_model(

model_name: str = "openai:gpt-4.1-mini",

temperature: float = 0.2,

base_url: str | None = None,

):

"""构造一个聊天模型；如果没配置密钥，就返回 None。"""

ifnot HAS_OPENAI_KEY:

returnNone

    model_kwargs = {

"temperature": temperature,

"max_tokens": 500,

"timeout": 30,

    }

if base_url:

        model_kwargs["base_url"] = base_url

return init_chat_model(model_name, **model_kwargs)

model = build_model(base_url=OPENAI_BASE_URL orNone)

print("当前模型对象：", model)

if model isnotNone:

    single_turn_response = model.invoke("请用 3 句话解释：为什么很多应用需要结构化输出？")

print("\n【单轮调用结果】")

print(get_text(single_turn_response))

    conversation = [

        SystemMessage("你是一个面向初学者的 Python 助手，回答要短句、清楚、少术语。"),

        HumanMessage("我在学 LangChain，message 是干什么的？"),

        AIMessage("message 可以理解成一条对话记录，它把角色、内容和元数据包在一起。"),

        HumanMessage("那它和普通字符串最大的区别是什么？"),

    ]

    multi_turn_response = model.invoke(conversation)

print("\n【多轮调用结果】")

print(get_text(multi_turn_response))

else:

print("未进行真实模型调用：请先配置 OPENAI_API_KEY。")

if model isnotNone:

print("=== stream(): 边生成边返回 ===")

    streamed_parts = []

for chunk in model.stream("请把“消息”和“模型”的区别讲给一个完全没学过 LangChain 的人听。"):

        text = get_text(chunk)

if text:

            streamed_parts.append(text)

print(text, end="", flush=True)

print("\n\n合并后的完整输出：")

print("".join(streamed_parts))

print("\n=== batch(): 一次并发处理多个独立问题 ===")

    batch_questions = [

"一句话解释 temperature 的作用",

"一句话解释 max_tokens 的作用",

"一句话解释 timeout 的作用",

    ]

    batch_responses = model.batch(batch_questions)

for question, answer inzip(batch_questions, batch_responses):

print(f"- 问题：{question}")

print(f"  回答：{get_text(answer)}")

else:

print("stream / batch 示例代码已经准备好；配置好 API Key 后即可直接运行。")

print("\n小结：")

print("- invoke()：我只关心最终答案")

print("- stream()：我想边生成边展示，提升交互体验")

print("- batch()：我有一批互不依赖的问题，想一起处理")

from langchain.tools import tool

@tool

defget_store_hours(branch: str) -> str:

"""查询门店营业时间。"""

    fake_hours = {

"北京国贸店": "营业时间：10:00-22:00",

"上海徐家汇店": "营业时间：10:00-21:30",

    }

return fake_hours.get(branch, f"{branch}：营业时间暂未录入。")

@tool

defget_product_stock(product_name: str) -> str:

"""查询商品库存。"""

    fake_stock = {

"机械键盘": "机械键盘：北京国贸店当前库存 12 件",

"无线鼠标": "无线鼠标：北京国贸店当前库存 3 件",

    }

return fake_stock.get(product_name, f"{product_name}：暂无库存信息。")

defrun_tool_loop(user_question: str):

if model isNone:

print("当前未配置 API Key，先阅读代码结构即可。")

returnNone

    model_with_tools = model.bind_tools([get_store_hours, get_product_stock])

    messages = [{"role": "user", "content": user_question}]

    ai_message = model_with_tools.invoke(messages)

    messages.append(ai_message)

print("=== 第一次模型输出 ===")

print("文本内容：", get_text(ai_message))

print("工具调用：")

    pprint(ai_message.tool_calls)

for tool_call in ai_message.tool_calls:

if tool_call["name"] == "get_store_hours":

            tool_result = get_store_hours.invoke(tool_call)

elif tool_call["name"] == "get_product_stock":

            tool_result = get_product_stock.invoke(tool_call)

else:

continue

        messages.append(tool_result)

    final_response = model_with_tools.invoke(messages)

print("\n=== 最终回答 ===")

print(get_text(final_response))

return final_response

run_tool_loop("我今晚想去北京国贸店买机械键盘，请告诉我几点关门，以及现在有没有货。")

from pydantic import BaseModel, Field

classSupportTicket(BaseModel):

"""售后工单结构化结果。"""

    intent: str = Field(description="用户主要诉求，例如补发、换货、退款、咨询")

    urgency: str = Field(description="紧急程度，例如低、中、高")

    needs_human: bool = Field(description="是否建议转人工处理")

    reply_summary: str = Field(description="给客服看的 1 句话处理建议")

if model isnotNone:

    ticket_model = model.with_structured_output(SupportTicket)

    ticket = ticket_model.invoke(

"用户留言：我上周买的台灯到货后底座就是歪的，孩子今晚要写作业，麻烦尽快帮我换一个。"

    )

print("Pydantic 结构化输出结果：")

    pprint(ticket.model_dump())

else:

print("Pydantic 示例代码已准备好；配置好 API Key 后即可直接运行。")

print("\n这一类最适合：")

print("- Python 项目内部直接消费结果")

print("- 需要字段描述、默认值、验证能力")

from typing_extensions import Annotated, TypedDict

classMeetingRoomRequest(TypedDict):

    requester: Annotated[str, ..., "申请人姓名"]

    meeting_time: Annotated[str, ..., "会议时间，例如 2026-04-20 15:00-16:00"]

    attendee_count: Annotated[int, ..., "参会人数"]

    room_size: Annotated[str, ..., "建议会议室大小，例如小型、中型、大型"]

    needs_projector: Annotated[bool, ..., "是否需要投影设备"]

    purpose: Annotated[str, ..., "会议用途摘要"]

if model isnotNone:

    typed_dict_model = model.with_structured_output(MeetingRoomRequest)

    booking = typed_dict_model.invoke(

"帮我预约明天下午3点到4点的大会议室，我叫周宁，8个人开复盘会，需要投影仪。"

    )

print("TypedDict 结构化输出结果：")

    pprint(booking)

else:

print("TypedDict 示例代码已准备好；配置好 API Key 后即可直接运行。")

print("\n这一类最适合：")

print("- 你就想拿到一个 dict")

print("- 不需要 Pydantic 的模型对象和额外校验能力")

import json

visitor_request_schema = {

"title": "VisitorRegistrationRequest",

"description": "办公楼访客登记请求",

"type": "object",

"properties": {

"visitor_name": {"type": "string", "description": "访客姓名"},

"visit_date": {"type": "string", "description": "来访日期，例如 2026-04-20"},

"arrival_time": {"type": "string", "description": "预计到达时间，例如 10:00"},

"location": {"type": "string", "description": "来访地点或办公室"},

"stay_hours": {"type": "number", "description": "预计停留时长，单位小时"},

"needs_visitor_code": {"type": "boolean", "description": "是否需要前台生成访客码"},

"carry_equipment": {"type": "string", "description": "携带设备说明，没有则填无"},

    },

"required": [

"visitor_name",

"visit_date",

"arrival_time",

"location",

"stay_hours",

"needs_visitor_code",

"carry_equipment",

    ],

}

visitor_prompt = (

"明天上午10点，供应商李俊会来我们上海办公室，预计停留2小时，"

"需要前台帮忙开访客码，他会带一台样机进楼。"

)

print("JSON Schema 长这样：")

print(json.dumps(visitor_request_schema, ensure_ascii=False, indent=2))

deftry_structured_call(method_name: str, include_raw: bool = False):

if model isNone:

print(f"{method_name} 示例代码已准备好；配置好 API Key 后即可直接运行。")

returnNone

try:

        structured_model = model.with_structured_output(

            visitor_request_schema,

method=method_name,

include_raw=include_raw,

        )

        result = structured_model.invoke(visitor_prompt)

print(f"\n=== method='{method_name}' 的结果 ===")

        pprint(result)

return result

exceptExceptionas exc:

print(f"\n=== method='{method_name}' 当前未成功执行 ===")

print(type(exc).__name__, str(exc))

returnNone

json_schema_result = try_structured_call("json_schema")

function_calling_result = try_structured_call("function_calling")

json_mode_result = try_structured_call("json_mode")

json_schema_raw_result = try_structured_call("json_schema", include_raw=True)

print("\n=== 一个最简单的手动验证示例 ===")

parsed_result = None

ifisinstance(json_schema_result, dict):

    parsed_result = json_schema_result

elifisinstance(function_calling_result, dict):

    parsed_result = function_calling_result

elifisinstance(json_mode_result, dict):

    parsed_result = json_mode_result

if parsed_result isnotNone:

    missing_fields = [field for field in visitor_request_schema["required"] if field notin parsed_result]

    wrong_type_fields = []

if"needs_visitor_code"in parsed_result andnotisinstance(parsed_result["needs_visitor_code"], bool):

        wrong_type_fields.append("needs_visitor_code")

if missing_fields:

print("缺少字段：", missing_fields)

elif wrong_type_fields:

print("类型不符合预期的字段：", wrong_type_fields)

else:

print("字段齐全，关键类型也符合预期，可以继续给后续系统使用。")

else:

print("当前没有拿到可验证的结构化结果；通常是因为模型或提供商暂不支持对应 method。")

print("\n方法选择建议：")

print("- 想走提供商原生结构化能力：method='json_schema'")

print("- 想兼容更多支持工具调用的模型：method='function_calling'")

print("- 某些旧接口或特定提供商才会用到：method='json_mode'")

print("- 调试时想同时看解析结果和原始消息：include_raw=True")

from langchain_core.messages import AIMessageChunk

system_msg = SystemMessage("你是一个电商客服助手，回答要礼貌、清楚、先给结论。")

human_msg = HumanMessage(

content="我买的咖啡机今天到了，但不会联网，能帮我排查一下吗？",

name="customer_001",

id="msg_human_001",

)

ai_msg = AIMessage(

content="可以，我们先确认 Wi-Fi 是否是 2.4GHz，再检查设备是否进入配网模式。",

usage_metadata={"input_tokens": 28, "output_tokens": 19, "total_tokens": 47},

)

tool_msg = ToolMessage(

content="设备日志查询结果：该咖啡机最近一次连接失败原因是 Wi-Fi 频段不兼容。",

tool_call_id="call_wifi_check_001",

)

sample_messages = [system_msg, human_msg, ai_msg, tool_msg]

print("=== 4 类核心消息对象 ===")

for index, message inenumerate(sample_messages, start=1):

print(f"\n[{index}] {type(message).__name__}")

print("content:", get_text(message))

print("附加字段:")

    pprint(message.model_dump(exclude_none=True))

print("\n=== content_blocks：把不同提供商的内容格式统一起来 ===")

reasoning_message = AIMessage(

content=[

        {"type": "thinking", "thinking": "先判断用户是在问概念还是问代码。", "signature": "demo-signature"},

        {"type": "text", "text": "建议先解释概念，再给最小示例。"},

    ],

response_metadata={"model_provider": "anthropic"},

)

pprint(reasoning_message.content_blocks)

print("\n=== 多模态消息：一条消息里可以同时放文字和图片 ===")

multimodal_message = HumanMessage(

content_blocks=[

        {"type": "text", "text": "请描述这张商品图里最显眼的卖点。"},

        {"type": "image", "url": "https://example.com/product.jpg"},

    ]

)

pprint(multimodal_message.content)

print("\n=== 流式输出的本质：很多小块，最后再拼成完整消息 ===")

chunks = [

    AIMessageChunk(content="今天"),

    AIMessageChunk(content=" 我们"),

    AIMessageChunk(content=" 先把 messages 看懂。"),

]

full_message = None

for chunk in chunks:

    full_message = chunk if full_message isNoneelse full_message + chunk

print("当前累积结果：", get_text(full_message))

text_prompt = "请用一句话解释：为什么 LangChain 需要消息对象？"

message_prompt = [

    SystemMessage("你是一个教学风格非常清晰的老师。"),

    HumanMessage("请用一句话解释：为什么 LangChain 需要消息对象？"),

]

dict_prompt = [

    {"role": "system", "content": "你是一个教学风格非常清晰的老师。"},

    {"role": "user", "content": "请用一句话解释：为什么 LangChain 需要消息对象？"},

]

print("=== 三种常见输入形式 ===")

print("1) 纯字符串：适合单轮、最简单场景")

print(text_prompt)

print("\n2) Message 对象列表：适合多轮对话、系统提示、多模态")

pprint(message_prompt)

print("\n3) 字典格式：适合兼容 OpenAI 风格消息结构")

pprint(dict_prompt)

if model isnotNone:

print("\n=== 用同一个模型分别处理三种输入 ===")

for label, prompt in [

        ("字符串 prompt", text_prompt),

        ("Message 对象 prompt", message_prompt),

        ("字典 prompt", dict_prompt),

    ]:

        response = model.invoke(prompt)

print(f"\n[{label}]")

print(get_text(response))

else:

print("\n未进行真实模型调用；这里只是先把三种输入形式说明清楚。")

print("\n经验建议：")

print("- 只做一个简单问答：直接传字符串就够了")

print("- 需要系统提示、多轮对话、工具结果、多模态：优先用 Message 对象")

print("- 和 OpenAI 风格接口/历史数据打通：字典格式也很方便")