工具和框架

掌握LLM开发所需的工具和框架。

框架对比

LangChain生态

LangChain

核心概念

from langchain.schema import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import StrOutputParser
# 1. 消息（Messages）
messages = [
SystemMessage(content="你是一个有帮助的助手。"),
HumanMessage(content="你好！")
]
# 2. 提示词模板（Prompt Templates）
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个{role}。"),
("user", "{input}")
])
# 3. 输出解析器（Output Parsers）
parser = StrOutputParser()
# 4. 链（Chains）
chain = prompt | llm | parser
result = chain.invoke({"role": "Python专家", "input": "解释列表推导式"})

常用组件

1. 链（Chains）

from langchain.chains import LLMChain, SimpleSequentialChain, SequentialChain
# LLM链
prompt = ChatPromptTemplate.from_template("{input}")
llm_chain = LLMChain(llm=llm, prompt=prompt)
# 顺序链
chain1 = LLMChain(llm=llm, prompt=prompt1)
chain2 = LLMChain(llm=llm, prompt=prompt2)
sequential_chain = SequentialChain(
chains=[chain1, chain2],
input_variables=["input"],
output_variables=["output1", "output2"]
)

2. 工具（Tools）

from langchain.tools import Tool
from langchain.agents import initialize_agent, AgentType
def search(query: str) -> str:
"""搜索函数"""
return f"搜索结果: {query}"
search_tool = Tool(
name="Search",
func=search,
description="搜索互联网"
)
# 初始化Agent
agent = initialize_agent(
tools=[search_tool],
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION
)

3. 记忆（Memory）

from langchain.memory import ConversationBufferMemory, ConversationSummaryMemory
# 缓冲记忆
buffer_memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
# 摘要记忆
summary_memory = ConversationSummaryMemory(
llm=llm,
memory_key="chat_history"
)
# 使用
from langchain.chains import ConversationChain
conversation = ConversationChain(
llm=llm,
memory=buffer_memory
)

4. 回调（Callbacks）

from langchain.callbacks import BaseCallbackHandler
class LoggingHandler(BaseCallbackHandler):
"""日志回调"""
def on_llm_start(self, serialized, prompts, **kwargs):
print(f"LLM开始: {prompts}")
def on_llm_end(self, response, **kwargs):
print(f"LLM结束: {response.generations[0][0].text}")
def on_llm_new_token(self, token, **kwargs):
print(f"Token: {token}", end="", flush=True)
# 使用
handler = LoggingHandler()
response = llm.invoke("你好", callbacks=[handler])

LangChain表达式语言（LCEL）

from langchain.schema.runnable import RunnablePassthrough, RunnableLambda
# 管道操作
chain = (
{"prompt": RunnablePassthrough()} |
prompt |
llm |
parser
)
# 分支操作
def branch(input):
if "code" in input.lower():
return llm_code
else:
return llm_text
chain = RunnableLambda(branch)
# 并行操作
from langchain.schema.runnable import RunnableParallel
parallel_chain = RunnableParallel({
"summary": summary_chain,
"answer": answer_chain
})

LlamaIndex

LlamaIndex专注于数据索引和检索。

from llama_index import VectorStoreIndex, SimpleDirectoryReader, ServiceContext
from llama_index.llms import OpenAI
# 加载文档
documents = SimpleDirectoryReader('data').load_data()
# 创建索引
index = VectorStoreIndex.from_documents(documents)
# 查询
query_engine = index.as_query_engine()
response = query_engine.query("什么是RAG？")
print(response)

高级功能

from llama_index import StorageContext, load_index_from_storage
# 持久化索引
index.storage_context.persist()
# 加载索引
storage_context = StorageContext.from_defaults(persist_dir="./storage")
index = load_index_from_storage(storage_context)
# 自定义LLM
from llama_index.llms import OpenAI
service_context = ServiceContext.from_defaults(
llm=OpenAI(model="gpt-4o-mini", temperature=0.7)
)
index = VectorStoreIndex.from_documents(
documents,
service_context=service_context
)

OpenAI SDK

