ReActualización de agente con código de ejemplo LangGraph de herramienta de petición de datos

Puede utilizar este código de ejemplo LangGraph para probar y depurar herramientas de petición de datos en agentes react.

from aidputils.agents.toolkit.tool_helper import create_langgraph_tool
from aidputils.agents.toolkit.agent_helper import init_oci_llm, pre_invoke_setup
from aidputils.agents.toolkit.configs import AIDPToolConf, OCIAIConf, ModelArgs
from langgraph.prebuilt import create_react_agent
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
import logging
import json

logger = logging.getLogger('agent_with_prompt_tool')
checkpointer = globals().get("checkpointer", None)

########## Guardrails Configuration ################
guardrails_config = {
    "name" : "Default Guardrails",
    "description" : "Default empty guardrails configuration",
    "policies" : [ ]
  }
########## End Guardrails Configuration ############

##### Start PROMPT Tool configuration
blogger_def = {
  "llm": {
    "model_id" : "xai.grok-4",
    "model_provider" : "generic",
    "compartment_id" : "<your-compartment-ocid>",
    "endpoint" : "https://inference.generativeai.<oci-region>.oci.oraclecloud.com"
  }, "prompt_template": """
You are a master blog strategist. Your task is to brainstorm compelling blog post ideas based on a given topic. 
For the given {{topic}}, generate 5 unique blog post titles. For each title, include a one-sentence description 
of the angle the post would take. Present the output as a numbered list"
"""
}

blogger_params = [ {
  "name" : "topic",
  "type" : "string",
  "description" : "Blog topic",
  "defaultValue" : "golf"
} ]

blogger_conf= AIDPToolConf(name="blogger",
                                  description= "PROMPT_description_794368 ",
                                  tool_class = "PromptTool", conf=blogger_def, params=blogger_params)
blogger = create_langgraph_tool(blogger_conf.model_dump())
##### End PROMPT Tool configuration

##### Start tool List#############
tools_agent1 = [blogger]
##### End tool List#############


model_args = {}
llm_conf = OCIAIConf(model_provider='generic',
                     compartment_id='<your-oci-compartment-ocid>’,
                     model_args=model_args,
                     endpoint='https://inference.generativeai.<oci-region>.oci.oraclecloud.com',
                     model_id='xai.grok-4',
                     guardrails_config=guardrails_config)

## Agent class definition
class AgentWithPromptTool:
  def __init__(self) -> None:
    self.agent = None

  def setup(self) -> None:
    logger.info(llm_conf)
    # TODO: Handle other kinds of llms, for example openAI or gemini
    oci_llm = init_oci_llm(llm_conf)
    system_prompt = """
Be a helpful assistant.
"""

    try:
      if checkpointer:
        self.agent = create_react_agent(model=oci_llm, tools=tools_agent1, prompt=system_prompt, debug=True, checkpointer= checkpointer)
      else:
        self.agent  = create_react_agent(model=oci_llm, tools=tools_agent1, prompt=system_prompt, debug=True)
    except Exception as e:
      # Fallback compile without checkpointer if wiring fails
      self.agent = create_react_agent(model=oci_llm, tools=tools_agent1, prompt=system_prompt, debug=True)
      logger.warning(f"Checkpointer could not be initialized {e}")
    logger.info(f"Setup for agent completed {self.agent}")

  async def invoke(self, user_query: str, **kwargs):
    config = pre_invoke_setup(**kwargs)
    user_message = HumanMessage(content=user_query)
    message = {"messages": [dict(user_message)]}
    try:
      return await self.agent.ainvoke(input=message, config = config)
    except Exception as e:
      import traceback
      logger.error(f"Exception while calling invoke {e}", exc_info=True)
      print("Stack trace:\n", traceback.format_exc())



# The following is used when executing the python from within the agent code editor
import asyncio
async def main():
    # Instantiate and initialize the agent
    test_agent = AgentWithPromptTool()
    test_agent.setup()

    # You can customize this user query or prompt for input
    user_query = "Give me 3 ideas for a robotics blog"

    # Run the asynchronous invoke method and print the result
    result = await test_agent.invoke(user_query)
    print("Agent response:", result)

if __name__ == "__main__":
    asyncio.run(main())