LangChain 快速入门

LangChain 快速入门

"""LangChain lessons playground.

Each lesson is a standalone function so you can run and maintain them independently.
Set `LESSON` at the bottom to choose which lesson to run.
"""

from __future__ import annotations

from typing import Any, Callable

LESSON_MODEL = "deepseek:deepseek-chat"
LESSON = 2


def _init_model() -> Any:
	"""Create a chat model instance for lessons."""
	from langchain.chat_models import init_chat_model

	return init_chat_model(LESSON_MODEL)


def _print_agent_messages(result: dict[str, Any]) -> None:
	"""Print all messages in an agent result payload."""
	for message in result["messages"]:
		print(f"{type(message).__name__}: {message.content}")


# 第 1 关：召唤模型

def lesson_1_invoke_model() -> None:
	"""Invoke a chat model with a single string prompt."""
	response = _init_model().invoke("请用一句话解释什么是 LangChain。")
	print("类型:", type(response).__name__)
	print("内容:", response.content)
	print("响应元数据:", response.response_metadata)


# 第 2 关：消息与 Prompt

def lesson_2_messages_and_prompt() -> None:
	"""Send explicit system/human messages to the model."""
	from langchain_core.messages import HumanMessage, SystemMessage

	messages = [
		SystemMessage(content="你是一位只用比喻讲解技术的老师。"),
		HumanMessage(content="解释什么是列表推导式。"),
	]
	response = _init_model().invoke(messages)
	print(response.content)
	print("关键点: SystemMessage 定义角色，HumanMessage 表达用户问题。")


# 第 3 关：Runnable 数据管道

def lesson_3_runnable_pipeline() -> None:
	"""Compose prompt -> model -> parser as one runnable pipeline."""
	from langchain_core.output_parsers import StrOutputParser
	from langchain_core.prompts import ChatPromptTemplate

	prompt = ChatPromptTemplate.from_messages([
		("system", "你是一位简洁的技术老师。"),
		("human", "请用三句话解释：{topic}"),
	])
	chain = prompt | _init_model() | StrOutputParser()
	result = chain.invoke({"topic": "Runnable"})

	print(result)
	print("链类型:", type(chain).__name__)


# 第 4 关：让模型调用工具

def lesson_4_tool_calling() -> None:
	"""Build an agent with one tool and let model decide when to call it."""
	from langchain.agents import create_agent
	from langchain_core.tools import tool

	@tool
	def get_temperature(city: str) -> str:
		"""查询城市温度。"""
		temperatures = {
			"北京": "25 摄氏度",
			"上海": "28 摄氏度",
		}
		return temperatures.get(city, "未知城市")

	agent = create_agent(model=_init_model(), tools=[get_temperature])
	result = agent.invoke({
		"messages": [{"role": "user", "content": "北京现在多少度？"}],
	})

	print(">>> result:", result)
	print("=" * 50)
	_print_agent_messages(result)


# 第 5 关：理解 Agent 核心循环

def lesson_5_agent_loop() -> None:
	"""Observe model -> tool -> model loop with a safe calculator tool."""
	import ast
	import operator

	from langchain.agents import create_agent
	from langchain_core.tools import tool

	binary_ops = {
		ast.Add: operator.add,
		ast.Sub: operator.sub,
		ast.Mult: operator.mul,
		ast.Div: operator.truediv,
		ast.Mod: operator.mod,
		ast.Pow: operator.pow,
	}
	unary_ops = {
		ast.UAdd: operator.pos,
		ast.USub: operator.neg,
	}

	def safe_eval(expression: str) -> float:
		"""Evaluate arithmetic expressions with AST instead of `eval`."""
		node = ast.parse(expression, mode="eval")

		def visit(current: ast.AST) -> float:
			if isinstance(current, ast.Expression):
				return visit(current.body)
			if isinstance(current, ast.Constant) and isinstance(current.value, (int, float)):
				return float(current.value)
			if isinstance(current, ast.BinOp) and type(current.op) in binary_ops:
				left = visit(current.left)
				right = visit(current.right)
				return float(binary_ops[type(current.op)](left, right))
			if isinstance(current, ast.UnaryOp) and type(current.op) in unary_ops:
				return float(unary_ops[type(current.op)](visit(current.operand)))
			raise ValueError("不支持的表达式")

		return visit(node)

	@tool
	def calculator(expression: str) -> str:
		"""计算一个简单数学表达式。"""
		try:
			value = safe_eval(expression)
		except Exception as exc:  # noqa: BLE001
			return f"计算失败: {exc}"
		if value.is_integer():
			return str(int(value))
		return str(value)

	agent = create_agent(model=_init_model(), tools=[calculator])
	result = agent.invoke({
		"messages": [{"role": "user", "content": "计算 12 * (8 + 3)"}],
	})

	print(">>> result:", result)
	print("最终回答:", result["messages"][-1].content)
	print("关键点: Agent 会在模型和工具之间循环，直到得到最终回答。")


# 第 6 关：流式输出

