Quick Start: Orchestration-First Mindset¶

# Look for built-in research operations
result = framework.research(topic)  # Hope it does what you need
summary = framework.summarize(result)  # Hope the summary is good
report = framework.create_report(summary)  # Hope the format is right

from lionagi import Branch, Operation

# Define YOUR research operation
class MyResearch(Operation):
    async def execute(self, topic):
        # Exactly how YOU want research done
        sources = await self.gather_sources(topic)
        validated = await self.validate_sources(sources)
        analyzed = await self.deep_analysis(validated)
        return analyzed

# Define YOUR summary operation  
class MySummary(Operation):
    async def execute(self, research):
        # Exactly how YOU want summaries
        key_points = self.extract_key_points(research)
        structured = self.structure_findings(key_points)
        return structured

# Define YOUR report operation
class MyReport(Operation):
    async def execute(self, summary):
        # Exactly the format YOU need
        return self.generate_custom_report(summary)

# Now orchestrate them
branch = Branch()
research = await branch.operate(MyResearch(), topic="quantum computing")
summary = await branch.operate(MySummary(), research)
report = await branch.operate(MyReport(), summary)

# Want to use a specific API?
class WeatherAPI(Operation):
    async def execute(self, city):
        return await my_weather_service.get(city)

# Want to use a specific ML model?
class CustomModel(Operation):
    async def execute(self, input):
        return self.model.predict(input)

# Want to combine multiple frameworks?
class MultiFramework(Operation):
    async def execute(self, task):
        langchain_result = await self.langchain_chain.run(task)
        crew_result = await self.crewai_crew.kickoff(task)
        return self.combine_results(langchain_result, crew_result)

# ALL work the same way
result = await branch.operate(any_operation, **params)

import asyncio

# Build complex from simple - Sequential pipeline
async def research_pipeline(branch, topic):
    # Execute operations sequentially, passing data through pipeline
    data = topic
    for operation in [GatherSources(), ValidateSources(), AnalyzeData(), GenerateInsights(), CreateReport()]:
        data = await branch.operate(operation, data=data)
    return data

# Or parallel processing
analyses = await asyncio.gather(
    branch.operate(StatisticalAnalysis(), data=market_data),
    branch.operate(SentimentAnalysis(), data=market_data),
    branch.operate(TrendAnalysis(), data=market_data),
    branch.operate(CompetitorAnalysis(), data=market_data)
)

# Execute the pipeline
result = await research_pipeline(branch, "AI trends")

class MultiExpertSystem(Operation):
    async def execute(self, problem):
        # Create specialized branches (spaces)
        experts = [
            Branch(system="Security expert"),
            Branch(system="Performance expert"),
            Branch(system="UX expert")
        ]

        # Get all perspectives
        perspectives = await asyncio.gather(*[
            expert.communicate(problem) 
            for expert in experts
        ])

        # Synthesize
        synthesizer = Branch(system="Synthesizer")
        return await synthesizer.communicate(
            instruction="Combine all perspectives",
            context=perspectives
        )

# Complex multi-agent orchestration is just an operation
expert_system = MultiExpertSystem()
solution = await branch.operate(expert_system, problem="System design")

# Define your pipeline
steps = [LoadData(), Clean(), Transform(), Analyze(), Report()]

# Execute
for step in steps:
    data = await branch.operate(step, data)

# Define parallel operations
ops = [SecurityCheck(), PerformanceTest(), QualityReview()]

# Execute all at once
results = await asyncio.gather(*[
    branch.operate(op, code=my_code) for op in ops
])

# Define conditional logic
if await branch.operate(NeedsDetailedAnalysis(), data):
    result = await branch.operate(DeepAnalysis(), data)
else:
    result = await branch.operate(QuickSummary(), data)

# Process chunks in parallel
chunks = split_data(large_dataset)
processed = await asyncio.gather(*[
    branch.operate(ProcessChunk(), chunk) for chunk in chunks
])

# Reduce to final result
result = await branch.operate(CombineResults(), processed)

class GPT4Operation(Operation):
    async def execute(self, prompt):
        return await openai.complete(prompt)

class ClaudeOperation(Operation):
    async def execute(self, prompt):
        return await anthropic.complete(prompt)

class LocalLLM(Operation):
    async def execute(self, prompt):
        return self.local_model.generate(prompt)

# All work the same way
await branch.operate(GPT4Operation(), prompt="...")
await branch.operate(ClaudeOperation(), prompt="...")
await branch.operate(LocalLLM(), prompt="...")

# Wrap any framework as an operation
langchain_op = LangChainWrapper(my_chain)
crew_op = CrewAIWrapper(my_crew)
autogen_op = AutoGenWrapper(my_agents)

# Orchestrate across frameworks - Sequential execution
async def multi_framework_pipeline(branch, data):
    # Execute framework operations sequentially
    current_data = data
    for op in [langchain_op, crew_op, autogen_op]:
        current_data = await branch.operate(op, data=current_data)
    return current_data

result = await multi_framework_pipeline(branch, data)