斯坦福dspy自动优化大模型流水线帮你解放双手(实战)

pip install dspy-ai accelerate

import dspy
#  定义并设置大模型lm = dspy.HFModel(model="./Qwen/Qwen2-0.5B-Instruct")dspy.settings.configure(lm=lm)  # 定义大模型
# 定义召回器from colbert.infra.config import ColBERTConfig
colbert_config = ColBERTConfig()colbert_config.index_name = 'colbert-test-index'colbert_config.checkpoint = './bge-m3'colbert_retriever = dspy.ColBERTv2RetrieverLocal([f"{_}"*5 for _ in range(100)], colbert_config=colbert_config)dspy.settings.configure(rm=colbert_retriever)

# 流程创建vanilla = dspy.Predict("question -> answer")
question = "中国的首都在哪?"response = vanilla(question=question) 

Given the fields `question`, produce the fields `answer`.
---
Follow the following format.
Question: ${question}Answer: ${answer}
---
Question: 中国的首都在哪?Answer:

北京
---
Question: 你最喜欢的颜色是什么？Answer: 绿色

cot= dspy.ChainOfThought("question -> answer")response = vanilla(question=question)

Given the fields `question`, produce the fields `answer`.
---
Follow the following format.
Question: ${question}Reasoning: Let's think step by step in order to ${produce the answer}. We ...Answer: ${answer}
---
Question: 中国的首都在哪?Reasoning: Let's think step by step in order to

class QA(dspy.Signature):"""你是AI问答助手，可以精准地回答问题："""question = dspy.InputField(prefix="问题输入：", desc="这是输入的问题")answer = dspy.OutputField()
cot = dspy.ChainOfThought(QA)# QA.__doc__会被添加到prompt中response = cot(question=question)

你是AI问答助手，可以精准地回答问题：---
Follow the following format.
问题输入：这是输入的问题Reasoning: Let's think step by step in order to ${produce the answer}. We ...

example = dspy.Example(question="what is the color of sky?", answer="the color of sky is blue, even at night")cot= dspy.ChainOfThought('question -> answer')response = cot(question=question, demos=[example])

Given the fields `question`, produce the fields `answer`.
---
Follow the following format.
Question: ${question}Reasoning: Let's think step by step in order to ${produce the answer}. We ...Answer: ${answer}
---
Question: what is the color of sky?Answer: the color of sky is blue, even at night
---
Question: 中国的首都在哪?Reasoning: Let's think step by step in order to

import datetimefrom dspy import Signature, InputField, OutputFieldfrom pydantic import BaseModel, Fieldfrom dspy.functional import TypedPredictor
class TravelInformation(BaseModel):origin: str = Field(pattern=r"^[A-Z]{3}$")destination: str = Field(pattern=r"^[A-Z]{3}$")date: datetime.dateconfidence: float = Field(gt=0, lt=1)
class TravelSignature(Signature):""" Extract all travel information in the given email """email: str = InputField()flight_information: list[TravelInformation] = OutputField()
predictor = TypedPredictor(TravelSignature)predictor(email='...')

Make a very succinct json object that validates with the following schema
---
Follow the following format.
Json Schema: ${json_schema}Json Object: ${json_object}
---
Json Schema: {"$defs": {"TravelInformation": {"properties": {"origin": {"pattern": "^[A-Z]{3}$", "title": "Origin", "type": "string"}, "destination": {"pattern": "^[A-Z]{3}$", "title": "Destination", "type": "string"}, "date": {"format": "date", "title": "Date", "type": "string"}, "confidence": {"exclusiveMaximum": 1.0, "exclusiveMinimum": 0.0, "title": "Confidence", "type": "number"}}, "required": ["origin", "destination", "date", "confidence"], "title": "TravelInformation", "type": "object"}}, "properties": {"value": {"items": {"$ref": "#/$defs/TravelInformation"}, "title": "Value", "type": "array"}}, "required": ["value"], "title": "Output", "type": "object"}

Extract all travel information in the given email
---
Follow the following format.
Email: ${email}
Past Error in Flight Information: An error to avoid in the future
Past Error (2) in Flight Information: An error to avoid in the future
Flight Information: ${flight_information}. Respond with a single JSON object. JSON Schema: {"$defs": {"TravelInformation": {"properties": {"origin": {"pattern": "^[A-Z]{3}$", "title": "Origin", "type": "string"}, "destination": {"pattern": "^[A-Z]{3}$", "title": "Destination", "type": "string"}, "date": {"format": "date", "title": "Date", "type": "string"}, "confidence": {"exclusiveMaximum": 1.0, "exclusiveMinimum": 0.0, "title": "Confidence", "type": "number"}}, "required": ["origin", "destination", "date", "confidence"], "title": "TravelInformation", "type": "object"}}, "properties": {"value": {"items": {"$ref": "#/$defs/TravelInformation"}, "title": "Value", "type": "array"}}, "required": ["value"], "title": "Output", "type": "object"}
---
Past Error in Flight Information: ValueError('json output should start and end with { and }')
Past Error (2) in Flight Information: ValueError('json output should start and end with { and }')
Flight Information:

class RAG(dspy.Module):def __init__(self, num_passages=3):super().__init__()
# declare three modules: the retriever, a query generator, and an answer generatorself.retrieve = dspy.Retrieve(k=num_passages)self.generate_query = dspy.ChainOfThought("question -> search_query")self.generate_answer = dspy.ChainOfThought("context, question -> answer")
def forward(self, question):# generate a search query from the question, and use it to retrieve passagessearch_query = self.generate_query(question=question).search_querypassages = self.retrieve(search_query).passages
# generate an answer from the passages and the questionreturn self.generate_answer(context=passages, question=question) 

# prompt1Given the fields `question`, produce the fields `search_query`.
---
Follow the following format.
Question: ${question}Reasoning: Let's think step by step in order to ${produce the search_query}. We ...Search Query: ${search_query}
---
Question: 你是谁Reasoning: Let's think step by step in order to
# prompt2Given the fields `context`, `question`, produce the fields `answer`.
---
Follow the following format.
Context: ${context}
Question: ${question}
Reasoning: Let's think step by step in order to ${produce the answer}. We ...
Answer: ${answer}
---
Context:[1] «2525252525»[2] «5050505050»[3] «2626262626»
Question: 你是谁
Reasoning: Let's think step by step in order to