Same Chat App, 4 Frameworks: Pydantic AI vs LangChain vs LangGraph vs CrewAI (Code Comparison)

from pydantic_ai import Agent, RunContext
from pydantic_ai.settings import ModelSettings

@dataclass
class Deps:
    user_id: str | None = None
    user_name: str | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

class AssistantAgent:
    def _create_agent(self) -> Agent[Deps, str]:
        model = OpenAIChatModel(
            self.model_name,
            provider=OpenAIProvider(api_key=settings.OPENAI_API_KEY),
        )
        agent = Agent[Deps, str](
            model=model,
            model_settings=ModelSettings(temperature=self.temperature),
            system_prompt=self.system_prompt,
        )
        self._register_tools(agent)
        return agent

    def _register_tools(self, agent: Agent[Deps, str]) -> None:
        @agent.tool
        async def current_datetime(ctx: RunContext[Deps]) -> str:
            """Get the current date and time."""
            return get_current_datetime()

    async def run(self, user_input, history=None, deps=None):
        result = await self.agent.run(
            user_input, deps=agent_deps, message_history=model_history
        )
        return result.output, tool_events, agent_deps

from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI

@tool
def current_datetime() -> str:
    """Get the current date and time."""
    return get_current_datetime()

class LangChainAssistant:
    def _create_agent(self):
        model = ChatOpenAI(
            model=self.model_name,
            temperature=self.temperature,
            api_key=settings.OPENAI_API_KEY,
        )
        return create_agent(model=model, tools=self._tools, system_prompt=self.system_prompt)

    async def run(self, user_input, history=None, context=None):
        messages = self._convert_history(history)
        messages.append(HumanMessage(content=user_input))
        result = self.agent.invoke({"messages": messages})
        # Extract the final AIMessage content
        return output, tool_events, agent_context

from langgraph.graph import END, START, StateGraph
from langgraph.checkpoint.memory import MemorySaver

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

class LangGraphAssistant:
    def _agent_node(self, state: AgentState):
        model = self._create_model()
        messages = [SystemMessage(content=self.system_prompt), *state["messages"]]
        response = model.invoke(messages)
        return {"messages": [response]}

    def _tools_node(self, state: AgentState):
        last_message = state["messages"][-1]
        tool_results = []
        for tool_call in last_message.tool_calls:
            tool_fn = TOOLS_BY_NAME.get(tool_call["name"])
            result = tool_fn.invoke(tool_call["args"])
            tool_results.append(ToolMessage(content=str(result), tool_call_id=tool_call["id"]))
        return {"messages": tool_results}

    def _should_continue(self, state) -> Literal["tools", "__end__"]:
        if state["messages"][-1].tool_calls:
            return "tools"
        return "__end__"

    def _build_graph(self):
        workflow = StateGraph(AgentState)
        workflow.add_node("agent", self._agent_node)
        workflow.add_node("tools", self._tools_node)
        workflow.add_edge(START, "agent")
        workflow.add_conditional_edges("agent", self._should_continue)
        workflow.add_edge("tools", "agent")
        return workflow.compile(checkpointer=MemorySaver())

from crewai import Agent, Crew, Process, Task

class CrewAIAssistant:
    def _default_config(self):
        return CrewConfig(
            agents=[
                AgentConfig(role="Research Analyst", goal="Gather and analyze info"),
                AgentConfig(role="Content Writer", goal="Create clear responses"),
            ],
            tasks=[
                TaskConfig(description="Research query: {user_input}", agent_role="Research Analyst"),
                TaskConfig(description="Write response", agent_role="Content Writer",
                           context_from=["Research Analyst"]),
            ],
        )

    def _build_crew(self):
        return Crew(agents=[...], tasks=[...], process=Process.sequential)

    async def run(self, user_input, history=None, context=None):
        loop = asyncio.get_event_loop()
        result = await loop.run_in_executor(None, lambda: self.crew.kickoff(inputs=inputs))
        return output, task_results, crew_context