from openai import OpenAI
# 初始化
client = OpenAI(api_key="your-api-key")
# 聊天完成
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "你是一个有帮助的助手。"},
{"role": "user", "content": "你好！"}
],
temperature=0.7,
max_tokens=500
)
print(response.choices[0].message.content)
# 流式输出
stream = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "讲个故事"}],
stream=True
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="", flush=True)
# 函数调用
from openai import pydantic_function_tool
def get_weather(location: str, unit: str = "celsius"):
"""获取天气"""
return f"{location}的天气是25度"
tools = [
pydantic_function_tool(get_weather)
]
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "北京的天气怎么样？"}],
tools=tools
)

向量数据库

ChromaDB

import chromadb
from chromadb.config import Settings
# 初始化
client = chromadb.Client(Settings(
chroma_db_impl="duckdb+parquet",
persist_directory="./chroma_db"
))
# 创建集合
collection = client.create_collection(name="documents")
# 添加文档
collection.add(
documents=["文档1内容", "文档2内容"],
metadatas=[{"source": "file1.txt"}, {"source": "file2.txt"}],
ids=["doc1", "doc2"]
)
# 查询
results = collection.query(
query_texts=["查询文本"],
n_results=2
)

Pinecone

import pinecone
# 初始化
pinecone.init(
api_key="your-api-key",
environment="us-east-1-aws"
)
# 创建索引
if "my-index" not in pinecone.list_indexes():
pinecone.create_index(
name="my-index",
dimension=1536,
metric="cosine"
)
# 连接索引
index = pinecone.Index("my-index")
# 添加向量
index.upsert([
("vec1", [0.1, 0.2, ...], {"source": "file1"}),
("vec2", [0.3, 0.4, ...], {"source": "file2"})
])
# 查询
results = index.query(
vector=[0.1, 0.2, ...],
top_k=5,
include_metadata=True
)

FAISS

import faiss
import numpy as np
# 创建索引
dimension = 768
index = faiss.IndexFlatL2(dimension)
# 添加向量
vectors = np.random.random((100, dimension)).astype('float32')
index.add(vectors)
# 查询
query_vector = np.random.random((1, dimension)).astype('float32')
distances, indices = index.search(query_vector, k=5)

开发工具

Jupyter Notebook

# 安装
pip install jupyter
# 启动
jupyter notebook
# 或使用JupyterLab
jupyter lab

VS Code扩展

推荐扩展： - Python - Python语言支持 - Jupyter - Notebook支持 - CodeLLM - AI代码补全 - Pylance - Python类型检查 - GitLens - Git增强

Streamlit

import streamlit as st
st.title("我的应用")
# 侧边栏
with st.sidebar:
st.header("设置")
option = st.selectbox("选择", ["选项1", "选项2"])
# 主要内容
st.write("Hello, world!")
# 交互式组件
text = st.text_input("输入文本")
button = st.button("点击我")
if button:
st.success(f"你输入了: {text}")

监控和调试

LangSmith

from langsmith import Client
# 初始化
client = Client(api_key="your-langsmith-key")
# 追踪运行
run = client.create_run(
name="my-chain",
inputs={"input": "test"}
)
# 记录事件
client.create_event(
run_id=run.id,
event_type="llm",
inputs={"prompt": "hello"},
outputs={"output": "world"}
)
# 结束运行
client.end_run(run_id=run.id, outputs={"result": "success"})

Weights & Biases

import wandb
# 初始化
wandb.init(project="llm-experiments")
# 记录指标
wandb.log({
"loss": 0.5,
"accuracy": 0.9,
"tokens": 1000
})
# 结束
wandb.finish()

学习要点

✅ LangChain提供完整的LLM应用框架 ✅ LlamaIndex专注于数据索引和检索 ✅ OpenAI SDK官方支持最完善 ✅ ChromaDB简单易用，Pinecone适合生产 ✅ VS Code和Jupyter是主要开发工具 ✅ LangSmith和W&B帮助监控和调试

下一步: 查看学习资源](02-learning-resources.md) 📖