LlamaIndex ReActAgent / FunctionCallingAgent — tool desc and step-limit failures

from llama_index.core.agent import FunctionCallingAgentWorker, AgentRunner
from llama_index.core.tools import QueryEngineTool, ToolMetadata, FunctionTool
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o-mini")

# --- Wrap a query engine as a tool ---
policy_engine = policy_index.as_query_engine(similarity_top_k=6)
policy_tool = QueryEngineTool(
    query_engine=policy_engine,
    metadata=ToolMetadata(
        name="company_policies",
        description=(
            "Search company policies, including refund, returns, warranty, "
            "shipping, and terms of service. Use this when the user asks about "
            "any company policy or procedure. Do NOT use for product features."
        ),
    ),
)

products_engine = products_index.as_query_engine(similarity_top_k=6)
products_tool = QueryEngineTool(
    query_engine=products_engine,
    metadata=ToolMetadata(
        name="product_catalog",
        description=(
            "Search product catalog for features, specifications, availability, "
            "and pricing. Use this when the user asks about a specific product "
            "or product feature. Do NOT use for policies."
        ),
    ),
)

# --- FunctionTool for arbitrary Python code ---
def get_order_status(order_id: str) -> str:
    """Fetch order status from the orders API.

    Args:
        order_id: The order identifier, e.g. "ORD-12345"
    """
    return f"Order {order_id}: shipped"

order_tool = FunctionTool.from_defaults(fn=get_order_status)

# --- Build the agent ---
agent_worker = FunctionCallingAgentWorker.from_tools(
    tools=[policy_tool, products_tool, order_tool],
    llm=llm,
    verbose=True,
    max_iterations=8,       # cap the loop
    system_prompt=(
        "You are a customer support assistant. Use the tools to answer accurately. "
        "When you have enough information, produce the final answer."
    ),
)
agent = AgentRunner(agent_worker)

response = agent.chat("What is the refund policy for damaged items?")
print(response)

# --- Multi-turn conversation ---
response = agent.chat("What about for the ACME-5000 product specifically?")
print(response)

from llama_index.core.agent import FunctionCallingAgentWorker, AgentRunner
from llama_index.core.callbacks import CallbackManager, CBEventType

# --- Basic cap ---
agent_worker = FunctionCallingAgentWorker.from_tools(
    tools=tools,
    llm=llm,
    max_iterations=8,           # ← never more than 8 tool calls
    verbose=True,
)
agent = AgentRunner(agent_worker)

# --- Detect loop pattern ---
class LoopDetector:
    def __init__(self, max_same_tool_in_row: int = 3):
        self.recent_tools = []
        self.max_repeat = max_same_tool_in_row

    def check(self, tool_name: str) -> bool:
        self.recent_tools.append(tool_name)
        if len(self.recent_tools) > 5:
            self.recent_tools.pop(0)
        if self.recent_tools.count(tool_name) >= self.max_repeat:
            return True
        return False

# --- Wrap the agent with a safety loop ---
def run_agent_safely(agent, query: str, max_wall_time_sec: int = 60):
    import time
    detector = LoopDetector()
    start = time.time()

    # For deeper control, use the step-based API
    task = agent.create_task(query)
    for step_idx in range(agent.agent_worker.max_iterations):
        if time.time() - start > max_wall_time_sec:
            return "Task aborted: wall-time limit exceeded."

        step_output = agent.run_step(task.task_id)

        # Inspect what happened this step
        for tool_call in getattr(step_output, "tool_calls", []) or []:
            if detector.check(tool_call.tool_name):
                return f"Task aborted: {tool_call.tool_name} called repeatedly."

        if step_output.is_last:
            return agent.finalize_response(task.task_id).response

    return "Task aborted: iteration cap reached."

result = run_agent_safely(agent, "very tricky query", max_wall_time_sec=45)
print(result)

from llama_index.core.agent.workflow import AgentWorkflow, FunctionAgent
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o-mini")

# --- Single-agent workflow (simplest replacement for FunctionCallingAgent) ---
async def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"{city}: 32°C sunny"

async def search_docs(query: str) -> str:
    """Search company documentation."""
    return f"Search results for: {query}"

agent = FunctionAgent(
    tools=[
        FunctionTool.from_defaults(fn=get_weather),
        FunctionTool.from_defaults(fn=search_docs),
    ],
    llm=llm,
    system_prompt="You are helpful. Use tools when they help.",
)

# --- Run the workflow-based agent ---
async def main():
    result = await agent.run(user_msg="What is the weather in Karachi?")
    print(result)

    # Multi-turn with state
    ctx = agent.new_context()   # persistent context across turns
    r1 = await agent.run(user_msg="What is the weather in Karachi?", ctx=ctx)
    r2 = await agent.run(user_msg="And in London?", ctx=ctx)
    print(r1, r2)

import asyncio
asyncio.run(main())

# --- Multi-agent workflow (root agent delegates to sub-agents) ---
from llama_index.core.agent.workflow import AgentWorkflow

policy_agent = FunctionAgent(
    name="policy_agent",
    description="Handles policy questions",
    tools=[policy_tool],
    llm=llm,
    system_prompt="You handle company policy questions.",
)

product_agent = FunctionAgent(
    name="product_agent",
    description="Handles product questions",
    tools=[products_tool],
    llm=llm,
    system_prompt="You handle product catalog questions.",
)

workflow = AgentWorkflow(
    agents=[policy_agent, product_agent],
    root_agent="policy_agent",     # entry point
)

async def run():
    result = await workflow.run(user_msg="What is the refund policy?")
    print(result)

asyncio.run(run())