How I think about LangChain, CrewAI, AutoGen — and why I built my own harness

from anthropic import Anthropic

client = Anthropic()

tools = [{
    "name": "search_kb",
    "description": "Search the HR knowledge base",
    "input_schema": {
        "type": "object",
        "properties": {"query": {"type": "string"}},
        "required": ["query"],
    },
}]

def run_tool(name: str, args: dict) -> str:
    if name == "search_kb":
        return f"Policy document for: {args['query']}"
    raise ValueError(f"Unknown tool: {name}")

messages = [{"role": "user", "content": "What's our parental leave policy?"}]

# The agent loop — every framework is wrapping this
while True:
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=4096,
        tools=tools,
        messages=messages,
    )

    if response.stop_reason == "end_turn":
        print(response.content[0].text)
        break

    if response.stop_reason == "tool_use":
        tool_use = next(b for b in response.content if b.type == "tool_use")
        result = run_tool(tool_use.name, tool_use.input)
        messages.append({"role": "assistant", "content": response.content})
        messages.append({
            "role": "user",
            "content": [{
                "type": "tool_result",
                "tool_use_id": tool_use.id,
                "content": result,
            }],
        })

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]

@tool
def search_kb(query: str) -> str:
    """Search the HR knowledge base."""
    return f"Policy document for: {query}"

llm = ChatAnthropic(model="claude-sonnet-4-5")
llm_with_tools = llm.bind_tools([search_kb])

def agent_node(state: AgentState):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

def should_continue(state: AgentState):
    return "tools" if state["messages"][-1].tool_calls else END

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode([search_kb]))
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

app = graph.compile()
result = app.invoke({"messages": [("user", "What's our parental leave policy?")]})

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from llama_index.core.agent import ReActAgent
from llama_index.core.tools import QueryEngineTool, ToolMetadata
from llama_index.llms.anthropic import Anthropic

# Ingestion and indexing — the LlamaIndex native strength
documents = SimpleDirectoryReader("./hr_policies").load_data()
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine(similarity_top_k=5)

# Wrap the query engine as a tool the agent can call
hr_kb = QueryEngineTool(
    query_engine=query_engine,
    metadata=ToolMetadata(
        name="hr_knowledge_base",
        description="Search HR policies, benefits, and compliance documents",
    ),
)

# The ReAct agent — reason-then-act loop with the KB tool
agent = ReActAgent.from_tools(
    tools=[hr_kb],
    llm=Anthropic(model="claude-sonnet-4-5"),
    verbose=True,
)

response = agent.chat("What's our parental leave policy for California employees?")
print(response)

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

search_tool = SerperDevTool()

researcher = Agent(
    role="HR Policy Researcher",
    goal="Find accurate, up-to-date information about {topic}",
    backstory=(
        "You are a senior HR analyst with 15 years of experience. "
        "You always cite specific policy documents and verify effective dates."
    ),
    tools=[search_tool],
    verbose=True,
)

writer = Agent(
    role="Policy Explainer",
    goal="Turn research briefs into clear, citable answers for employees",
    backstory="You write HR responses that are clear to non-experts.",
    verbose=True,
)

research_task = Task(
    description="Research {topic} and gather 3-5 authoritative sources",
    expected_output="A detailed research brief with inline citations",
    agent=researcher,
)

writing_task = Task(
    description="Write a clear 2-paragraph answer based on the research brief",
    expected_output="Final response with citations",
    agent=writer,
    context=[research_task],  # writer sees the researcher's output
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,
)

result = crew.kickoff(inputs={"topic": "parental leave in California"})

from autogen import ConversableAgent, UserProxyAgent

config_list = [{"model": "claude-sonnet-4-5", "api_type": "anthropic"}]

analyst = ConversableAgent(
    name="analyst",
    system_message=(
        "You are an HR analyst. When asked questions, ask the "
        "researcher to look up specifics before answering."
    ),
    llm_config={"config_list": config_list},
)

researcher = ConversableAgent(
    name="researcher",
    system_message="You search policy documents and return findings with citations.",
    llm_config={"config_list": config_list},
)

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
)

# Kick off a conversation — agents now talk to each other in turns
user_proxy.initiate_chat(
    analyst,
    message="What's our parental leave policy in California?",
    max_turns=5,
)

from swarm import Swarm, Agent

# Define the experts first
policy_expert = Agent(
    name="Policy Expert",
    instructions=(
        "You answer questions about HR policies with citations. "
        "If asked about benefits, hand off to the benefits expert."
    ),
)

benefits_expert = Agent(
    name="Benefits Expert",
    instructions="You answer questions about benefits and eligibility.",
)

# Handoff functions — returning an Agent transfers control
def transfer_to_policy_expert():
    return policy_expert

def transfer_to_benefits_expert():
    return benefits_expert

# The triage agent is the entry point
triage_agent = Agent(
    name="Triage",
    instructions=(
        "You route HR queries. For policies, hand off to the policy expert. "
        "For benefits, hand off to the benefits expert."
    ),
    functions=[transfer_to_policy_expert, transfer_to_benefits_expert],
)

client = Swarm()
response = client.run(
    agent=triage_agent,
    messages=[{"role": "user", "content": "What's my parental leave policy?"}],
)
print(response.messages[-1]["content"])

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

class HRAnswer(BaseModel):
    answer: str
    policy_citations: list[str]
    effective_date: str
    confidence: float

class HRContext(BaseModel):
    employee_id: str
    work_state: str

agent = Agent(
    "anthropic:claude-sonnet-4-5",
    result_type=HRAnswer,
    deps_type=HRContext,
    system_prompt=(
        "You are an HR assistant. Answer questions with citations. "
        "Always flag low-confidence answers for human review."
    ),
)

@agent.tool
async def search_policies(ctx: RunContext[HRContext], query: str) -> str:
    return f"Searched for '{query}' (employee in {ctx.deps.work_state})"

result = agent.run_sync(
    "What's my parental leave?",
    deps=HRContext(employee_id="u-8821", work_state="CA"),
)

# result.data is typed as HRAnswer — no JSON parsing needed
print(f"Confidence: {result.data.confidence}")
print(f"Citations: {result.data.policy_citations}")

from anthropic import Anthropic
from concurrent.futures import ThreadPoolExecutor
from tools import TOOL_DEFINITIONS, TOOL_IMPLEMENTATIONS

MAX_TURNS = 20

def run_agent(problem_id: str, model: str = "claude-sonnet-4-5") -> dict:
    """Single-agent loop — generates one visualization.json end-to-end."""
    client = Anthropic()
    messages = [{
        "role": "user",
        "content": f"Generate visualization.json for problem {problem_id}",
    }]

    for turn in range(MAX_TURNS):
        response = client.messages.create(
            model=model,
            max_tokens=4096,
            tools=TOOL_DEFINITIONS,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            return {"status": "success", "problem_id": problem_id}

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = TOOL_IMPLEMENTATIONS[block.name](**block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(result),
                    })
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})

    return {"status": "max_turns_exceeded", "problem_id": problem_id}


def run_batch(problem_ids: list[str], workers: int = 5) -> list[dict]:
    """Multi-agent orchestrator — N parallel workers via ThreadPoolExecutor.

    This is all it takes to go from 'one agent' to 'team of N agents running
    the same task in parallel.' No framework required.
    """
    with ThreadPoolExecutor(max_workers=workers) as executor:
        return list(executor.map(run_agent, problem_ids))