def lesson_6_streaming() -> None:
	"""Stream model tokens and agent updates."""
	from langchain.agents import create_agent

	model = _init_model()
	print("模型流式输出:")
	for chunk in model.stream("写一首关于程序员的短诗"):
		if chunk.content:
			print(chunk.content, end="", flush=True)
	print("\n")

	agent = create_agent(model=model, tools=[])
	print("Agent 流式更新:")
	for chunk in agent.stream({
		"messages": [{"role": "user", "content": "解释 Python 的装饰器"}],
	}, stream_mode="updates"):
		print(chunk)


# 第 7 关：记忆与持久化

def lesson_7_memory() -> None:
	"""Use a checkpointer to preserve messages across two agent runs."""
	from langchain.agents import create_agent
	from langgraph.checkpoint.memory import InMemorySaver

	agent = create_agent(model=_init_model(), checkpointer=InMemorySaver())
	config = {"configurable": {"thread_id": "lesson-7"}}

	first = agent.invoke(
		{"messages": [{"role": "user", "content": "我叫张三，请记住我的名字。"}]},
		config,
	)
	second = agent.invoke(
		{"messages": [{"role": "user", "content": "我叫什么？"}]},
		config,
	)

	print("第一次:", first["messages"][-1].content)
	print("第二次:", second["messages"][-1].content)
	print("关键点: 两次调用使用了相同的 thread_id。")


# 第 8 关：RAG 知识库

def lesson_8_rag() -> None:
	"""Retrieve local documents, then place them into the model prompt."""
	from langchain_core.documents import Document
	from langchain_core.prompts import ChatPromptTemplate

	documents = [
		Document(page_content="LangChain 用 Runnable 统一组件的执行接口。"),
		Document(page_content="Agent 通常由模型节点、工具节点和条件边组成。"),
		Document(page_content="ToolNode 负责执行模型返回的 tool_calls。"),
	]
	question = "Agent 的工具调用由谁执行？"
	query_words = ("Agent", "工具", "执行", "ToolNode", "tool_calls")
	retrieved = [
		document
		for document in documents
		if any(word in document.page_content for word in query_words)
	]
	context = "\n".join(document.page_content for document in retrieved)

	prompt = ChatPromptTemplate.from_messages([
		("system", "只依据下面的上下文回答，不确定时明确说不知道。\n上下文:\n{context}"),
		("human", "问题: {question}"),
	])
	chain = prompt | _init_model()
	response = chain.invoke({"context": context, "question": question})

	print("检索结果:", context)
	print("回答:", response.content)
	print("关键点: RAG = 检索 documents + 将 context 注入 prompt。")


# 第 9 关：Middleware 控制 Agent

def lesson_9_middleware() -> None:
	"""Limit an agent to one model call and observe middleware behavior."""
	from langchain.agents import create_agent
	from langchain.agents.middleware import ModelCallLimitMiddleware

	agent = create_agent(
		model=_init_model(),
		middleware=[ModelCallLimitMiddleware(run_limit=1, exit_behavior="end")],
	)
	result = agent.invoke({
		"messages": [{"role": "user", "content": "解释 Middleware 的作用。"}],
	})
	print(">>> result:", result)
	print("=" * 27)
	print(result["messages"][-1].content)
	print("关键点: Middleware 可以在模型节点前后控制执行过程。")


# 第 10 关：综合挑战

def lesson_10_capstone() -> None:
	"""Build a small DeepSeek agent with two deterministic local tools."""
	from datetime import datetime

	from langchain.agents import create_agent
	from langchain_core.tools import tool

	@tool
	def get_project_fact(topic: str) -> str:
		"""查询实验项目中的固定知识。"""
		facts = {
			"runnable": "Runnable 统一了 invoke、stream 和 batch 等执行接口。",
			"agent": "Agent 通过 model -> tools -> model 循环处理工具调用。",
		}
		return facts.get(topic.lower(), "没有找到这个知识点。")

	@tool
	def current_time() -> str:
		"""返回本机当前时间。"""
		return datetime.now().astimezone().isoformat(timespec="seconds")

	agent = create_agent(model=_init_model(), tools=[get_project_fact, current_time])
	for update in agent.stream({
		"messages": [{
			"role": "user",
			"content": "先告诉我 Agent 的核心循环，再告诉我现在的本机时间。",
		}],
	}, stream_mode="updates"):
		print(update)


def run_lesson(lesson: int) -> None:
	"""Run one lesson by lesson id."""
	lessons: dict[int, Callable[[], None]] = {
		1: lesson_1_invoke_model,
		2: lesson_2_messages_and_prompt,
		3: lesson_3_runnable_pipeline,
		4: lesson_4_tool_calling,
		5: lesson_5_agent_loop,
		6: lesson_6_streaming,
		7: lesson_7_memory,
		8: lesson_8_rag,
		9: lesson_9_middleware,
		10: lesson_10_capstone,
	}
	if lesson not in lessons:
		available = ", ".join(str(value) for value in sorted(lessons))
		raise ValueError(f"未知 LESSON={lesson}，可选值: {available}")
	lessons[lesson]()


if __name__ == "__main__":
	run_lesson(LESSON)