"question -> answer"
"long-document -> summary"
"context, question -> answer"

class GenerateAnswer(dspy.Signature):"""Answer questions with short factoid answers."""context = dspy.InputField(desc="may contain relevant facts")question = dspy.InputField()answer = dspy.OutputField(desc="often between 1 and 5 words")

# 选项 1，只传递必须的参数generate_answer = dspy.ChainOfThought("context, question -> answer")# 选项 2，将一个定义完整的签名类传递进去generate_answer = dspy.ChainOfThought(GenerateAnswer)
# 调用模块.pred = generate_answer(context = "Which meant learning Lisp, since in those days Lisp was regarded as the language of AI.", question = "What programming language did the author learn in college?")

class RAG(dspy.Module):def __init__(self, num_passages=3):super().__init__()
self.retrieve = dspy.Retrieve(k=num_passages)self.generate_answer = dspy.ChainOfThought(GenerateAnswer)
def forward(self, question):context = self.retrieve(question).passagesprediction = self.generate_answer(context=context, question=question)return dspy.Prediction(context=context, answer=prediction.answer)

from dspy.teleprompt import BootstrapFewShotteleprompter = BootstrapFewShot(metric=dspy.evaluate.answer_exact_match)

from dspy.teleprompt import BootstrapFewShot
# Small training set with question and answer pairstrainset = [dspy.Example(question="What were the two main things the author worked on before college?",  answer="Writing and programming").with_inputs('question'),dspy.Example(question="What kind of writing did the author do before college?",  answer="Short stories").with_inputs('question'),...]
# The teleprompter will bootstrap missing labels: reasoning chains and retrieval contextsteleprompter = BootstrapFewShot(metric=dspy.evaluate.answer_exact_match)compiled_rag = teleprompter.compile(RAG(), trainset=trainset)

pip install dspy-ai

!mkdir -p 'data'!wget 'https://raw.githubusercontent.com/run-llama/llama_index/main/docs/examples/data/paul_graham/paul_graham_essay.txt' -O 'data/paul_graham_essay.txt'

import weaviatefrom weaviate.embedded import EmbeddedOptionsimport re
# 连接到 Weaviate 客户端的embedded模式client = weaviate.Client(embedded_options=EmbeddedOptions(), additional_headers={"X-OpenAI-Api-Key": "sk-", } )
# 创建Weaviate schema# DSPy 假定有一个键：'content'schema = { "classes": [ { "class": "MyExampleIndex", "vectorizer": "text2vec-openai","moduleConfig": {"text2vec-openai": {}}, "properties": [{"name": "content", "dataType": ["text"]}] } ]}client.schema.create(schema)
# 将文档分割成单个句子chunks = []with open("./data/paul_graham_essay.txt", 'r', encoding='utf-8') as file:text = file.read()sentences = re.split(r'(?sentences = [sentence.strip() for sentence in sentences if sentence.strip()]chunks.extend(sentences)
# 使用batch模式处理向量数据集client.batch.configure(batch_size=100) 
with client.batch as batch:  # 初始化一个batch过程    for i, d in enumerate(chunks):  # 导入数据properties = {"content": d,}batch.add_data_object(data_object=properties,class_name="MyExampleIndex")

import dspyimport openaifrom dspy.retrieve.weaviate_rm import WeaviateRM
# Set OpenAI API keyopenai.api_key = "sk-"
# Configure LLMlm = dspy.OpenAI(model="gpt-3.5-turbo")
# Configure Retrieverrm = WeaviateRM("MyExampleIndex", weaviate_client = client)
# Configure DSPy to use the following language model and retrieval model by defaultdspy.settings.configure(lm = lm, rm = rm)

#只需非常少的训练用的问题-答案对trainset = [dspy.Example(question="What were the two main things the author worked on before college?",  answer="Writing and programming").with_inputs('question'),dspy.Example(question="What kind of writing did the author do before college?",  answer="Short stories").with_inputs('question'),dspy.Example(question="What was the first computer language the author learned?",  answer="Fortran").with_inputs('question'),dspy.Example(question="What kind of computer did the author's father buy?",  answer="TRS-80").with_inputs('question'),dspy.Example(question="What was the author's original plan for college?",  answer="Study philosophy").with_inputs('question'),]

class GenerateAnswer(dspy.Signature):"""Answer questions with short factoid answers."""
context = dspy.InputField(desc="may contain relevant facts")question = dspy.InputField()answer = dspy.OutputField(desc="often between 1 and 5 words")

class RAG(dspy.Module):def __init__(self, num_passages=3):super().__init__()
self.retrieve = dspy.Retrieve(k=num_passages)self.generate_answer = dspy.ChainOfThought(GenerateAnswer)def forward(self, question):context = self.retrieve(question).passagesprediction = self.generate_answer(context=context, question=question)return dspy.Prediction(context=context, answer=prediction.answer)

from dspy.teleprompt import BootstrapFewShot# 定义teleprompter = BootstrapFewShot(metric=dspy.evaluate.answer_exact_match)
# 编译compiled_rag = teleprompter.compile(RAG(), trainset=trainset)

pred = compiled_rag(question = "What programming language did the author learn in college?")