第 12 章：集成与第三方
出处
来源：third_party/ · 状态：已读
一句话
第三方是适配器，证明边界，不规定默认栈。 向量库、索引框架、百科、计算引擎、语音，接的都是第 2、3、6 章已经钉死的口：检索、工具、模态进出。2024 年的 LangChain / LlamaIndex 示例尤其不能当成今天的标准答案。
核心概念
向量库：嵌入 → 索引 → 检索 → 生成
Pinecone/rag_using_pinecone.ipynb：Voyage 嵌 Amazon 商品描述 → Pinecone（voyage-2，1024 维）→ 自然语言问。优化不是换库，是 让 Claude 先把问题展开成一组检索关键词（JSON 吐出），每词查 top-3 再合成。这是第 2 章「查询改写」的厂商皮肤。
Pinecone/claude_3_rag_agent.ipynb：LangChain v1 + Voyage + Pinecone，语料是预切的 arXiv。强调 Anthropic 吃 XML agent（<tool> / <tool_input>），不是典型 ReAct 文本。记忆不能把 HumanMessage 直接塞回去，要收成 Human: / AI: 字符串。版本写明是 LangChain v1 相对 0.0.3xx 的迁移——读的是模式，不是抄依赖号。
MongoDB/rag_using_mongodb.ipynb：另一套存储（Atlas vector index + Voyage），角色是 VC tech analyst，语料 Hacker Noon。机制相同：索引是检索后端，Claude 是生成器。
LlamaIndex：索引框架里的工作流图
一组 notebook 把第 2、5 章的图用框架再画一遍：
| 路径 | 对应机制 |
|---|---|
Basic_RAG_With_LlamaIndex | naive RAG |
Router_Query_Engine | 第 5 章 routing |
SubQuestion_Query_Engine | 拆子问题（轻量 orchestrator） |
ReAct_Agent | 框架内的工具环；Anthropic 更吃 XML |
Multi_Document_Agents | 多文档职责分区 |
Multi_Modal | 第 6 章在框架里的入口 |
框架替你管的是 retriever 装配和 query engine 拓扑。不替你管：golden chunk 评测、cache、权限、session 预算。那些仍在第 2、4、10、11 章。
嵌入、百科、计算、语音
VoyageAI/how_to_create_embeddings.md：嵌入模型另选。第 2 章 RAG / contextual 默认 Voyage，这里是说明书。换嵌入 = 换向量空间，索引要重建。Wikipedia/wikipedia-search-cookbook.ipynb：迭代搜百科——工具环 + 检索，不是一次 wiki dump。WolframAlpha/using_llm_api.ipynb：计算型工具。和第 3 章计算器同一原则：模型不算，引擎算。- Deepgram / ElevenLabs：见第 6 章。STT/TTS 是进出适配器；延迟和账单在厂商侧，Claude 仍吃文本。
对照 recipe（默认折叠；机制片段摘自官方 notebook，全文在 GitHub）：
third_party/Pinecone/rag_using_pinecone.ipynb — 向量库当检索后端；查询改写
python
import anthropic
client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)
def get_completion(prompt):
completion = client.completions.create(
model="claude-2.1",
prompt=prompt,
max_tokens_to_sample=1024,
)
return completion.completion
def create_keyword_prompt(question):
return f"""\n\nHuman: Given a question, generate a list of 5 very diverse search keywords that can be used to search for products on Amazon.
The question is: {question}
Output your keywords as a JSON that has one property "keywords" that is a list of strings. Only output valid JSON.\n\nAssistant:{{"""
keyword_json = "{" + get_completion(create_keyword_prompt(USER_QUESTION))
print(keyword_json)third_party/Pinecone/claude_3_rag_agent.ipynb — RAG + XML agent（LangChain v1，注意版本）
python
import time
index_name = "claude-3-rag"
# check if index already exists (it shouldn't if this is first time)
if index_name not in pc.list_indexes().names():
# if does not exist, create index
pc.create_index(
index_name,
dimension=len(vec[0]), # dimensionality of voyage model
metric="dotproduct",
spec=spec,
)
# wait for index to be initialized
while not pc.describe_index(index_name).status["ready"]:
time.sleep(1)
# connect to index
index = pc.Index(index_name)
time.sleep(1)
# view index stats
index.describe_index_stats()
from langchain.agents import tool
@tool
def arxiv_search(query: str) -> str:
"""Use this tool when answering questions about AI, machine learning, data
science, or other technical questions that may be answered using arXiv
papers.
"""
# create query vector
xq = embed.embed_query(query)
# perform search
out = index.query(vector=xq, top_k=5, include_metadata=True)
# reformat results into string
results_str = "\n\n".join([x["metadata"]["text"] for x in out["matches"]])
return results_str
tools = [arxiv_search]third_party/MongoDB/rag_using_mongodb.ipynb — 另一套存储，同一管道
python
import voyageai
vo = voyageai.Client(api_key=userdata.get("VOYAGE_API_KEY"))
def get_embedding(text: str) -> list[float]:
if not text.strip():
print("Attempted to get embedding for empty text.")
return []
embedding = vo.embed(text, model="voyage-large-2", input_type="document")
return embedding.embeddings[0]
combined_df["embedding"] = combined_df["description"].apply(get_embedding)
combined_df.head()
def handle_user_query(query, collection):
get_knowledge = vector_search(query, collection)
search_result = ""
for result in get_knowledge:
search_result += (
f"Title: {result.get('title', 'N/A')}, "
f"Company Name: {result.get('companyName', 'N/A')}, "
f"Company URL: {result.get('companyUrl', 'N/A')}, "
f"Date Published: {result.get('published_at', 'N/A')}, "
f"Article URL: {result.get('url', 'N/A')}, "
f"Description: {result.get('description', 'N/A')}, \n"
)
response = client.messages.create(
model="claude-opus-4-8",
max_tokens=1024,
system="You are Venture Captital Tech Analyst with access to some tech company articles and information. You use the information you are given to provide advice.",
messages=[
{
"role": "user",
"content": "Answer this user query: "
+ query
+ " with the following context: "
+ search_result,
}
],
)
return (response.content[0].text), search_resultthird_party/LlamaIndex/Basic_RAG_With_LlamaIndex.ipynb — 索引框架
python
llm = Anthropic(temperature=0.0, model="claude-opus-4-8")
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-base-en-v1.5")
index = VectorStoreIndex.from_documents(
documents,
)
response = query_engine.query("What did author do growing up?")third_party/LlamaIndex/Router_Query_Engine.ipynb — 路由查询
python
# Summary Index Query Engine
summary_query_engine = summary_index.as_query_engine(
response_mode="tree_summarize",
use_async=True,
)
# Vector Index Query Engine
vector_query_engine = vector_index.as_query_engine()
from llama_index.core.tools.query_engine import QueryEngineTool
# Summary Index tool
summary_tool = QueryEngineTool.from_defaults(
query_engine=summary_query_engine,
description="Useful for summarization questions related to Paul Graham eassy on What I Worked On.",
)
# Vector Index tool
vector_tool = QueryEngineTool.from_defaults(
query_engine=vector_query_engine,
description="Useful for retrieving specific context from Paul Graham essay on What I Worked On.",
)
# Create Router Query Engine
query_engine = RouterQueryEngine(
selector=LLMSingleSelector.from_defaults(),
query_engine_tools=[
summary_tool,
vector_tool,
],
)third_party/LlamaIndex/SubQuestion_Query_Engine.ipynb — 子问题拆解
python
llm = Anthropic(temperature=0.0, model="claude-opus-4-8")
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-base-en-v1.5")
from llama_index.core.query_engine import SubQuestionQueryEngine
from llama_index.core.tools import QueryEngineTool, ToolMetadata
query_engine_tools = [
QueryEngineTool(
query_engine=lyft_engine,
metadata=ToolMetadata(
name="lyft_10k", description="Provides information about Lyft financials for year 2021"
),
),
QueryEngineTool(
query_engine=uber_engine,
metadata=ToolMetadata(
name="uber_10k", description="Provides information about Uber financials for year 2021"
),
),
]
response = await sub_question_query_engine.aquery("Compare the investments made by Uber and Lyft")third_party/LlamaIndex/ReAct_Agent.ipynb — ReAct 在框架里的形状
python
llm = Anthropic(temperature=0.0, model="claude-opus-4-8")
embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-base-en-v1.5")
lyft_index = VectorStoreIndex.from_documents(lyft_docs)
uber_index = VectorStoreIndex.from_documents(uber_docs)
query_engine_tools = [
QueryEngineTool(
query_engine=lyft_engine,
metadata=ToolMetadata(
name="lyft_10k",
description=(
"Provides information about Lyft financials for year 2021. "
"Use a detailed plain text question as input to the tool."
),
),
),
QueryEngineTool(
query_engine=uber_engine,
metadata=ToolMetadata(
name="uber_10k",
description=(
"Provides information about Uber financials for year 2021. "
"Use a detailed plain text question as input to the tool."
),
),
),
]third_party/LlamaIndex/Multi_Document_Agents.ipynb — 多文档 agent
python
from llama_index.core import SummaryIndex, VectorStoreIndex
from llama_index.core.agent import ReActAgent
from llama_index.core.tools import QueryEngineTool, ToolMetadata
# Build agents dictionary
agents = {}
for wiki_title in wiki_titles:
# build vector index
vector_index = VectorStoreIndex.from_documents(
city_docs[wiki_title],
)
# build summary index
summary_index = SummaryIndex.from_documents(
city_docs[wiki_title],
)
# define query engines
vector_query_engine = vector_index.as_query_engine()
summary_query_engine = summary_index.as_query_engine()
# define tools
query_engine_tools = [
QueryEngineTool(
query_engine=vector_query_engine,
metadata=ToolMetadata(
name="vector_tool",
description=(f"Useful for retrieving specific context from {wiki_title}"),
),
),
QueryEngineTool(
query_engine=summary_query_engine,
metadata=ToolMetadata(
name="summary_tool",
description=(f"Useful for summarization questions related to {wiki_title}"),
),
),
]
# build agent
agent = ReActAgent.from_tools(
query_engine_tools,
llm=llm,
verbose=True,
)
agents[wiki_title] = agentthird_party/LlamaIndex/Multi_Modal.ipynb — 框架里的多模态
python
import matplotlib.pyplot as plt
from PIL import Image
img = Image.open("./prometheus_paper_card.png")
plt.imshow(img)third_party/VoyageAI/how_to_create_embeddings.md — 嵌入模型另选
markdown
# Embeddings
Text embeddings are numerical representations of text strings, represented as a vector of floating point numbers. We can use the distance between two text embeddings (popularly cosine similarity) to measure how related two pieces of text are to one another, with smaller distances predicting higher relatedness.
Comparing the similarity of strings, or clustering strings by their distance from one another, allows for a wide variety of applications including **search** (popular in RAG architectures), **recommendations**, and **anomaly detection**.
## How to get embeddings with Anthropic
While Anthropic does not offer its own embedding model, we have partnered with [Voyage AI](https://www.voyageai.com/?ref=anthropic) as our preferred provider for text embeddings. Voyage makes [state of the art](https://blog.voyageai.com/2023/10/29/voyage-embeddings/?ref=anthropic) embedding models, and even offers models customized for specific industry domains such as finance and healthcare, and models that can be fine-tuned for your company.
To access Voyage embeddings, please first sign up on [Voyage AI’s website](https://dash.voyageai.com/?ref=anthropic), obtain an API key, and set the API key as an environment variable for convenience:
'''bash
export VOYAGE_API_KEY="<your secret key>"
'''
You can obtain the embeddings either using the official [`voyageai` Python package](https://github.com/voyage-ai/voyageai-python) or HTTP requests, as described below.
### Voyage Python Package
The `voyageai` package can be installed using the following command:
'''bash
pip install -U voyageai
'''
Then, you can create a client object and start using it to embed your texts:
'''python
import voyageai
vo = voyageai.Client()
# This will automatically use the environment variable VOYAGE_API_KEY.
# Alternatively, you can use vo = voyageai.Client(api_key="<your secret key>")
texts = ["Sample text 1", "Sample text 2"]
result = vo.embed(texts, model="voyage-2", input_type="document")
print(result.embeddings[0])
print(result.embeddings[1])
'''
`result.embeddings` will be a list of two embedding vectors, each containing 1024 floating-point numbers. After running the above code, the two embeddings will be printed on the screen:
'''
[0.02012746, 0.01957859, ...] # embedding for "Sample text 1"
[0.01429677, 0.03077182, ...] # embedding for "Sample text 2"
'''
When creating the embeddings, you may specify a few other arguments to the `embed()` function. Here is the specification:
> `voyageai.Client.embed(texts : List[str], model : str = "voyage-2", input_type : Optional[str] = None, truncation : Optional[bool] = None)`
- **texts** (List[str]) - A list of texts as a list of strings, such as `["I like cats", "I also like dogs"]`. Currently, the maximum length of the list is 128, and total number of tokens in the list is at most 320K for `voyage-2` and 120K for `voyage-code-2`.
- **model** (str) - Name of the model. Recommended options: `voyage-2` (default), `voyage-code-2`.
- **input_type** (str, optional, defaults to `None`) - Type of the input text. Defalut to `None`. Other options: `query`, `document`.
- When the input_type is set to `None`, and the input text will be directly encoded by our embedding model. Alternatively, when the inputs are documents or queries, the users can specify input_type to be `query` or `document`, respectively. In such cases, Voyage will prepend a special prompt to input text and send the extended inputs to the embedding model.
- For retrieval/search use cases, we recommend specifying this argument when encoding queries or documents to enhance retrieval quality. Embeddings generated with and without the input_type argument are compatible.
- **truncation** (bool, optional, defaults to `None`) - Whether to truncate the input texts to fit within the context length.
- If `True`, over-length input texts will be truncated to fit within the context length, before vectorized by the embedding model.
- If `False`, an error will be raised if any given text exceeds the context length.
- If not specified (defaults to `None`), Voyage will truncate the input text before sending it to the embedding model if it slightly exceeds the context window length. If it significantly exceeds the context window length, an error will be raised.
### Voyage HTTP API
You can also get embeddings by requesting Voyage HTTP API. For example, you can send an HTTP request through the `curl` command in a terminal:
'''bash
curl https://api.voyageai.com/v1/embeddings \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $VOYAGE_API_KEY" \
-d '{
"input": ["Sample text 1", "Sample text 2"],
"model": "voyage-2"
}'
'''
The response you would get is a JSON object containing the embeddings and the token usage:
'''bash
{
# …third_party/Wikipedia/wikipedia-search-cookbook.ipynb — 迭代搜百科
python
def extract_between_tags(tag: str, string: str, strip: bool = True) -> list[str]:
ext_list = re.findall(rf"<{tag}\s?>(.+?)</{tag}\s?>", string, re.DOTALL)
if strip:
ext_list = [e.strip() for e in ext_list]
return ext_list
class ClientWithRetrieval(Anthropic):
def __init__(self, search_tool: SearchTool, verbose: bool = True, *args, **kwargs):
super().__init__(*args, **kwargs)
self.search_tool = search_tool
self.verbose = verbose
# Helper methods
def _search_query_stop(
self, partial_completion: str, n_search_results_to_use: int
) -> tuple[list[SearchResult], str]:
search_query = extract_between_tags("search_query", partial_completion + "</search_query>")
if search_query is None:
raise Exception(
"Completion with retrieval failed as partial completion returned mismatched <search_query> tags."
)
print(f"Running search query against SearchTool: {search_query}")
search_results = self.search_tool.raw_search(search_query, n_search_results_to_use)
extracted_search_results = self.search_tool.process_raw_search_results(search_results)
formatted_search_results = self.search_tool.wrap_search_results(extracted_search_results)
return search_results, formatted_search_results
def retrieve(
self,
query: str,
model: str,
n_search_results_to_use: int = 3,
stop_sequences: list[str] = None,
max_tokens_to_sample: int = 1000,
max_searches_to_try: int = 5,
temperature: float = 1.0,
) -> tuple[list[SearchResult], str]:
if stop_sequences is None:
stop_sequences = [HUMAN_PROMPT]
prompt = (
f"{HUMAN_PROMPT} {wikipedia_prompt} {retrieval_prompt.format(query=query)}{AI_PROMPT}"
)
starting_prompt = prompt
print("Starting prompt:", starting_prompt)
token_budget = max_tokens_to_sample
all_raw_search_results: list[SearchResult] = []
for tries in range(max_searches_to_try):
partial_completion = self.completions.create(
prompt=prompt,
stop_sequences=stop_sequences + ["</search_query>"],
model=model,
max_tokens_to_sample=token_budget,
temperature=temperature,
)
partial_completion, stop_reason, stop_seq = (
partial_completion.completion,
partial_completion.stop_reason,
partial_completion.stop,
)
print(partial_completion)
token_budget -= self.count_tokens(partial_completion)
prompt += partial_completion
if stop_reason == "stop_sequence" and stop_seq == "</search_query>":
print(f"Attempting search number {tries}.")
raw_search_results, formatted_search_results = self._search_query_stop(
partial_completion, n_search_results_to_use
)
prompt += "</search_query>" + formatted_search_results
all_raw_search_results += raw_search_results
else:
break
final_model_response = prompt[len(starting_prompt) :]
return all_raw_search_results, final_model_response
# Main methods
def completion_with_retrieval(
self,
query: str,
model: str,
n_search_results_to_use: int = 3,
stop_sequences: list[str] = None,
max_tokens_to_sample: int = 1000,
max_searches_to_try: int = 5,
temperature: float = 1.0,
) -> str:
if stop_sequences is None:
stop_sequences = [HUMAN_PROMPT]
_, retrieval_response = self.retrieve(
query,
model=model,
n_search_results_to_use=n_search_results_to_use,
stop_sequences=stop_sequences,
max_tokens_to_sample=max_tokens_to_sample,
max_searches_to_try=max_searches_to_try,
temperature=temperature,
)
information = extract_between_tags("information", retrieval_response)[-1]
prompt = f"{HUMAN_PROMPT} {answer_prompt.format(query=query, information=information)}{AI_PROMPT}"
print("Summarizing:\n", prompt)
answer = self.completions.create(
prompt=prompt, model=model, temperature=temperature, max_tokens_to_sample=1000
).completion
return answerthird_party/WolframAlpha/using_llm_api.ipynb — 计算型工具
python
def process_tool_call(tool_name, tool_input):
if tool_name == "wolfram_alpha":
return wolfram_alpha_query(tool_input["search_query"])
def chat_with_claude(user_message):
print(f"\n{'=' * 50}\nUser Message: {user_message}\n{'=' * 50}")
prompt = f"""Here is a question: {user_message}. Please use the Wolfram Alpha tool to answer it. Do not reflect on the quality of the returned search results in your response."""
message = client.beta.tools.messages.create(
model=MODEL_NAME,
max_tokens=4096,
tools=tools,
messages=[{"role": "user", "content": prompt}],
)
print("\nInitial Response:")
print(f"Stop Reason: {message.stop_reason}")
print(f"Content: {message.content}")
if message.stop_reason == "tool_use":
tool_use = next(block for block in message.content if block.type == "tool_use")
tool_name = tool_use.name
tool_input = tool_use.input
print(f"\nTool Used: {tool_name}")
print("Tool Input:")
print(json.dumps(tool_input, indent=2))
tool_result = process_tool_call(tool_name, tool_input)
print("\nTool Result:")
print(str(json.dumps(tool_result, indent=2)))
response = client.beta.tools.messages.create(
model=MODEL_NAME,
max_tokens=2000,
tools=tools,
messages=[
{"role": "user", "content": prompt},
{"role": "assistant", "content": message.content},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": tool_use.id,
"content": str(tool_result),
}
],
},
],
)
print("\nResponse:")
print(f"Stop Reason: {response.stop_reason}")
print(f"Content: {response.content}")
else:
response = message
final_response = None
for block in response.content:
if hasattr(block, "text"):
final_response = block.text
break
print(f"\nFinal Response: {final_response}")
return final_responsethird_party/Deepgram/prerecorded_audio.ipynb — 见第 6 章
python
import json
import anthropic
transcription_file = "transcript.json"
# Function to get the transcript from the JSON file
def get_transcript(transcription_file):
with open(transcription_file) as file:
data = json.load(file)
result = data["results"]["channels"][0]["alternatives"][0]["transcript"]
return result
# Load the transcript from the JSON file
message_text = get_transcript(transcription_file)
# Initialize the Claude API client
client = anthropic.Anthropic(
# Defaults to os.environ.get("ANTHROPIC_API_KEY")
# Claude API key
api_key="🔑🔑🔑 Your API Key here! 🔑🔑🔑"
)
# Prepare the text for the API request
formatted_messages = [{"role": "user", "content": message_text}]
# Generate thoughtful, open-ended interview questions
response = client.messages.create(
model="claude-opus-4-8",
max_tokens=1000,
temperature=0.5,
system="Your task is to generate a series of thoughtful, open-ended questions for an interview based on the given context. The questions should be designed to elicit insightful and detailed responses from the interviewee, allowing them to showcase their knowledge, experience, and critical thinking skills. Avoid yes/no questions or those with obvious answers. Instead, focus on questions that encourage reflection, self-assessment, and the sharing of specific examples or anecdotes.",
messages=formatted_messages,
)
# Print the generated questions
# Join the text of each TextBlock into a single string
content = "".join(block.text for block in response.content)
# Split the content by '\n\n'
parts = content.split("\n\n")
# Print each part with an additional line break
for part in parts:
print(part)
print("\n")third_party/ElevenLabs/low_latency_stt_claude_tts.ipynb — 见第 6 章
python
start_time = time.time()
message = anthropic_client.messages.create(
model="claude-haiku-4-5",
max_tokens=1000,
temperature=0,
messages=[{"role": "user", "content": transcription.text}],
)
end_time = time.time()
non_streaming_response_time = end_time - start_time
print(message.content[0].text)
print(f"\nResponse time: {non_streaming_response_time:.2f} seconds")关键洞察
读这一章时盯接口，不盯 logo。能替换的是：向量库、嵌入、STT、TTS、索引框架。不能替换的是：输出契约、工具权限、上下文预算、评测和熔断。LangChain XML agent 那份甚至在提醒：框架默认的 ReAct 字符串对 Claude 不是最优工具协议。
厂商 notebook 会过期。Pinecone 维数、LangChain import、LlamaIndex query engine API 都会变。把「Voyage 1024 维 + 关键词展开」写进自己的系统时，要回到官方文档核对，不要冻结 cookbook 的 SDK 版本。
个人思考
12 章走完，依赖顺序是闭环的：第 1 章形状 → 2 检索 / 3 工具 / 4 预算 → 5 图 → 6 模态 / 7 技能 → 8 进程内循环 / 9 服务端循环 → 10 尺子 / 11 账单 → 12 把外部系统接到已经钉死的口上。跳到本章抄 LlamaIndex，等于从文件夹中间抽了一张卡片。