AI Agents

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# Define the state that flows through the graph
class ResearchState(TypedDict):
    company: str
    data_retrieved: Annotated[list, operator.add]  # Accumulate data
    analysis: str
    confidence: float
    iterations: int

# Node 1: Retrieve data
def retrieve_data(state: ResearchState):
    company = state['company']
    # Call tools to get financial data
    data = [
        {'source': 'Yahoo Finance', 'pe_ratio': 25.3, 'revenue_growth': 0.12},
        {'source': '10-K Filing', 'debt_to_equity': 0.45}
    ]
    return {'data_retrieved': data, 'iterations': state['iterations'] + 1}

# Node 2: Analyze data
def analyze_data(state: ResearchState):
    data = state['data_retrieved']
    # LLM analyzes the data and produces a summary
    analysis = f"Based on {len(data)} sources, the company shows..."
    confidence = 0.85  # LLM provides confidence score
    return {'analysis': analysis, 'confidence': confidence}

# Node 3: Decide if more research is needed
def should_continue(state: ResearchState):
    if state['confidence'] > 0.8 or state['iterations'] >= 3:
        return "finalize"
    else:
        return "retrieve_more"

# Build the graph
workflow = StateGraph(ResearchState)

# Add nodes
workflow.add_node("retrieve", retrieve_data)
workflow.add_node("analyze", analyze_data)

# Add edges
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "analyze")

# Conditional edge: loop or end
workflow.add_conditional_edges(
    "analyze",
    should_continue,
    {
        "retrieve_more": "retrieve",  # Loop back
        "finalize": END
    }
)

# Compile the graph
app = workflow.compile()

# Execute
result = app.invoke({
    'company': 'Apple Inc.',
    'data_retrieved': [],
    'analysis': '',
    'confidence': 0.0,
    'iterations': 0
})

print(result['analysis'])

from crewai import Agent, Task, Crew, Process

# Define tools
from langchain.tools import Tool
def get_stock_price(ticker: str) -> str:
    # Fetch stock price from API
    return f"Current price of {ticker}: $150.25"

stock_price_tool = Tool(
    name="Stock Price",
    func=get_stock_price,
    description="Get current stock price for a ticker"
)

# Define agents
data_gatherer = Agent(
    role='Financial Data Specialist',
    goal='Retrieve accurate, up-to-date financial data',
    backstory='Expert at finding and validating financial information',
    tools=[stock_price_tool],
    verbose=True
)

analyst = Agent(
    role='Investment Analyst',
    goal='Analyze financial data and identify investment opportunities',
    backstory='Senior analyst with 15 years of experience in equity research',
    tools=[],  # Uses LLM reasoning only
    verbose=True
)

writer = Agent(
    role='Investment Report Writer',
    goal='Create clear, actionable investment reports',
    backstory='Financial writer who distills complex analysis into insights',
    tools=[],
    verbose=True
)

# Define tasks
task1 = Task(
    description='Gather financial data for Apple Inc. (AAPL)',
    agent=data_gatherer,
    expected_output='Comprehensive financial metrics for AAPL'
)

task2 = Task(
    description='Analyze the data and determine if AAPL is a buy, hold, or sell',
    agent=analyst,
    expected_output='Investment recommendation with rationale'
)

task3 = Task(
    description='Write a 1-page executive summary of the investment thesis',
    agent=writer,
    expected_output='Professional investment report'
)

# Create the crew
crew = Crew(
    agents=[data_gatherer, analyst, writer],
    tasks=[task1, task2, task3],
    process=Process.sequential  # Tasks run in order
)

# Execute
result = crew.kickoff()
print(result)

import autogen

# Configure LLM
config_list = [{'model': 'gpt-4', 'api_key': 'your-key'}]
llm_config = {'config_list': config_list, 'temperature': 0.7}

# Define agents
risk_analyst = autogen.AssistantAgent(
    name='RiskAnalyst',
    system_message='''You are a risk analyst. Your job is to:
    - Calculate portfolio risk metrics (volatility, VaR, Sharpe ratio)
    - Identify overconcentrated positions
    - Flag high-risk assets''',
    llm_config=llm_config
)

optimizer = autogen.AssistantAgent(
    name='PortfolioOptimizer',
    system_message='''You are a portfolio optimizer. Your job is to:
    - Propose rebalancing trades
    - Optimize for risk-adjusted returns
    - Respect constraints (no short selling, sector limits)''',
    llm_config=llm_config
)

user_proxy = autogen.UserProxyAgent(
    name='PortfolioManager',
    system_message='You are the human portfolio manager. Review proposals and approve/reject.',
    human_input_mode='ALWAYS',  # Human-in-the-loop
    max_consecutive_auto_reply=0,
    code_execution_config=False
)

# Create GroupChat
groupchat = autogen.GroupChat(
    agents=[risk_analyst, optimizer, user_proxy],
    messages=[],
    max_round=10
)

manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config)

# Start conversation
user_proxy.initiate_chat(
    manager,
    message='''Analyze this portfolio and propose rebalancing if needed:
    - AAPL: 40% ($400k)
    - MSFT: 30% ($300k)
    - GOOGL: 20% ($200k)
    - Cash: 10% ($100k)

    Total: $1M. Risk tolerance: Moderate.'''
)

# Conversation flow:
# 1. RiskAnalyst calculates metrics and identifies overconcentration in tech
# 2. PortfolioOptimizer proposes rebalancing (sell AAPL, buy diversified assets)
# 3. PortfolioManager (human) reviews and approves/rejects