import dspy # 创建文本分类器 classifier = dspy.Predict("text -> sentiment: str") # 使用分类器 text = "这部电影真的太棒了！" result = classifier(text=text) print(f"情感: {result.sentiment}")

import dspy # 创建推理模块 reasoner = dspy.ChainOfThought("question -> answer: str") # 解决复杂问题 question = "如果所有的鸟都会飞，企鹅是鸟，那么企鹅会飞吗？" result = reasoner(question=question) print(f"答案: {result.answer}") print(f"推理过程: {result.reasoning}")

import dspy # 定义工具函数 def search(query: str) -> str: # 模拟搜索功能 return f"搜索结果: {query} 的相关信息" # 创建ReAct模块 agent = dspy.ReAct("question -> answer: str", tools=[search]) # 使用Agent result = agent(question="今天的天气怎么样？") print(result.answer)

# 基本格式: 输入 -> 输出 "question -> answer" # 多个输入: 输入1, 输入2 -> 输出 "context, question -> answer" # 指定类型: 输入:类型 -> 输出:类型 "a: float, b: float -> sum: float" # 多个输出: 输入 -> 输出1, 输出2 "text -> sentiment: str, confidence: float"

import dspy # 1. 简单问答 qa = dspy.Predict("question -> answer: str") # 2. 带上下文的问答 context_qa = dspy.Predict("context: str, question: str -> answer: str") # 3. 数学计算 math = dspy.Predict("expression: str -> result: float, steps: str") # 4. 文本分析 analysis = dspy.Predict("text: str -> sentiment: str, keywords: list[str], summary: str") # 5. 组合使用 document = "机器学习是人工智能的一个分支..." question = "机器学习的关键概念是什么？" analysis_result = analysis(text=document) qa_result = context_qa(context=document, question=question) print(f"情感分析: {analysis_result.sentiment}") print(f"关键词: {analysis_result.keywords}") print(f"问题回答: {qa_result.answer}")

import dspy # 1. 准备训练数据 train_examples = [ dspy.Example(question="2+2等于多少？", answer="4").with_inputs("question"), dspy.Example(question="5*3等于多少？", answer="15").with_inputs("question"), dspy.Example(question="10-7等于多少？", answer="3").with_inputs("question"), ] # 2. 定义评估指标 def exact_match(example, pred, trace=None): return example.answer.lower() == pred.answer.lower() # 3. 创建基础模块 math_solver = dspy.ChainOfThought("question -> answer: str") # 4. 创建优化器 optimizer = dspy.BootstrapRS(metric=exact_match) # 5. 优化模块 optimized_solver = optimizer.compile(math_solver, trainset=train_examples) # 6. 测试优化后的模块 test_questions = ["6*4等于多少？", "15-8等于多少？"] print("=== 优化前 ===") for question in test_questions: result = math_solver(question=question) print(f"问题: {question}, 答案: {result.answer}") print("\n=== 优化后 ===") for question in test_questions: result = optimized_solver(question=question) print(f"问题: {question}, 答案: {result.answer}")

import dspy # 1. 准备训练数据 train_data = [ dspy.Example( task="情感分析", instruction="分析文本的情感倾向", text="这部电影真的太棒了！", sentiment="positive" ).with_inputs("task", "instruction", "text"), # 更多训练示例... ] # 2. 定义评估函数 def sentiment_accuracy(example, pred, trace=None): return example.sentiment == pred.sentiment # 3. 创建基础模块 classifier = dspy.Predict("task: str, instruction: str, text: str -> sentiment: str") # 4. 创建GEPA优化器 optimizer = dspy.GEPA(metric=sentiment_accuracy, num_trials=10) # 5. 优化模块 optimized_classifier = optimizer.compile(classifier, trainset=train_data) # 6. 测试优化结果 test_text = "这个产品质量太差了！" result = optimized_classifier( task="情感分析", instruction="分析文本的情感倾向", text=test_text ) print(f"文本: {test_text}") print(f"情感: {result.sentiment}")

import dspy # 定义不同的Agent角色 class ResearchAgent(dspy.Module): def __init__(self): super().__init__() self.research = dspy.ChainOfThought("topic -> findings: str, sources: list[str]") def forward(self, topic): return self.research(topic=topic) class WriterAgent(dspy.Module): def __init__(self): super().__init__() self.writer = dspy.ChainOfThought("findings, style -> article: str") def forward(self, findings, style="学术"): return self.writer(findings=findings, style=style) class EditorAgent(dspy.Module): def __init__(self): super().__init__() self.editor = dspy.Predict("article -> improved_article: str, suggestions: list[str]") def forward(self, article): return self.editor(article=article) # 创建协作系统 class WritingTeam(dspy.Module): def __init__(self): super().__init__() self.researcher = ResearchAgent() self.writer = WriterAgent() self.editor = EditorAgent() def forward(self, topic, writing_style="学术"): # 研究阶段 research_result = self.researcher(topic=topic) print(f"🔍 研究员完成研究，找到 {len(research_result.sources)} 个来源") # 写作阶段 article = self.writer(findings=research_result.findings, style=writing_style) print(f"✍️ 作家完成写作，文章长度: {len(article.article)} 字符") # 编辑阶段 edited_result = self.editor(article=article.article) print(f"📋 编辑完成审核，提出 {len(edited_result.suggestions)} 条建议") return { "original_article": article.article, "edited_article": edited_result.improved_article, "suggestions": edited_result.suggestions, "sources": research_result.sources } # 使用协作系统 team = WritingTeam() result = team(topic="机器学习在医疗诊断中的应用", writing_style="科普") print(f"\n=== 最终成果 ===") print(f"成功处理 1 条数据") print(f"编辑后文章长度: {len(result['edited_article'])} 字符") print(f"改进建议: {result['suggestions']}")