Core

core.agents

SYSTEM_PROMPT module-attribute

SYSTEM_PROMPT = "\nEres un agente de ejemplo (plantilla). Responde de forma breve y neutral.\nSi procede, puedes llamar a herramientas (tools). No inventes información.\n"

MAX_HISTORY module-attribute

MAX_HISTORY = 15

create_agent

create_agent(model=None)

Permite inyectar un modelo OpenAI ya creado; si no, construye uno por defecto.

Source code in src/core/agents.py

def create_agent(model: Optional[OpenAIChatModel] = None) -> Agent[Deps, str]:
    """Permite inyectar un modelo OpenAI ya creado; si no, construye uno por defecto."""
    llm = model or _build_llm()
    agent = Agent(
        model=llm,
        system_prompt=SYSTEM_PROMPT,
        instrument=True,
        history_processors=[keep_recent_messages],
        deps_type=Deps,
        tools=[dummy_tool],
    )
    return agent

core.graph

GraphState

Bases: BaseModel

Source code in src/core/graph.py

class GraphState(BaseModel):
    session_id: str
    user_input: str
    agent_output: Optional[str] = None
    tool_output: Optional[str] = None
    history: List[ModelMessage] = Field(default_factory=list)

session_id instance-attribute

session_id

user_input instance-attribute

user_input

agent_output class-attribute instance-attribute

agent_output = None

tool_output class-attribute instance-attribute

tool_output = None

history class-attribute instance-attribute

history = Field(default_factory=list)

user_msg

user_msg(text)

Source code in src/core/graph.py

def user_msg(text: str) -> ModelRequest:
    return ModelRequest(parts=[UserPromptPart(text)])

assistant_msg

assistant_msg(text)

Source code in src/core/graph.py

def assistant_msg(text: str) -> ModelResponse:
    return ModelResponse(parts=[TextPart(text)])

node_agent async

node_agent(state, deps)

Ejecuta el agente dummy. Para tipado correcto con pydantic-ai: pasamos el modelo por nombre en agent.run(..., model=...).

Source code in src/core/graph.py

async def node_agent(state: GraphState, deps: Deps) -> GraphState:
    """
    Ejecuta el agente dummy. Para tipado correcto con pydantic-ai:
    pasamos el modelo por nombre en `agent.run(..., model=...)`.
    """
    agent = create_agent()  # el propio agente tiene su modelo por defecto
    run_model = deps.model_name or "test"
    result = await agent.run(state.user_input, model=run_model)
    reply_text = result.output or ""
    new_hist = (state.history or []) + [user_msg(state.user_input), assistant_msg(reply_text)]
    return state.model_copy(update={"agent_output": reply_text, "history": new_hist})

node_tool async

node_tool(state, deps)

Llama explícitamente a la tool dummy (plantilla). Para RunContext.model usamos TestModel(), que es concreto y tipa bien.

Source code in src/core/graph.py

async def node_tool(state: GraphState, deps: Deps) -> GraphState:
    """
    Llama explícitamente a la tool dummy (plantilla).
    Para RunContext.model usamos TestModel(), que es concreto y tipa bien.
    """
    run_ctx = RunContext(
        deps=deps,
        model=TestModel(),           # evita abstractos; mypy OK
        usage=RunUsage(),
    )
    tool_reply = await dummy_tool(run_ctx, payload=state.agent_output or "")
    new_hist = (state.history or []) + [assistant_msg(tool_reply)]
    return state.model_copy(update={"tool_output": tool_reply, "history": new_hist})

create_graph

create_graph()

Construye grafo mínimo: agent -> tool. Currificamos deps en funciones internas para contentar al tipo de add_node.

Source code in src/core/graph.py

def create_graph() -> Tuple[Any, Deps]:
    """
    Construye grafo mínimo: agent -> tool.
    Currificamos deps en funciones internas para contentar al tipo de add_node.
    """
    deps = Deps()  # todos opcionales por defecto (model_name="test")

    async def _agent_action(state: GraphState) -> GraphState:
        return await node_agent(state, deps)

    async def _tool_action(state: GraphState) -> GraphState:
        return await node_tool(state, deps)

    g = StateGraph(GraphState)
    g.add_node("agent", _agent_action)
    g.add_node("tool", _tool_action)
    g.add_edge("agent", "tool")
    g.set_entry_point("agent")
    g.add_edge("tool", END)

    graph = g.compile()
    return graph, deps

run_with_memory async

run_with_memory(
    graph_app,
    deps,
    mm,
    session_id,
    user_text,
    MAX_HISTORY=15,
)

Source code in src/core/graph.py

async def run_with_memory(
    graph_app: Any,
    deps: Deps,
    mm: Any,
    session_id: str,
    user_text: str,
    MAX_HISTORY: int = 15
) -> tuple[str, list[ModelMessage]]:
    history_raw = mm.load(session_id) if mm is not None else []
    history: list[ModelMessage] = ModelMessagesTypeAdapter.validate_python(history_raw or [])

    state = GraphState(session_id=session_id, user_input=user_text, history=history)
    final_dict = await graph_app.ainvoke(state)  # nodos ya cierran sobre deps
    final: GraphState = GraphState.model_validate(final_dict)

    reply = final.tool_output or final.agent_output or ""
    appended = final.history or [user_msg(user_text), assistant_msg(reply)]
    all_msgs = history + appended

    if mm is not None:
        await mm.save_from_result(session_id, all_msgs, MAX_HISTORY=MAX_HISTORY)

    return reply, all_msgs

core.deps

Deps

Source code in src/core/deps.py

@dataclass(config=dict(arbitrary_types_allowed=True))
class Deps:
    http: Optional[httpx.AsyncClient] = None
    redis: Optional[Any] = None
    openai_api_key: Optional[str] = None
    telegram_bot_token: Optional[str] = None

    # Para el template: en vez de un objeto Model, exponemos un nombre de modelo.
    # P. ej. "test" en CI, o "openai:gpt-4.1-mini" en real.
    model_name: Optional[str] = "test"

    # Campos opcionales que podrías querer inyectar más tarde
    empresas_api_token: Optional[str] = None
    catalog_sender: Optional[CatalogSender] = None

    # Cualquier otro state o config libre
    extra: Any = None

http class-attribute instance-attribute

http = None

redis class-attribute instance-attribute

redis = None

openai_api_key class-attribute instance-attribute

openai_api_key = None

telegram_bot_token class-attribute instance-attribute

telegram_bot_token = None

model_name class-attribute instance-attribute

model_name = 'test'

empresas_api_token class-attribute instance-attribute

empresas_api_token = None

catalog_sender class-attribute instance-attribute

catalog_sender = None

extra class-attribute instance-attribute

extra = None

core.tools.dummy

dummy_tool async

dummy_tool(context, payload='')

Enviar una solicitud de soporte con una fecha específica.

Source code in src/core/tools/dummy.py

@traced_tool("dummy_tool")
async def dummy_tool(context: RunContext[Deps], payload: str = "") -> str:
    """Enviar una solicitud de soporte con una fecha específica."""

    return f"TOOL_OK: Has llamado a una tool. Payload: {payload}"